[Fuzz task steps duration] Add process_crash duration metric

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

@@ -2244,25 +2244,27 @@ def run(self):
     # Process and save crashes to datastore.
     bot_name = environment.get_value('BOT_NAME')
     project_name = environment.get_value('PROJECT_NAME')
-    crash_groups = process_crashes(
-        crashes=crashes,
-        context=Context(
-            project_name=project_name,
-            bot_name=bot_name,
-            job_type=self.job_type,
-            fuzz_target=self.fuzz_target,
-            redzone=self.redzone,
-            disable_ubsan=self.disable_ubsan,
-            platform_id=environment.get_platform_id(),
-            crash_revision=crash_revision,
-            fuzzer_name=self.fuzzer_name,
-            window_argument=self.window_argument,
-            fuzzer_metadata=fuzzer_metadata,
-            testcases_metadata=testcases_metadata,
-            timeout_multiplier=self.timeout_multiplier,
-            test_timeout=self.test_timeout,
-            data_directory=self.data_directory),
-        upload_urls=list(self.uworker_input.fuzz_task_input.crash_upload_urls))
+    with self.steps_time_tracker.track('process_crash'):
+      crash_groups = process_crashes(
+          crashes=crashes,
+          context=Context(
+              project_name=project_name,
+              bot_name=bot_name,
+              job_type=self.job_type,
+              fuzz_target=self.fuzz_target,
+              redzone=self.redzone,
+              disable_ubsan=self.disable_ubsan,
+              platform_id=environment.get_platform_id(),
+              crash_revision=crash_revision,
+              fuzzer_name=self.fuzzer_name,
+              window_argument=self.window_argument,
+              fuzzer_metadata=fuzzer_metadata,
+              testcases_metadata=testcases_metadata,
+              timeout_multiplier=self.timeout_multiplier,
+              test_timeout=self.test_timeout,
+              data_directory=self.data_directory),
+          upload_urls=list(
+              self.uworker_input.fuzz_task_input.crash_upload_urls))
 
     # Delete the fuzzed testcases. This was once explicitly needed since some
     # testcases resided on NFS and would otherwise be left forever. Now it's

src/clusterfuzz/_internal/metrics/fuzzer_stats.py

@@ -35,7 +35,13 @@
 
 STATS_FILE_EXTENSION = '.stats2'
 
-_STEPS_DURATION_FIELDS = []
+_STEPS_DURATION_FIELDS = [
+    {
+        'name': 'process_crash',
+        'type': 'INTERVAL',
+        'mode': 'NULLABLE',
+    },
+]
 JOB_RUN_SCHEMA = {
     'fields': [
         {

src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/fuzz_task_test.py

@@ -1388,6 +1388,8 @@ def setUp(self):
     self.mock.get_queue.return_value = queue.Queue()
     self.mock.get_process.return_value = threading.Thread
 
+    monitor.metrics_store().reset_for_testing()
+
   def _setup_utask_env(self):
     """Sets up the utask environment for fuzzing."""
     helpers.patch(self, [
@@ -1603,6 +1605,68 @@ def test_utask_main_blackbox_fuzzer(self, mock_generate_blackbox_testcases):
     self.assertEqual(3, actual.fuzz_task_output.testcases_generated)
     self.assertEqual(3, actual.fuzz_task_output.testcases_executed)
 
+    self.mock.process_crashes.assert_called_once()
+    self.assertGreater(
+        actual.fuzz_task_output.steps_duration['process_crash']
+        .ToTimedelta().total_seconds(), 0)
+    dist = monitoring_metrics.FUZZ_TASK_STEPS_DURATION.get({
+        'fuzzer': fuzzer.name,
+        'job': 'asan_test',
+        'platform': environment.platform(),
+        'runtime': environment.get_runtime().value,
+        'step': 'process_crash',
+        'success': True,
+    })
+    self.assertEqual(1, dist.count)
+    self.assertGreater(dist.sum, 0)
+
+  @mock.patch(
+      'clusterfuzz._internal.bot.tasks.utasks.fuzz_task.FuzzingSession.generate_blackbox_testcases'
+  )
+  def test_utask_main_process_crashes_exception(
+      self, mock_generate_blackbox_testcases):
+    """Test that an exception in crash processing is tracked as a failure."""
+    self._setup_utask_env()
+    self.mock.process_crashes.side_effect = RuntimeError('crash processing')
+    fuzzer = self._create_test_fuzzer()
+
+    fuzz_task_input = uworker_msg_pb2.FuzzTaskInput()
+    setup_input = uworker_msg_pb2.SetupInput(
+        fuzzer_name=fuzzer.name, fuzzer=uworker_io.entity_to_protobuf(fuzzer))
+
+    uworker_input = uworker_msg_pb2.Input(
+        fuzzer_name=fuzzer.name,
+        job_type='asan_test',
+        fuzz_task_input=fuzz_task_input,
+        setup_input=setup_input,
+        uworker_env={'FUZZ_TEST_TIMEOUT': '10'})
+
+    mock_generate_blackbox_testcases.return_value = (
+        fuzz_task.GenerateBlackboxTestcasesResult(
+            True, ['/tests/0', '/tests/1', '/tests/2'],
+            {'fuzzer_binary_name': fuzzer.name}))
+
+    with self.assertRaises(RuntimeError):
+      fuzz_task.utask_main(uworker_input)
+
+    labels = {
+        'fuzzer': fuzzer.name,
+        'job': 'asan_test',
+        'platform': environment.platform(),
+        'runtime': environment.get_runtime().value,
+        'step': 'process_crash',
+    }
+    failed_dist = monitoring_metrics.FUZZ_TASK_STEPS_DURATION.get({
+        **labels, 'success': False
+    })
+    self.assertEqual(1, failed_dist.count)
+    self.assertGreater(failed_dist.sum, 0)
+    self.assertEqual(
+        0,
+        monitoring_metrics.FUZZ_TASK_STEPS_DURATION.get({
+            **labels, 'success': True
+        }).count)
+
 
 class DoBlackboxFuzzingDeferredLoggingTest(fake_filesystem_unittest.TestCase):
   """do_blackbox_fuzzing deferred log upload unit tests without datastore dependency."""