[Fuzz task steps duration] Change corpus sync duration metric

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

@@ -15,7 +15,6 @@
 
 import collections
 import datetime
-import enum
 import itertools
 import json
 import os
@@ -624,26 +623,6 @@ def _track_testcase_run_result(fuzzer, job_type, new_crash_count,
       })
 
 
-class CorpusSyncStage(enum.Enum):
-  """Stages of corpus synchronization."""
-  PREPROCESS = 'preprocess'
-  DOWNLOAD = 'download'
-  UPLOAD = 'upload'
-
-
-def _track_sync_corpus_metric(fuzzer: str, job: str, stage: CorpusSyncStage):
-  """Context manager to track time spent synchronizing corpus with GCS."""
-  labels = {
-      'fuzzer': fuzzer,
-      'job': job,
-      'platform': environment.platform(),
-      'runtime': environment.get_runtime().value,
-      'stage': stage.value,
-  }
-  return monitor.TimeMetricTracker(
-      monitoring_metrics.SYNC_CORPUS_TIME, labels, has_success_label=True)
-
-
 def _last_sync_time(sync_file_path):
   """Read and parse the last sync file for the GCS corpus."""
   if not os.path.exists(sync_file_path):
@@ -1610,8 +1589,7 @@ def fully_qualified_fuzzer_name(self):
 
   def sync_corpus(self, sync_corpus_directory):
     """Sync corpus from GCS."""
-    with _track_sync_corpus_metric(self.fully_qualified_fuzzer_name,
-                                   self.job_type, CorpusSyncStage.DOWNLOAD):
+    with self.steps_time_tracker.track('corpus_download'):
       # Corpus should always be set at this point.
       self.gcs_corpus = GcsCorpus(self.fuzzer_name,
                                   self.fuzz_target.project_qualified_name(),
@@ -1634,9 +1612,7 @@ def _file_size(self, file_path):
 
   def sync_new_corpus_files(self):
     """Sync new files from corpus to GCS."""
-    with _track_sync_corpus_metric(self.fully_qualified_fuzzer_name,
-                                   self.job_type,
-                                   CorpusSyncStage.UPLOAD) as tracker:
+    with self.steps_time_tracker.track('corpus_upload'):
       new_files = self.gcs_corpus.get_new_files()
       new_files_count = len(new_files)
       logs.info('%d new corpus files generated by fuzzer %s (job %s).' %
@@ -1666,7 +1642,7 @@ def sync_new_corpus_files(self):
         success = bool(upload_success)
 
       if not success:
-        tracker.fail()
+        raise monitor.TimeStatTrackerFailureException
 
   def generate_blackbox_testcases(
       self, fuzzer, job_type, fuzzer_directory,
@@ -2516,16 +2492,14 @@ def _utask_preprocess(fuzzer_name, job_type, uworker_env):
     logs.log_contexts.add_metadata('fuzz_target', fuzz_target.binary)
     fuzz_task_input.fuzz_target.CopyFrom(
         uworker_io.entity_to_protobuf(fuzz_target))
-    with _track_sync_corpus_metric(fuzz_target.fully_qualified_name(), job_type,
-                                   CorpusSyncStage.PREPROCESS):
-      fuzz_task_input.corpus.CopyFrom(
-          corpus_manager.get_fuzz_target_corpus(
-              fuzzer_name,
-              fuzz_target.project_qualified_name(),
-              include_delete_urls=False,
-              max_upload_urls=_get_max_corpus_uploads_per_task(),
-              max_download_urls=25000,
-              use_backup=True).serialize())
+    fuzz_task_input.corpus.CopyFrom(
+        corpus_manager.get_fuzz_target_corpus(
+            fuzzer_name,
+            fuzz_target.project_qualified_name(),
+            include_delete_urls=False,
+            max_upload_urls=_get_max_corpus_uploads_per_task(),
+            max_download_urls=25000,
+            use_backup=True).serialize())
 
   fuzz_task_input.trials.extend(trials.preprocess_get_db_trials())
   web_tests_url = environment.get_value('WEB_TESTS_URL')

src/clusterfuzz/_internal/metrics/fuzzer_stats.py

@@ -76,6 +76,16 @@
         'type': 'INTERVAL',
         'mode': 'NULLABLE',
     },
+    {
+        'name': 'corpus_download',
+        'type': 'INTERVAL',
+        'mode': 'NULLABLE',
+    },
+    {
+        'name': 'corpus_upload',
+        'type': 'INTERVAL',
+        'mode': 'NULLABLE',
+    },
 ]
 JOB_RUN_SCHEMA = {
     'fields': [

src/clusterfuzz/_internal/metrics/monitor.py

@@ -627,44 +627,6 @@ def insert(self, value: float, labels: dict):
     self.job_metric.insert(value, labels=job_labels)
 
 
-class TimeMetricTracker:
-  """Context manager to track time for any metric."""
-
-  def __init__(self,
-               metric: Metric | SplitMetricHelper,
-               labels=None,
-               has_success_label: bool = False):
-    """Initializes the tracker.
-
-    Args:
-      metric: The metric which is being tracked.
-      labels: The dictionary containing the labels of the metric.
-      has_success_label: If True, a label 'success' is added to the labels
-        dictionary upon exit, which is True if and only if no exception
-        was raised and the fail method was not called.
-    """
-    self.metric = metric
-    self.labels = dict(labels) if labels else {}
-    self.has_success_label = has_success_label
-    self.start_time = None
-    self.failed = False
-
-  def __enter__(self):
-    self.start_time = time.time()
-    self.failed = False
-    return self
-
-  def __exit__(self, exc_type, value, traceback):
-    duration = time.time() - self.start_time
-    if self.has_success_label:
-      self.labels['success'] = not self.failed and exc_type is None
-
-    self.metric.insert(duration, labels=self.labels)
-
-  def fail(self):
-    self.failed = True
-
-
 class TimeStatTrackerFailureException(Exception):
   pass
 

src/clusterfuzz/_internal/metrics/monitoring_metrics.py

@@ -34,33 +34,6 @@
         monitor.BooleanField('success'),
     ])
 
-FUZZER_SYNC_CORPUS_TIME = monitor.CounterMetric(
-    'task/fuzz/fuzzer/sync_corpus_time',
-    description=('The total time spent setting up and synchronizing corpus '
-                 'files with GCS in seconds (grouped by fuzzer).'),
-    field_spec=[
-        monitor.StringField('fuzzer'),
-        monitor.StringField('platform'),
-        monitor.StringField('runtime'),
-        monitor.StringField('stage'),
-        monitor.BooleanField('success'),
-    ])
-
-JOB_SYNC_CORPUS_TIME = monitor.CounterMetric(
-    'task/fuzz/job/sync_corpus_time',
-    description=('The total time spent setting up and synchronizing corpus '
-                 'files with GCS in seconds (grouped by job).'),
-    field_spec=[
-        monitor.StringField('job'),
-        monitor.StringField('platform'),
-        monitor.StringField('runtime'),
-        monitor.StringField('stage'),
-        monitor.BooleanField('success'),
-    ])
-
-SYNC_CORPUS_TIME = monitor.SplitMetricHelper(FUZZER_SYNC_CORPUS_TIME,
-                                             JOB_SYNC_CORPUS_TIME)
-
 CF_TIP_BOOT_FAILED_COUNT = monitor.CounterMetric(
     'tip_boot_failure',
     description=

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

@@ -242,7 +242,7 @@ def test_timeout(self):
 
 
 class TrackSyncCorpusTimeTest(unittest.TestCase):
-  """Test SYNC_CORPUS_TIME metric emission during corpus sync."""
+  """Test corpus sync steps emitted to FUZZ_TASK_STEPS_DURATION."""
 
   def setUp(self):
     monitor.metrics_store().reset_for_testing()
@@ -268,29 +268,28 @@ def setUp(self):
         engine='libFuzzer', binary='test_target')
     self.session.data_directory = '/data'
 
-  def _assert_metrics(self, expected_value, stage, success):
-    """Assert that both fuzzer and job sync corpus metrics match expected."""
-    common_labels = {
+  def _get_dist(self, step, success):
+    return monitoring_metrics.FUZZ_TASK_STEPS_DURATION.get({
+        'fuzzer': 'libFuzzer',
+        'job': 'libfuzzer_asan_test',
         'platform': 'some_platform',
         'runtime': 'kata_container',
-        'stage': stage,
+        'step': step,
         'success': success,
-    }
-    self.assertEqual(
-        expected_value,
-        monitoring_metrics.FUZZER_SYNC_CORPUS_TIME.get({
-            **common_labels,
-            'fuzzer': 'libFuzzer_test_target',
-        }))
+    })
+
+  def _assert_step(self, step, expected_seconds, success):
+    """Assert that a single measurement of the step was emitted and saved."""
+    dist = self._get_dist(step, success)
+    self.assertEqual(1, dist.count)
+    self.assertAlmostEqual(expected_seconds, dist.sum)
+    self.assertEqual(0, self._get_dist(step, not success).count)
     self.assertEqual(
-        expected_value,
-        monitoring_metrics.JOB_SYNC_CORPUS_TIME.get({
-            **common_labels,
-            'job': 'libfuzzer_asan_test',
-        }))
+        datetime.timedelta(seconds=expected_seconds),
+        self.session.steps_time_tracker.stats[step])
 
   def test_sync_corpus_success(self):
-    """Test metric recorded when sync_corpus succeeds."""
+    """Test corpus_download step recorded when sync_corpus succeeds."""
 
     def sync_side_effect():
       self.mock_time.advance(5.8)
@@ -300,10 +299,11 @@ def sync_side_effect():
         sync_side_effect)
 
     self.session.sync_corpus('/corpus')
-    self._assert_metrics(5, stage='download', success=True)
+    self._assert_step('corpus_download', 5.8, success=True)
 
   def test_sync_corpus_failure(self):
-    """Test metric recorded when sync_corpus fails and raises FuzzTaskError."""
+    """Test corpus_download step recorded as failure when sync_corpus fails
+    and raises FuzzTaskError."""
 
     def sync_side_effect():
       self.mock_time.advance(5.8)
@@ -315,10 +315,10 @@ def sync_side_effect():
     with self.assertRaises(fuzz_task.FuzzTaskError):
       self.session.sync_corpus('/corpus')
 
-    self._assert_metrics(5, stage='download', success=False)
+    self._assert_step('corpus_download', 5.8, success=False)
 
   def test_sync_new_corpus_files_success(self):
-    """Test metric recorded when sync_new_corpus_files succeeds."""
+    """Test corpus_upload step recorded when sync_new_corpus_files succeeds."""
     self.session.gcs_corpus = mock.MagicMock()
     self.session.gcs_corpus.get_new_files.return_value = []
 
@@ -329,10 +329,11 @@ def upload_side_effect(_):
     self.session.gcs_corpus.upload_files.side_effect = upload_side_effect
 
     self.session.sync_new_corpus_files()
-    self._assert_metrics(5, stage='upload', success=True)
+    self._assert_step('corpus_upload', 5.8, success=True)
 
   def test_sync_new_corpus_files_failure(self):
-    """Test metric recorded when sync_new_corpus_files fails."""
+    """Test corpus_upload step recorded as failure, without raising, when
+    some uploads fail."""
     self.session.gcs_corpus = mock.MagicMock()
     self.session.gcs_corpus.get_new_files.return_value = []
 
@@ -343,62 +344,22 @@ def upload_side_effect(_):
     self.session.gcs_corpus.upload_files.side_effect = upload_side_effect
 
     self.session.sync_new_corpus_files()
-    self._assert_metrics(5, stage='upload', success=False)
-
-  def _patch_preprocess(self):
-    """Patch dependencies of _utask_preprocess."""
-    helpers.patch(self, [
-        'clusterfuzz._internal.bot.tasks.setup.preprocess_update_fuzzer_and_data_bundles',
-        'clusterfuzz._internal.bot.tasks.trials.preprocess_get_db_trials',
-        'clusterfuzz._internal.bot.tasks.utasks.fuzz_task._preprocess_get_fuzz_target',
-        'clusterfuzz._internal.bot.tasks.utasks.fuzz_task.preprocess_store_fuzzer_run_results',
-        'clusterfuzz._internal.bot.tasks.utasks.fuzz_task_knobs.do_multiarmed_bandit_strategy_selection',
-        'clusterfuzz._internal.bot.tasks.utasks.uworker_io.entity_to_protobuf',
-        'clusterfuzz._internal.datastore.data_handler.get_project_name',
-        'clusterfuzz._internal.fuzzing.corpus_manager.get_fuzz_target_corpus',
-        'clusterfuzz._internal.google_cloud_utils.blobs.generate_new_blob_name',
-        'clusterfuzz._internal.google_cloud_utils.blobs.get_signed_upload_url',
-    ])
-    self.mock.preprocess_update_fuzzer_and_data_bundles.return_value = (
-        uworker_msg_pb2.SetupInput())
-    self.mock.entity_to_protobuf.return_value = (
-        uworker_msg_pb2.FuzzTaskInput().fuzz_target)
-    self.mock.get_project_name.return_value = 'test_project'
-    self.mock._preprocess_get_fuzz_target.return_value = (
-        self.session.fuzz_target)
-    self.mock.preprocess_get_db_trials.return_value = []
-    self.mock.generate_new_blob_name.return_value = 'blob_key'
-    self.mock.get_signed_upload_url.return_value = 'https://signed_url'
-
-  def test_preprocess_sync_corpus_success(self):
-    """Test metric recorded during _utask_preprocess when getting corpus."""
-    self._patch_preprocess()
-
-    def get_corpus_side_effect(*args, **kwargs):  # pylint: disable=unused-argument
-      self.mock_time.advance(5.8)
-      mock_corpus = mock.MagicMock()
-      mock_corpus.serialize.return_value = uworker_msg_pb2.FuzzTargetCorpus()
-      return mock_corpus
-
-    self.mock.get_fuzz_target_corpus.side_effect = get_corpus_side_effect
-
-    fuzz_task._utask_preprocess('libFuzzer', 'libfuzzer_asan_test', {})
-    self._assert_metrics(5, stage='preprocess', success=True)
+    self._assert_step('corpus_upload', 5.8, success=False)
 
-  def test_preprocess_sync_corpus_failure(self):
-    """Test metric recorded when get_fuzz_target_corpus raises in preprocess."""
-    self._patch_preprocess()
-
-    def get_corpus_side_effect(*args, **kwargs):  # pylint: disable=unused-argument
-      self.mock_time.advance(5.8)
-      raise RuntimeError('GCS failure')
+  def test_sync_new_corpus_files_upload_returns_false(self):
+    """Test corpus_upload step recorded as failure when upload_files returns a
+    falsy non-list value."""
+    self.session.gcs_corpus = mock.MagicMock()
+    self.session.gcs_corpus.get_new_files.return_value = []
 
-    self.mock.get_fuzz_target_corpus.side_effect = get_corpus_side_effect
+    def upload_side_effect(_):
+      self.mock_time.advance(2.5)
+      return False
 
-    with self.assertRaises(RuntimeError):
-      fuzz_task._utask_preprocess('libFuzzer', 'libfuzzer_asan_test', {})
+    self.session.gcs_corpus.upload_files.side_effect = upload_side_effect
 
-    self._assert_metrics(5, stage='preprocess', success=False)
+    self.session.sync_new_corpus_files()
+    self._assert_step('corpus_upload', 2.5, success=False)
 
 
 class GetFuzzerMetadataFromOutputTest(unittest.TestCase):

src/clusterfuzz/_internal/tests/core/metrics/fuzzer_stats_test.py

@@ -250,12 +250,26 @@ def test_job_run_steps_duration_all_unknown(self, mock_log_error):
 
   @mock.patch('clusterfuzz._internal.metrics.logs.error')
   def test_job_run_steps_duration_empty_schema(self, mock_log_error):
-    """Tests that with the default (empty) schema all keys are dropped."""
+    """Tests that keys missing from the default schema are dropped."""
     job_run = self._make_job_run({'fuzzing': datetime.timedelta(seconds=1)})
 
     self.assertNotIn('steps_duration', job_run.data)
     mock_log_error.assert_called_once()
 
+  @mock.patch('clusterfuzz._internal.metrics.logs.error')
+  def test_job_run_steps_duration_corpus_sync(self, mock_log_error):
+    """Tests that the corpus sync steps are accepted by the default schema."""
+    job_run = self._make_job_run({
+        'corpus_download': datetime.timedelta(seconds=4.5),
+        'corpus_upload': datetime.timedelta(seconds=1.25),
+    })
+
+    self.assertEqual({
+        'corpus_download': 'P0DT0H0M4.500000S',
+        'corpus_upload': 'P0DT0H0M1.250000S',
+    }, job_run['steps_duration'])
+    mock_log_error.assert_not_called()
+
   @mock.patch('clusterfuzz._internal.metrics.logs.error')
   def test_job_run_steps_duration_none(self, mock_log_error):
     """Tests that no steps_duration produces no field and no logs."""