[Fuzz task steps duration] Add cleanup duration metric

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

@@ -1529,7 +1529,8 @@ def run_engine_fuzzer(engine_impl, target_name, sync_corpus_directory,
   _add_issue_metadata_from_environment(fuzzer_metadata)
 
   # Cleanup fuzzer temporary artifacts (e.g. mutations dir, merge dirs. etc).
-  fuzzer_utils.cleanup()
+  with steps_time_tracker.track('cleanup'):
+    fuzzer_utils.cleanup()
 
   return result, fuzzer_metadata, options.strategies
 
@@ -2033,7 +2034,8 @@ def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type):
 
       temp_queue = process_handler.get_queue()
       if not temp_queue:
-        process_handler.terminate_stale_application_instances()
+        with self.steps_time_tracker.track('cleanup'):
+          process_handler.terminate_stale_application_instances()
         logs.error('Unable to create temporary crash queue.')
         break
 
@@ -2052,7 +2054,8 @@ def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type):
         try:
           thread.start()
         except:
-          process_handler.terminate_stale_application_instances()
+          with self.steps_time_tracker.track('cleanup'):
+            process_handler.terminate_stale_application_instances()
           thread_error_occurred = True
           logs.error('Unable to start new thread.')
           break
@@ -2070,15 +2073,16 @@ def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type):
                           job_type) as tracker:
         tracker.timeout = utils.wait_until_timeout(threads, thread_timeout)
 
-      # Allow for some time to finish processing before terminating the
-      # processes.
-      process_handler.terminate_hung_threads(threads)
+      with self.steps_time_tracker.track('cleanup'):
+        # Allow for some time to finish processing before terminating the
+        # processes.
+        process_handler.terminate_hung_threads(threads)
 
-      # It is not necessary to clean up stale instances on every batch, but
-      # should be done at regular intervals to ensure we are in a good state.
-      if needs_stale_process_cleanup:
-        process_handler.terminate_stale_application_instances()
-        needs_stale_process_cleanup = False
+        # It is not necessary to clean up stale instances on every batch, but
+        # should be done at regular intervals to ensure we are in a good state.
+        if needs_stale_process_cleanup:
+          process_handler.terminate_stale_application_instances()
+          needs_stale_process_cleanup = False
 
       self._add_crashes_from_temp_queue(temp_queue, crashes)
       process_handler.close_queue(temp_queue)

src/clusterfuzz/_internal/metrics/fuzzer_stats.py

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

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

@@ -1663,6 +1663,23 @@ def test_utask_main_blackbox_fuzzer(self, mock_generate_blackbox_testcases):
     self.assertEqual(1, parse_dist.count)
     self.assertGreater(parse_dist.sum, 0)
 
+    # MAX_FUZZ_THREADS is 1, so the 3 testcases run in 3 batches, each followed
+    # by a cleanup. The task-level value is the sum over all batches.
+    cleanup_dist = monitoring_metrics.FUZZ_TASK_STEPS_DURATION.get({
+        'fuzzer': fuzzer.name,
+        'job': 'asan_test',
+        'platform': environment.platform(),
+        'runtime': environment.get_runtime().value,
+        'step': 'cleanup',
+        'success': True,
+    })
+    self.assertEqual(3, cleanup_dist.count)
+    self.assertEqual(3, self.mock.terminate_hung_threads.call_count)
+    total_cleanup = actual.fuzz_task_output.steps_duration[
+        'cleanup'].ToTimedelta().total_seconds()
+    self.assertGreater(total_cleanup, 0)
+    self.assertAlmostEqual(cleanup_dist.sum, total_cleanup, places=5)
+
   @mock.patch(
       'clusterfuzz._internal.bot.tasks.utasks.fuzz_task.FuzzingSession.generate_blackbox_testcases'
   )
@@ -2233,6 +2250,54 @@ def test_engine_prepare_exception(self):
     self.assertEqual(1, self._get_engine_prepare_dist(success=False).count)
     self.assertEqual(0, self._get_engine_prepare_dist(success=True).count)
 
+  def _get_cleanup_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': 'cleanup',
+        'success': success,
+    })
+
+  @mock.patch('clusterfuzz._internal.metrics.monitor.time')
+  @mock.patch('clusterfuzz._internal.bot.fuzzers.utils.cleanup')
+  def test_cleanup_accumulated_over_rounds(self, mock_cleanup,
+                                           mock_monitor_time):
+    """Test that cleanup is tracked once per round and accumulated."""
+    mock_time = helpers.MockTime(start_time=100.0)
+    mock_monitor_time.time.side_effect = mock_time.time
+    mock_cleanup.side_effect = lambda: mock_time.advance(1.25)
+    session = self._create_engine_fuzzing_session()
+    engine_impl = self._create_engine_impl()
+
+    session.do_engine_fuzzing(engine_impl)
+
+    # MAX_TESTCASES is 2, so there are two rounds of 1.25 seconds each.
+    self.assertEqual(2, mock_cleanup.call_count)
+    self.assertEqual(
+        datetime.timedelta(seconds=2.5),
+        session.steps_time_tracker.stats['cleanup'])
+    dist = self._get_cleanup_dist(success=True)
+    self.assertEqual(2, dist.count)
+    self.assertAlmostEqual(2.5, dist.sum)
+    self.assertEqual(0, self._get_cleanup_dist(success=False).count)
+
+  @mock.patch('clusterfuzz._internal.bot.fuzzers.utils.cleanup')
+  def test_cleanup_exception(self, mock_cleanup):
+    """Test that an exception in cleanup is tracked as a failure."""
+    mock_cleanup.side_effect = RuntimeError('cleanup failed')
+    session = self._create_engine_fuzzing_session()
+    engine_impl = self._create_engine_impl()
+
+    with self.assertRaises(RuntimeError):
+      session.do_engine_fuzzing(engine_impl)
+
+    self.assertIn('cleanup', session.steps_time_tracker.stats)
+    # The task stops at the first failing round.
+    self.assertEqual(1, self._get_cleanup_dist(success=False).count)
+    self.assertEqual(0, self._get_cleanup_dist(success=True).count)
+
   def test_basic(self):
     """Test basic fuzzing session."""
     target = 'test_target'