diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py index 750c517d4d..82350fd7b2 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py +++ b/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') diff --git a/src/clusterfuzz/_internal/metrics/fuzzer_stats.py b/src/clusterfuzz/_internal/metrics/fuzzer_stats.py index 463447aa49..c146a531cd 100644 --- a/src/clusterfuzz/_internal/metrics/fuzzer_stats.py +++ b/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': [ diff --git a/src/clusterfuzz/_internal/metrics/monitor.py b/src/clusterfuzz/_internal/metrics/monitor.py index c09bcf9c40..aa939ab5f9 100644 --- a/src/clusterfuzz/_internal/metrics/monitor.py +++ b/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 diff --git a/src/clusterfuzz/_internal/metrics/monitoring_metrics.py b/src/clusterfuzz/_internal/metrics/monitoring_metrics.py index a5baa186e5..2e5421f5be 100644 --- a/src/clusterfuzz/_internal/metrics/monitoring_metrics.py +++ b/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= diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/fuzz_task_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/fuzz_task_test.py index 9ec22bfa0e..0329f16b8f 100644 --- a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/fuzz_task_test.py +++ b/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): diff --git a/src/clusterfuzz/_internal/tests/core/metrics/fuzzer_stats_test.py b/src/clusterfuzz/_internal/tests/core/metrics/fuzzer_stats_test.py index 891c5a73a0..898159561f 100644 --- a/src/clusterfuzz/_internal/tests/core/metrics/fuzzer_stats_test.py +++ b/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.""" diff --git a/src/clusterfuzz/_internal/tests/core/metrics/monitor_test.py b/src/clusterfuzz/_internal/tests/core/metrics/monitor_test.py index 198fbdbcec..a71018e607 100644 --- a/src/clusterfuzz/_internal/tests/core/metrics/monitor_test.py +++ b/src/clusterfuzz/_internal/tests/core/metrics/monitor_test.py @@ -403,86 +403,6 @@ def test_cumulative_distribution_metric_geometric(self): ], result.buckets) -class TimeMetricTrackerTest(unittest.TestCase): - """Tests for TimeMetricTracker.""" - - def setUp(self): - monitor.metrics_store().reset_for_testing() - helpers.patch(self, [ - 'clusterfuzz._internal.metrics.monitor.check_module_loaded', - 'time.time', - ]) - self.mock.check_module_loaded.return_value = True - self.mock_time = helpers.MockTime(start_time=10.0) - self.mock.time.side_effect = self.mock_time.time - - def test_counter_metric_success(self): - """Test tracking time with CounterMetric and has_success_label=True.""" - metric = monitor.CounterMetric( - 'test_counter', - description='test', - field_spec=[ - monitor.StringField('tag'), - monitor.BooleanField('success'), - ]) - labels = {'tag': 'val'} - with monitor.TimeMetricTracker(metric, labels, has_success_label=True): - self.mock_time.advance(4.9) - - self.assertEqual({'tag': 'val'}, labels) - self.assertEqual(4, metric.get({'tag': 'val', 'success': True})) - - def test_counter_metric_fail_called(self): - """Test tracking time when fail() is explicitly called.""" - metric = monitor.CounterMetric( - 'test_counter', - description='test', - field_spec=[monitor.BooleanField('success')]) - with monitor.TimeMetricTracker(metric, has_success_label=True) as tracker: - self.mock_time.advance(3.2) - tracker.fail() - - self.assertEqual(3, metric.get({'success': False})) - - def test_counter_metric_exception_raised(self): - """Test tracking time when an exception is raised inside the block.""" - metric = monitor.CounterMetric( - 'test_counter', - description='test', - field_spec=[monitor.BooleanField('success')]) - with self.assertRaises(RuntimeError): - with monitor.TimeMetricTracker(metric, has_success_label=True): - self.mock_time.advance(7.6) - raise RuntimeError('test error') - - self.assertEqual(7, metric.get({'success': False})) - - def test_cumulative_distribution_metric(self): - """Test tracking float duration with CumulativeDistributionMetric.""" - metric = monitor.CumulativeDistributionMetric( - 'test_dist', - description='test', - bucketer=monitor.FixedWidthBucketer(width=1.0, num_finite_buckets=5), - field_spec=[monitor.StringField('tag')]) - with monitor.TimeMetricTracker(metric, {'tag': 'val'}): - self.mock_time.advance(2.5) - - dist = metric.get({'tag': 'val'}) - self.assertEqual(1, dist.count) - self.assertAlmostEqual(2.5, dist.sum) - - def test_gauge_metric(self): - """Test tracking float duration with Gauge.""" - metric = monitor.GaugeMetric( - 'test_gauge', - description='test', - field_spec=[monitor.StringField('tag')]) - with monitor.TimeMetricTracker(metric, {'tag': 'val'}): - self.mock_time.advance(3.7) - - self.assertEqual(3, metric.get({'tag': 'val'})) - - class TimeStatTrackerTest(unittest.TestCase): """Tests for TimeStatTracker."""