[symbolize] Split symbolize task to trusted/untrusted parts (#3538)

src/clusterfuzz/_internal/bot/tasks/utasks/symbolize_task.py

@@ -29,6 +29,7 @@
 from clusterfuzz._internal.datastore import data_handler
 from clusterfuzz._internal.datastore import data_types
 from clusterfuzz._internal.metrics import logs
+from clusterfuzz._internal.protos import uworker_msg_pb2
 from clusterfuzz._internal.system import environment
 from clusterfuzz._internal.system import process_handler
 
@@ -38,16 +39,18 @@
 STACK_FRAME_COUNT = 128
 
 
-def utask_preprocess(testcase_id, job_type, uworker_env):
-  return uworker_io.UworkerInput(
-      job_type=job_type, testcase_id=testcase_id, uworker_env=uworker_env)
+def handle_build_setup_error(output):
+  testcase_id = output.uworker_input.testcase_id
+  job_type = output.uworker_input.job_type
+  testcase = data_handler.get_testcase_by_id(testcase_id)
+  build_fail_wait = environment.get_value('FAIL_WAIT')
+  data_handler.update_testcase_comment(testcase, data_types.TaskState.ERROR,
+                                       'Build setup failed')
+  tasks.add_task('symbolize', testcase_id, job_type, wait_time=build_fail_wait)
 
 
-def utask_main(uworker_input):
-  """Execute a symbolize command."""
-  job_type = uworker_input.job_type
-  testcase_id = uworker_input.testcase_id
-
+def utask_preprocess(testcase_id, job_type, uworker_env):
+  """Runs preprocessing for symbolize task."""
   # Locate the testcase associated with the id.
   testcase = data_handler.get_testcase_by_id(testcase_id)
 
@@ -59,18 +62,28 @@ def utask_main(uworker_input):
 
   # Setup testcase and its dependencies.
   setup_input = setup.preprocess_setup_testcase(testcase)
+  return uworker_io.UworkerInput(
+      job_type=job_type,
+      testcase_id=testcase_id,
+      uworker_env=uworker_env,
+      setup_input=setup_input,
+      testcase=testcase)
+
+
+def utask_main(uworker_input):
+  """Execute the untrusted part of a symbolize command."""
+  job_type = uworker_input.job_type
+  testcase = uworker_input.testcase
+  setup_input = uworker_input.setup_input
+
   _, testcase_file_path, error = setup.setup_testcase(testcase, job_type,
                                                       setup_input)
   if error:
-    # Because this is trusted, we can trust the error.
-    all_errors = uworker_handle_errors.get_all_handled_errors()
-    error.uworker_input.module_name = __name__
-    uworker_handle_errors.handle(error, all_errors)
-    return None
+    return error
 
   # Initialize variables.
-  build_fail_wait = environment.get_value('FAIL_WAIT')
-
+  #TODO(alhijazi): check with @metzman if the blobstore is read-only and
+  # accessible within utasks.
   old_crash_stacktrace = data_handler.get_stacktrace(testcase)
   sym_crash_type = testcase.crash_type
   sym_crash_address = testcase.crash_address
@@ -94,12 +107,9 @@ def utask_main(uworker_input):
   crash_revision = environment.get_value('APP_REVISION')
 
   if not build_manager.check_app_path():
-    testcase = data_handler.get_testcase_by_id(testcase_id)
-    data_handler.update_testcase_comment(testcase, data_types.TaskState.ERROR,
-                                         'Build setup failed')
-    tasks.add_task(
-        'symbolize', testcase_id, job_type, wait_time=build_fail_wait)
-    return None
+    return uworker_io.UworkerOutput(
+        error_message='Build setup failed',
+        error_type=uworker_msg_pb2.ErrorType.SYMBOLIZE_BUILD_SETUP_ERROR)
 
   # ASAN tool settings (if the tool is used).
   # See if we can get better stacks with higher redzone sizes.
@@ -147,12 +157,9 @@ def utask_main(uworker_input):
   if (not symbolized_builds or
       (not build_manager.check_app_path() and
        not build_manager.check_app_path('APP_PATH_DEBUG'))):
-    testcase = data_handler.get_testcase_by_id(testcase_id)
-    data_handler.update_testcase_comment(testcase, data_types.TaskState.ERROR,
-                                         'Build setup failed')
-    tasks.add_task(
-        'symbolize', testcase_id, job_type, wait_time=build_fail_wait)
-    return None
+    return uworker_io.UworkerOutput(
+        error_message='Build setup failed',
+        error_type=uworker_msg_pb2.ErrorType.SYMBOLIZE_BUILD_SETUP_ERROR)
 
   # Increase malloc_context_size to get all stack frames. Default is 30.
   environment.reset_current_memory_tool_options(
@@ -170,46 +177,26 @@ def utask_main(uworker_input):
       get_symbolized_stacktraces(testcase_file_path, testcase,
                                  old_crash_stacktrace, sym_crash_state))
 
-  # Update crash parameters.
-  testcase = data_handler.get_testcase_by_id(testcase_id)
-  testcase.crash_type = sym_crash_type
-  testcase.crash_address = sym_crash_address
-  testcase.crash_state = sym_crash_state
-  testcase.crash_stacktrace = (
-      data_handler.filter_stacktrace(sym_crash_stacktrace))
+  symbolize_task_outptut = uworker_io.SymbolizeTaskOutput(
+      crash_type=sym_crash_type,
+      crash_address=sym_crash_address,
+      crash_state=sym_crash_state,
+      crash_stacktrace=(data_handler.filter_stacktrace(sym_crash_stacktrace)),
+      symbolized=result,
+      crash_revision=int(crash_revision))
 
-  if not result:
-    data_handler.update_testcase_comment(
-        testcase, data_types.TaskState.ERROR,
-        'Unable to reproduce crash, skipping '
-        'stacktrace update')
-  else:
-    # Switch build url to use the less-optimized symbolized build with better
-    # stacktrace.
+  if result:
     build_url = environment.get_value('BUILD_URL')
     if build_url:
-      testcase.set_metadata('build_url', build_url, update_testcase=False)
-
-    data_handler.update_testcase_comment(testcase,
-                                         data_types.TaskState.FINISHED)
-
-  testcase.symbolized = True
-  testcase.crash_revision = crash_revision
-  testcase.put()
-
-  # We might have updated the crash state. See if we need to marked as duplicate
-  # based on other testcases.
-  data_handler.handle_duplicate_entry(testcase)
-
-  task_creation.create_blame_task_if_needed(testcase)
+      symbolize_task_outptut.build_url = build_url
 
   # Switch current directory before builds cleanup.
   root_directory = environment.get_value('ROOT_DIR')
   os.chdir(root_directory)
 
   # Cleanup symbolized builds which are space-heavy.
   symbolized_builds.delete()
-  return None
+  return uworker_io.UworkerOutput(symbolize_task_outptut=symbolize_task_outptut)
 
 
 def get_symbolized_stacktraces(testcase_file_path, testcase,
@@ -298,5 +285,48 @@ def get_symbolized_stacktraces(testcase_file_path, testcase,
   return symbolized, stacktrace
 
 
+HANDLED_ERRORS = [
+    uworker_msg_pb2.ErrorType.SYMBOLIZE_BUILD_SETUP_ERROR,  # pylint: disable=no-member
+    uworker_msg_pb2.ErrorType.UNHANDLED  # pylint: disable=no-member
+] + setup.HANDLED_ERRORS
+
+
 def utask_postprocess(output):
-  del output
+  """Handle the output from utask_main."""
+  if output.error_type is not None:
+    uworker_handle_errors.handle(output, HANDLED_ERRORS)
+    return
+
+  symbolize_task_output = output.symbolize_task_output
+
+  # Update crash parameters.
+  testcase = data_handler.get_testcase_by_id(output.uworker_input.testcase_id)
+  testcase.crash_type = symbolize_task_output.crash_type
+  testcase.crash_address = symbolize_task_output.crash_address
+  testcase.crash_state = symbolize_task_output.crash_state
+  testcase.crash_stacktrace = symbolize_task_output.crash_stacktrace
+
+  if not symbolize_task_output.symbolized:
+    data_handler.update_testcase_comment(
+        testcase, data_types.TaskState.ERROR,
+        'Unable to reproduce crash, skipping '
+        'stacktrace update')
+  else:
+    # Switch build url to use the less-optimized symbolized build with better
+    # stacktrace.
+    if symbolize_task_output.build_url:
+      testcase.set_metadata(
+          'build_url', symbolize_task_output.build_url, update_testcase=False)
+
+    data_handler.update_testcase_comment(testcase,
+                                         data_types.TaskState.FINISHED)
+
+  testcase.symbolized = True
+  testcase.crash_revision = symbolize_task_output.crash_revision
+  testcase.put()
+
+  # We might have updated the crash state. See if we need to marked as duplicate
+  # based on other testcases.
+  data_handler.handle_duplicate_entry(testcase)
+
+  task_creation.create_blame_task_if_needed(testcase)

src/clusterfuzz/_internal/bot/tasks/utasks/uworker_handle_errors.py

@@ -18,6 +18,7 @@
 from clusterfuzz._internal.bot.tasks.utasks import minimize_task
 from clusterfuzz._internal.bot.tasks.utasks import progression_task
 from clusterfuzz._internal.bot.tasks.utasks import regression_task
+from clusterfuzz._internal.bot.tasks.utasks import symbolize_task
 from clusterfuzz._internal.bot.tasks.utasks import variant_task
 from clusterfuzz._internal.protos import uworker_msg_pb2
 
@@ -83,6 +84,8 @@ def get_handle_all_errors_mapping():
           regression_task.handle_revision_list_error,
       uworker_msg_pb2.ErrorType.REGRESSION_BUILD_NOT_FOUND:
           regression_task.handle_build_not_found_error,
+      uworker_msg_pb2.ErrorType.SYMBOLIZE_BUILD_SETUP_ERROR:
+          symbolize_task.handle_build_setup_error,
       uworker_msg_pb2.ErrorType.UNHANDLED:
           noop,
   }

src/clusterfuzz/_internal/bot/tasks/utasks/uworker_io.py

@@ -538,6 +538,11 @@ class VariantTaskOutput(UworkerMsg):  # pylint: disable=abstract-method
   PROTO_CLS = uworker_msg_pb2.VariantTaskOutput
 
 
+class SymbolizeTaskOutput(UworkerMsg):  # pylint: disable=abstract-method
+  """Output from symbolize_task.uworker_main."""
+  PROTO_CLS = uworker_msg_pb2.SymbolizeTaskOutput
+
+
 class DeserializedUworkerMsg:
 
   def __init__(self, testcase=None, error_type=None, **kwargs):

src/clusterfuzz/_internal/protos/uworker_msg.proto

@@ -89,7 +89,13 @@ message Input {
 }
 
 message SymbolizeTaskOutput {
- // TODO(metzman): Fill this out.
+  optional string crash_type = 1;
+  optional string crash_address = 2;
+  optional string crash_state = 3;
+  optional string crash_stacktrace = 4;
+  optional bool symbolized = 5;
+  optional int32 crash_revision = 6;
+  optional string build_url = 7;
 }
 
 message AnalyzeTaskOutput {
@@ -189,6 +195,7 @@ enum ErrorType {
   PROGRESSION_BUILD_SETUP_ERROR = 18;
   REGRESSION_REVISION_LIST_ERROR = 19;
   REGRESSION_BUILD_NOT_FOUND = 20;
+  SYMBOLIZE_BUILD_SETUP_ERROR = 21;
 }
 
 message Output {