[Fuzz task steps duration] Add fuzz duration metric

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

@@ -1484,8 +1484,9 @@ def run_engine_fuzzer(engine_impl, target_name, sync_corpus_directory,
     raise FuzzTaskError(f'Invalid engine timeout: '
                         f'{fuzz_test_timeout} - {additional_processing_time}')
 
-  result = engine_impl.fuzz(target_path, options, testcase_directory,
-                            adjusted_fuzz_test_timeout)
+  with steps_time_tracker.track('fuzz'):
+    result = engine_impl.fuzz(target_path, options, testcase_directory,
+                              adjusted_fuzz_test_timeout)
 
   logs.info('Used strategies.', strategies=options.strategies)
   for strategy, value in options.strategies.items():
@@ -2010,44 +2011,48 @@ def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type):
 
       temp_queue = process_handler.get_queue()
       if not temp_queue:
-        with self.steps_time_tracker.track('cleanup'):
-          process_handler.terminate_stale_application_instances()
+        process_handler.terminate_stale_application_instances()
         logs.error('Unable to create temporary crash queue.')
         break
 
-      while thread_index < max_threads and test_number < len(
-          testcase_file_paths):
-        testcase_file_path = testcase_file_paths[test_number]
-        gestures = testcases_metadata[testcase_file_path]['gestures']
+      with self.steps_time_tracker.track('fuzz'):
+        failed = False
+        while thread_index < max_threads and test_number < len(
+            testcase_file_paths):
+          testcase_file_path = testcase_file_paths[test_number]
+          gestures = testcases_metadata[testcase_file_path]['gestures']
 
-        env_copy = environment.copy()
+          env_copy = environment.copy()
 
-        thread = process_handler.get_process()(
-            target=testcase_manager.run_testcase_and_return_result_in_queue,
-            args=(temp_queue, thread_index, testcase_file_path, gestures,
-                  env_copy))
+          thread = process_handler.get_process()(
+              target=testcase_manager.run_testcase_and_return_result_in_queue,
+              args=(temp_queue, thread_index, testcase_file_path, gestures,
+                    env_copy))
 
-        try:
-          thread.start()
-        except:
-          with self.steps_time_tracker.track('cleanup'):
+          try:
+            thread.start()
+          except:
             process_handler.terminate_stale_application_instances()
-          thread_error_occurred = True
-          logs.error('Unable to start new thread.')
-          break
+            thread_error_occurred = True
+            logs.error('Unable to start new thread.')
+            failed = True
+            break
 
-        threads.append(thread)
-        thread_index += 1
-        test_number += 1
+          threads.append(thread)
+          thread_index += 1
+          test_number += 1
 
-        if test_number % testcases_before_stale_process_cleanup == 0:
-          needs_stale_process_cleanup = True
+          if test_number % testcases_before_stale_process_cleanup == 0:
+            needs_stale_process_cleanup = True
 
-        time.sleep(thread_delay)
+          time.sleep(thread_delay)
+
+        with _TrackFuzzTime(self.fully_qualified_fuzzer_name,
+                            job_type) as tracker:
+          tracker.timeout = utils.wait_until_timeout(threads, thread_timeout)
 
-      with _TrackFuzzTime(self.fully_qualified_fuzzer_name,
-                          job_type) as tracker:
-        tracker.timeout = utils.wait_until_timeout(threads, thread_timeout)
+        if failed:
+          raise monitor.TimeStatTrackerFailureException
 
       with self.steps_time_tracker.track('cleanup'):
         # Allow for some time to finish processing before terminating the

src/clusterfuzz/_internal/metrics/fuzzer_stats.py

@@ -86,6 +86,11 @@
         'type': 'INTERVAL',
         'mode': 'NULLABLE',
     },
+    {
+        'name': 'fuzz',
+        'type': 'INTERVAL',
+        'mode': 'NULLABLE',
+    },
 ]
 JOB_RUN_SCHEMA = {
     'fields': [

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

@@ -1655,6 +1655,16 @@ def test_utask_main_blackbox_fuzzer(self, mock_generate_blackbox_testcases):
     self.assertGreater(total_cleanup, 0)
     self.assertAlmostEqual(cleanup_dist.sum, total_cleanup, places=5)
 
+    # MAX_FUZZ_THREADS is 1, so the 3 testcases run in 3 batches, and the
+    # wait for each batch is tracked as a fuzz step.
+    fuzz_dist = self._get_fuzz_dist(fuzzer.name, success=True)
+    self.assertEqual(3, fuzz_dist.count)
+    self.assertAlmostEqual(
+        fuzz_dist.sum,
+        actual.fuzz_task_output.steps_duration['fuzz'].ToTimedelta()
+        .total_seconds(),
+        places=5)
+
   @mock.patch(
       'clusterfuzz._internal.bot.tasks.utasks.fuzz_task.FuzzingSession.generate_blackbox_testcases'
   )
@@ -1863,6 +1873,63 @@ def _create_blackbox_uworker_input(self, fuzzer):
         setup_input=setup_input,
         uworker_env={'FUZZ_TEST_TIMEOUT': '10'})
 
+  def _get_fuzz_dist(self, fuzzer_name, success):
+    return monitoring_metrics.FUZZ_TASK_STEPS_DURATION.get({
+        'fuzzer': fuzzer_name,
+        'job': 'asan_test',
+        'platform': environment.platform(),
+        'runtime': environment.get_runtime().value,
+        'step': 'fuzz',
+        'success': success,
+    })
+
+  @mock.patch('clusterfuzz._internal.metrics.monitor.time')
+  @mock.patch('clusterfuzz._internal.base.utils.wait_until_timeout')
+  @mock.patch(
+      'clusterfuzz._internal.bot.tasks.utasks.fuzz_task.FuzzingSession.generate_blackbox_testcases'
+  )
+  def test_utask_main_blackbox_fuzz_accumulated(
+      self, mock_generate_blackbox_testcases, mock_wait_until_timeout,
+      mock_monitor_time):
+    """Test that the fuzz step is accumulated over all testcase batches."""
+    self._setup_utask_env()
+    fuzzer = self._create_test_fuzzer()
+    mock_time = helpers.MockTime(start_time=100.0)
+    mock_monitor_time.time.side_effect = mock_time.time
+
+    def wait_until_timeout(threads, _):
+      mock_time.advance(2.0)
+      for thread in threads:
+        thread.join()
+      return False
+
+    mock_wait_until_timeout.side_effect = wait_until_timeout
+
+    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}))
+
+    actual = fuzz_task.utask_main(uworker_input)
+
+    # 3 batches of 2 seconds each.
+    self.assertEqual(3, mock_wait_until_timeout.call_count)
+    self.assertEqual(
+        datetime.timedelta(seconds=6),
+        actual.fuzz_task_output.steps_duration['fuzz'].ToTimedelta())
+    fuzz_dist = self._get_fuzz_dist(fuzzer.name, success=True)
+    self.assertEqual(3, fuzz_dist.count)
+    self.assertAlmostEqual(6.0, fuzz_dist.sum)
+
 
 class DoBlackboxFuzzingDeferredLoggingTest(fake_filesystem_unittest.TestCase):
   """do_blackbox_fuzzing deferred log upload unit tests without datastore dependency."""
@@ -2320,6 +2387,57 @@ def test_cleanup_exception(self, mock_cleanup):
     self.assertEqual(1, self._get_cleanup_dist(success=False).count)
     self.assertEqual(0, self._get_cleanup_dist(success=True).count)
 
+  def _get_fuzz_dist(self, success):
+    return monitoring_metrics.FUZZ_TASK_STEPS_DURATION.get({
+        'fuzzer': 'libFuzzer_fuzz',
+        'job': 'libfuzzer_asan_test',
+        'platform': environment.platform(),
+        'runtime': environment.get_runtime().value,
+        'step': 'fuzz',
+        'success': success,
+    })
+
+  @mock.patch('clusterfuzz._internal.metrics.monitor.time')
+  def test_fuzz_accumulated_over_rounds(self, mock_monitor_time):
+    """Test that the engine fuzz call is tracked per round and accumulated."""
+    mock_time = helpers.MockTime(start_time=100.0)
+    mock_monitor_time.time.side_effect = mock_time.time
+    session = self._create_engine_fuzzing_session()
+    engine_impl = self._create_engine_impl()
+
+    def fuzz(*_):
+      mock_time.advance(10.0)
+      return engine.FuzzResult('logs', ['cmd'], [], {'stat': 1}, 10.0)
+
+    engine_impl.fuzz.side_effect = fuzz
+
+    session.do_engine_fuzzing(engine_impl)
+
+    # MAX_TESTCASES is 2, so there are two rounds of 10 seconds each.
+    self.assertEqual(2, engine_impl.fuzz.call_count)
+    self.assertEqual(
+        datetime.timedelta(seconds=20),
+        session.steps_time_tracker.stats['fuzz'])
+    dist = self._get_fuzz_dist(success=True)
+    self.assertEqual(2, dist.count)
+    self.assertAlmostEqual(20.0, dist.sum)
+    self.assertEqual(0, self._get_fuzz_dist(success=False).count)
+
+  def test_fuzz_exception(self):
+    """Test that an exception in the engine fuzz call is tracked as a
+    failure."""
+    session = self._create_engine_fuzzing_session()
+    engine_impl = self._create_engine_impl()
+    engine_impl.fuzz.side_effect = RuntimeError('fuzz failed')
+
+    with self.assertRaises(RuntimeError):
+      session.do_engine_fuzzing(engine_impl)
+
+    self.assertIn('fuzz', session.steps_time_tracker.stats)
+    # The task stops at the first failing round.
+    self.assertEqual(1, self._get_fuzz_dist(success=False).count)
+    self.assertEqual(0, self._get_fuzz_dist(success=True).count)
+
   def test_basic(self):
     """Test basic fuzzing session."""
     target = 'test_target'