From 6d392ef4e2d40cd83e91966de629923130926227 Mon Sep 17 00:00:00 2001 From: Steffen Smolka Date: Mon, 26 Oct 2026 23:45:32 -0600 Subject: [PATCH] Emit structured stage_summary telemetry across all 11 bot tasks. Currently, diagnosing where a ClusterFuzz task failed or spent its time requires scraping unstructured Cloud Logging prose across preprocess, main, and postprocess stages. With this change, every stage exit emits a single `Stage finished.` log carrying `jsonPayload.extras.stage_summary` with stage status, duration, error type, variant, and per-step status, duration, and failure or skip reason across all 8 UTasks and 3 TrustedTasks. Also provision a filtered Cloud Logging sink (`stage-summaries`) in Terraform that routes only `stage_summary` records into a partitioned BigQuery dataset (`stage_summaries`). --- infra/terraform/monitoring.tf | 29 ++ .../_internal/bot/tasks/blame_task.py | 5 + .../_internal/bot/tasks/impact_task.py | 11 + .../_internal/bot/tasks/task_types.py | 3 +- .../_internal/bot/tasks/unpack_task.py | 9 + .../_internal/bot/tasks/utasks/__init__.py | 93 ++++- .../bot/tasks/utasks/analyze_task.py | 17 + .../bot/tasks/utasks/corpus_pruning_task.py | 14 + .../_internal/bot/tasks/utasks/fuzz_task.py | 27 +- .../bot/tasks/utasks/minimize_task.py | 36 +- .../bot/tasks/utasks/progression_task.py | 42 +- .../bot/tasks/utasks/regression_task.py | 24 ++ .../bot/tasks/utasks/symbolize_task.py | 14 + .../bot/tasks/utasks/variant_task.py | 14 +- src/clusterfuzz/_internal/metrics/logs.py | 388 ++++++++++++++++++ .../tests/core/bot/tasks/impact_task_test.py | 65 +++ .../tests/core/bot/tasks/task_types_test.py | 18 + .../bot/tasks/utasks/analyze_task_test.py | 37 ++ .../core/bot/tasks/utasks/utasks_test.py | 70 ++++ .../_internal/tests/core/metrics/logs_test.py | 253 ++++++++++++ 20 files changed, 1150 insertions(+), 19 deletions(-) diff --git a/infra/terraform/monitoring.tf b/infra/terraform/monitoring.tf index 0c755e10256..28b08a235fb 100644 --- a/infra/terraform/monitoring.tf +++ b/infra/terraform/monitoring.tf @@ -1632,3 +1632,32 @@ resource "google_monitoring_dashboard" "clusterfuzz_sli_dashboard" { } JSON } + +resource "google_bigquery_dataset" "stage_summaries" { + project = var.project_id + dataset_id = "stage_summaries" + friendly_name = "ClusterFuzz Stage Summaries" + description = "Structured stage and step telemetry exported from Cloud Logging logs/python." + location = "US" + default_table_expiration_ms = 7776000000 # 90 days +} + +resource "google_logging_project_sink" "stage_summaries" { + project = var.project_id + name = "stage-summaries" + destination = "bigquery.googleapis.com/projects/${var.project_id}/datasets/${google_bigquery_dataset.stage_summaries.dataset_id}" + filter = "logName=\"projects/${var.project_id}/logs/python\" AND jsonPayload.extras.stage_summary.stage:*" + unique_writer_identity = true + + bigquery_options { + use_partitioned_tables = true + } +} + +resource "google_bigquery_dataset_iam_member" "stage_summaries_sink_writer" { + project = var.project_id + dataset_id = google_bigquery_dataset.stage_summaries.dataset_id + role = "roles/bigquery.dataEditor" + member = google_logging_project_sink.stage_summaries.writer_identity +} + diff --git a/src/clusterfuzz/_internal/bot/tasks/blame_task.py b/src/clusterfuzz/_internal/bot/tasks/blame_task.py index 690e0b079b0..a56975658c8 100644 --- a/src/clusterfuzz/_internal/bot/tasks/blame_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/blame_task.py @@ -271,6 +271,7 @@ def _clear_blame_result_and_set_pending_flag(testcase): def _execute_task(testcase_id): """Attempt to find the CL introducing the bug associated with testcase_id.""" + step = logs.step(1, 'validate_testcase') # Locate the testcase associated with the id. testcase = data_handler.get_testcase_by_id(testcase_id) @@ -278,20 +279,24 @@ def _execute_task(testcase_id): topic = db_config.get_value('predator_crash_topic') if not topic: logs.info('Predator is not configured, skipping blame task.') + step.complete_stage('Predator is not configured', skip=True) return data_handler.update_testcase_comment(testcase, data_types.TaskState.STARTED) # Prepare pubsub message to send to predator. + step = logs.step(2, 'compute_dependency_rolls') message = _prepare_predator_message(testcase) if not message: testcase = data_handler.get_testcase_by_id(testcase_id) data_handler.update_testcase_comment( testcase, data_types.TaskState.ERROR, 'Failed to generate request for Predator') + step.fail('Failed to generate request for Predator') return # Clear existing results and mark blame result as pending. + logs.step(3, 'publish_to_predator') testcase = data_handler.get_testcase_by_id(testcase_id) _clear_blame_result_and_set_pending_flag(testcase) diff --git a/src/clusterfuzz/_internal/bot/tasks/impact_task.py b/src/clusterfuzz/_internal/bot/tasks/impact_task.py index b1f86cb4844..1bd359b3a71 100644 --- a/src/clusterfuzz/_internal/bot/tasks/impact_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/impact_task.py @@ -273,8 +273,10 @@ def _execute_task(testcase_id, job_type): # We don't need job_type but it's supplied to all tasks. del job_type + step = logs.step(1, 'validate_testcase') # This shouldn't ever get scheduled, but check just in case. if not utils.is_chromium(): + step.complete_stage('Not applicable for non-Chromium', skip=True) return # Locate the testcase associated with the id. @@ -282,11 +284,13 @@ def _execute_task(testcase_id, job_type): # If this testcase is fixed, we should no longer be doing impact testing. if testcase.fixed and testcase.is_impact_set_flag: + step.complete_stage('Testcase already fixed with impact set', skip=True) return # For testcases with status unreproducible, we just do impact analysis just # once. if testcase.is_status_unreproducible() and testcase.is_impact_set_flag: + step.complete_stage('Unreproducible with impact already set', skip=True) return # Update comments only after checking the above bailout conditions. @@ -297,6 +301,8 @@ def _execute_task(testcase_id, job_type): data_handler.update_testcase_comment( testcase, data_types.TaskState.ERROR, 'Not applicable for unreproducible testcases') + step.complete_stage( + 'Not applicable for unreproducible testcases', skip=True) return # This task is not applicable for custom binaries. We cannot remove the @@ -306,8 +312,10 @@ def _execute_task(testcase_id, job_type): data_handler.update_testcase_comment(testcase, data_types.TaskState.FINISHED, 'Not applicable for custom binaries') + step.complete_stage('Not applicable for custom binaries', skip=True) return + step = logs.step(2, 'check_regression_range') logs.info('Preparing to calculate impact.') # Formerly ClusterFuzz had buckets containing builds for stable, # beta and dev builds, and attempted reproduction on them. That had @@ -321,10 +329,13 @@ def _execute_task(testcase_id, job_type): testcase, data_types.TaskState.FINISHED, 'Cannot run without regression range, will re-run once regression ' 'task finishes') + step.complete_stage('Cannot run without regression range', skip=True) return + logs.step(3, 'calculate_impacts') logs.info('Calculating impact from URL.') impacts = get_impacts_from_url(testcase.regression, testcase.job_type) + logs.step(4, 'save_impacts') testcase = data_handler.get_testcase_by_id(testcase_id) set_testcase_with_impacts(testcase, impacts) testcase_utils.emit_testcase_triage_duration_metric( diff --git a/src/clusterfuzz/_internal/bot/tasks/task_types.py b/src/clusterfuzz/_internal/bot/tasks/task_types.py index 909fcea3d00..b28af95d09a 100644 --- a/src/clusterfuzz/_internal/bot/tasks/task_types.py +++ b/src/clusterfuzz/_internal/bot/tasks/task_types.py @@ -60,7 +60,8 @@ def execute(self, task_argument, job_type, uworker_env): events.emit_task_event(task_command, event_data, events.TaskStatus.STARTED) try: - self.module.execute_task(task_argument, job_type) + with logs.stage_summary_context(logs.Stage.MAIN): + self.module.execute_task(task_argument, job_type) except Exception as e: events.emit_task_event( task_command, diff --git a/src/clusterfuzz/_internal/bot/tasks/unpack_task.py b/src/clusterfuzz/_internal/bot/tasks/unpack_task.py index bd24f27efbc..0dd5013a977 100644 --- a/src/clusterfuzz/_internal/bot/tasks/unpack_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/unpack_task.py @@ -31,9 +31,11 @@ def execute_task(metadata_id, job_type): """Unpack a bundled testcase archive and create analyze jobs for each item.""" + step = logs.step(1, 'load_bundle_metadata') metadata = ndb.Key(data_types.BundledArchiveMetadata, int(metadata_id)).get() if not metadata: logs.error('Invalid bundle metadata id %s.' % metadata_id) + step.fail('Invalid bundle metadata id %s.' % metadata_id) return bot_name = environment.get_value('BOT_NAME') @@ -42,11 +44,13 @@ def execute_task(metadata_id, job_type): metadata.blobstore_key).get() if not upload_metadata: logs.error('Invalid upload metadata key %s.' % metadata.blobstore_key) + step.fail('Invalid upload metadata key %s.' % metadata.blobstore_key) return job = data_types.Job.query(data_types.Job.name == metadata.job_type).get() if not job: logs.error('Invalid job_type %s.' % metadata.job_type) + step.fail('Invalid job_type %s.' % metadata.job_type) return # Update the upload metadata with this bot name. @@ -55,12 +59,14 @@ def execute_task(metadata_id, job_type): # We can't use FUZZ_INPUTS directory since it is constrained # by tmpfs limits. + step = logs.step(2, 'download_and_extract_archive') testcases_directory = environment.get_value('FUZZ_INPUTS_DISK') # Retrieve multi-testcase archive. archive_path = os.path.join(testcases_directory, metadata.archive_filename) if not blobs.read_blob_to_disk(metadata.blobstore_key, archive_path): logs.error('Could not retrieve archive for bundle %d.' % metadata_id) + step.fail('Could not retrieve archive for bundle %d.' % metadata_id) tasks.add_task('unpack', metadata_id, job_type) return @@ -69,10 +75,12 @@ def execute_task(metadata_id, job_type): reader.extract_all(testcases_directory) except: logs.error('Could not unpack archive for bundle %d.' % metadata_id) + step.fail('Could not unpack archive for bundle %d.' % metadata_id) tasks.add_task('unpack', metadata_id, job_type) return # Get additional testcase metadata (if any). + logs.step(3, 'create_testcases') additional_metadata = None if upload_metadata.additional_metadata_string: additional_metadata = json.loads(upload_metadata.additional_metadata_string) @@ -112,6 +120,7 @@ def execute_task(metadata_id, job_type): # The upload metadata for the archive is not needed anymore since we created # one for each testcase. + logs.step(4, 'cleanup') upload_metadata.key.delete() shell.clear_testcase_directories() diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/__init__.py b/src/clusterfuzz/_internal/bot/tasks/utasks/__init__.py index ff822258f24..9ecb5e869c5 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/__init__.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/__init__.py @@ -115,6 +115,27 @@ def __init__(self, subtask: _Subtask): self.post_utask_main_failure = None # Whether preprocess returned the uworker input. self.preprocess_returned: bool | None = None + self.utask_main_returned: bool | None = None + self.utask_main_failure_reason: str | None = None + stage_name = ( + logs.Stage.MAIN.value + if subtask == _Subtask.UWORKER_MAIN else subtask.value) + self._stage_tracker = logs.StageSummaryTracker( + stage_name, start_time_ns=self.start_time_ns) + logs.push_stage_tracker(self._stage_tracker) + + def record_utask_main_output(self, uworker_output) -> None: + """Records outcome and failure details from `utask_main` output.""" + if uworker_output is None: + self.utask_main_returned = False + return + self.utask_main_returned = True + if uworker_output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + self.utask_main_failure = uworker_output.error_type + error_message = getattr(uworker_output, 'error_message', None) + if error_message: + self.utask_main_failure_reason = ( + str(error_message).strip().splitlines()[0][:500]) def set_task_details( self, @@ -222,13 +243,32 @@ def _emit_event_on_exit(self, exc_type): def __exit__(self, _exc_type, _exc_value, _traceback): # Ignore exception details, let Python continue unwinding the stack. + now = time.time_ns() + exc_reason = None + if _exc_type is not None: + lines = str(_exc_value).strip().splitlines() if _exc_value else [] + msg = lines[0] if lines else '' + exc_reason = f'{_exc_type.__name__}: {msg}' if msg else _exc_type.__name__ + if (self._stage_tracker.active_step is not None and + not self._stage_tracker.active_step.finished): + self._stage_tracker.active_step.fail(exc_reason) if self._labels is None: # `set_task_details()` was not called, we are missing information. + try: + if _exc_type is not None or self._stage_tracker.steps: + exc_error_type = ( + _exc_type.__name__ if _exc_type is not None else None) + logs.emit_stage_summary( + self._stage_tracker, + end_time_ns=now, + failed=bool(_exc_type), + error_type=exc_error_type, + failure_reason=exc_reason) + finally: + logs.pop_stage_tracker(self._stage_tracker) return - now = time.time_ns() - duration_secs = (now - self.start_time_ns) / 10**9 monitoring_metrics.UTASK_SUBTASK_DURATION_SECS.add(duration_secs, self._labels) @@ -268,6 +308,45 @@ def __exit__(self, _exc_type, _exc_value, _traceback): monitoring_metrics.TASK_OUTCOME_COUNT_BY_ERROR_TYPE.increment( trimmed_labels) + # In `stage_summary` telemetry, `terminal=True` marks a legitimate early + # completion of the current stage (where remaining steps in the stage are + # intentionally skipped and any non-NO_ERROR `ErrorType` returned from + # `utask_main` is a benign outcome such as `ANALYZE_NO_CRASH`). + preprocess_failed = ( + self._subtask == _Subtask.PREPROCESS and + self.preprocess_returned is False and not self._stage_tracker.terminal) + main_failed = ( + self._subtask == _Subtask.UWORKER_MAIN and + not self._stage_tracker.terminal and + (self.utask_main_returned is False or + self.utask_main_failure is not None)) + stage_failed = bool(_exc_type) or preprocess_failed or main_failed + stage_error_type = None + if preprocess_failed: + stage_error_type = events.TaskOutcome.PREPROCESS_NO_RETURN + elif self.utask_main_failure is not None: + stage_error_type = uworker_msg_pb2.ErrorType.Name( # pylint: disable=no-member + self.utask_main_failure) + elif error_condition != 'N/A': + stage_error_type = error_condition + + main_error_type = None + if (self.post_utask_main_failure is not None and + self.post_utask_main_failure != uworker_msg_pb2.ErrorType.NO_ERROR): # pylint: disable=no-member + main_error_type = uworker_msg_pb2.ErrorType.Name( # pylint: disable=no-member + self.post_utask_main_failure) + + try: + logs.emit_stage_summary( + self._stage_tracker, + end_time_ns=now, + failed=stage_failed, + error_type=stage_error_type, + main_error_type=main_error_type, + failure_reason=exc_reason or self.utask_main_failure_reason) + finally: + logs.pop_stage_tracker(self._stage_tracker) + def ensure_uworker_env_type_safety(uworker_env): """Converts all values in |uworker_env| to str types. @@ -302,7 +381,8 @@ def _preprocess(utask_module, task_argument, job_type, uworker_env, uworker_env) if not uworker_input: recorder.preprocess_returned = False - logs.error('No uworker_input returned from preprocess') + if not recorder._stage_tracker.terminal: # pylint: disable=protected-access + logs.error('No uworker_input returned from preprocess') return None recorder.preprocess_returned = True logs.info('Preprocess finished.') @@ -361,12 +441,11 @@ def uworker_main_no_io(utask_module, serialized_uworker_input): uworker_input.preprocess_start_time) uworker_output = utask_module.utask_main(uworker_input) + recorder.record_utask_main_output(uworker_output) if uworker_output is None: return None # NOTE: Keep this in sync with `uworker_main()`. - if uworker_output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member - recorder.utask_main_failure = uworker_output.error_type uworker_output.bot_name = environment.get_value('BOT_NAME', '') uworker_output.platform_id = environment.get_platform_id() @@ -468,9 +547,7 @@ def uworker_main(input_download_url) -> None: logs.info('Starting utask_main: %s.' % utask_module) uworker_output = utask_module.utask_main(uworker_input) - - if uworker_output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member - recorder.utask_main_failure = uworker_output.error_type + recorder.record_utask_main_output(uworker_output) # NOTE: Keep this in sync with `uworker_main_no_io()`. uworker_output.bot_name = environment.get_value('BOT_NAME', '') diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/analyze_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/analyze_task.py index 9ed661a24ce..7132b2c121f 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/analyze_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/analyze_task.py @@ -304,6 +304,7 @@ def update_testcase_after_crash(testcase, state, job_type, http_flag, def _utask_preprocess(testcase_id, job_type, uworker_env): """Runs preprocessing for analyze task.""" + step = logs.step(1, 'load_testcase') # Get the testcase from the database and mark it as started. testcase = data_handler.get_testcase_by_id(testcase_id) data_handler.update_testcase_comment(testcase, data_types.TaskState.STARTED) @@ -312,6 +313,7 @@ def _utask_preprocess(testcase_id, job_type, uworker_env): testcase_id) if not testcase_upload_metadata: logs.error('Testcase %s has no associated upload metadata.' % testcase_id) + step.fail('Testcase has no associated upload metadata') testcase.key.delete() return None @@ -326,6 +328,7 @@ def _utask_preprocess(testcase_id, job_type, uworker_env): int(testcase_id), testcase_utils.TESTCASE_TRIAGE_DURATION_ANALYZE_LAUNCHED_STEP) + logs.step(2, 'prepare_env') initialize_testcase_for_main(testcase, job_type) setup_input = setup.preprocess_setup_testcase(testcase, uworker_env) @@ -380,6 +383,7 @@ def _build_task_output( def _utask_main(uworker_input): """Executes the untrusted part of analyze_task.""" + logs.step(1, 'setup_build') testcase_upload_metadata = uworker_io.entity_from_protobuf( uworker_input.testcase_upload_metadata, data_types.TestcaseUploadMetadata) testcase = uworker_io.entity_from_protobuf(uworker_input.testcase, @@ -406,6 +410,7 @@ def _utask_main(uworker_input): analyze_task_output = _build_task_output(testcase) # Initialize some variables. + step = logs.step(2, 'test_for_crash') test_timeout = environment.get_value('TEST_TIMEOUT') result, http_flag = test_for_crash_with_retries( fuzz_target, testcase, testcase_file_path, test_timeout) @@ -442,6 +447,7 @@ def _utask_main(uworker_input): 'reproducible': False, 'platform': environment.platform(), }) + step.complete_stage('Testcase did not crash') return uworker_msg_pb2.Output( # pylint: disable=no-member analyze_task_output=analyze_task_output, error_type=uworker_msg_pb2.ErrorType.ANALYZE_NO_CRASH, # pylint: disable=no-member @@ -452,10 +458,12 @@ def _utask_main(uworker_input): # See if we have to ignore this crash. if crash_analyzer.ignore_stacktrace(state.crash_stacktrace): + step.complete_stage('Ignored stacktrace') return uworker_msg_pb2.Output( # pylint: disable=no-member analyze_task_output=analyze_task_output, error_type=uworker_msg_pb2.ErrorType.ANALYZE_CLOSE_INVALID_UPLOADED) # pylint: disable=no-member + logs.step(3, 'check_reproducibility') test_for_reproducibility(fuzz_target, testcase, testcase_file_path, state, test_timeout) @@ -470,6 +478,7 @@ def _utask_main(uworker_input): 'platform': environment.platform(), }) + logs.step(4, 'package_output') fuzz_target_metadata = engine_common.get_fuzz_target_issue_metadata( fuzz_target) @@ -586,11 +595,17 @@ def _update_testcase(output): def _utask_postprocess(output): """Trusted: Cleans up after a uworker execute_task, writing anything needed to the db.""" + step = logs.step(1, 'update_testcase') testcase_utils.emit_testcase_triage_duration_metric( int(output.uworker_input.testcase_id), testcase_utils.TESTCASE_TRIAGE_DURATION_ANALYZE_COMPLETED_STEP) _update_testcase(output) if output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + is_benign = output.error_type in ( + uworker_msg_pb2.ErrorType.ANALYZE_NO_CRASH, # pylint: disable=no-member + uworker_msg_pb2.ErrorType.ANALYZE_CLOSE_INVALID_UPLOADED, # pylint: disable=no-member + ) + step.complete_stage(output, skip=not is_benign) _ERROR_HANDLER.handle(output) return @@ -617,6 +632,7 @@ def _utask_postprocess(output): # For testcase uploaded by bots (with quiet flag), don't create additional # tasks. if testcase_upload_metadata.quiet_flag: + step.complete_stage('Duplicate') data_handler.close_invalid_uploaded_testcase( testcase, testcase_upload_metadata, 'Duplicate') return @@ -645,6 +661,7 @@ def _utask_postprocess(output): testcase_upload_metadata.put() _add_default_issue_metadata(testcase, json.loads(output.issue_metadata)) + logs.step(2, 'create_post_analyze_tasks') logs.info('Creating post-analyze tasks.') # Create tasks to diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py index 58bfdfd8c31..d1af534c888 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py @@ -688,6 +688,7 @@ def do_corpus_pruning(context, revision) -> CorpusPruningResult: """Run corpus pruning.""" # Set |FUZZ_TARGET| environment variable to help with unarchiving only fuzz # target and its related files. + logs.step(2, 'fetch_build') environment.set_value('FUZZ_TARGET', context.fuzz_target.binary) if not build_manager.setup_build( @@ -698,6 +699,7 @@ def do_corpus_pruning(context, revision) -> CorpusPruningResult: pruner, runner = _get_pruner_and_runner(context) fuzzer_binary_name = os.path.basename(runner.target_path) + logs.step(3, 'sync_and_restore_quarantine') logs.info('Getting the initial corpus to process from GCS.') context.sync_to_disk() initial_corpus_size = shell.get_directory_file_count( @@ -707,10 +709,12 @@ def do_corpus_pruning(context, revision) -> CorpusPruningResult: context.restore_quarantined_units() # Shrink to a minimized corpus using corpus merge. + logs.step(4, 'merge_corpus') pruner_stats = pruner.run(context.initial_corpus_path, context.minimized_corpus_path, context.bad_units_path) + logs.step(5, 'sync_and_backup_corpus') logs.info('Syncing the minimized corpus back to GCS.') context.sync_to_gcs() @@ -742,6 +746,7 @@ def do_corpus_pruning(context, revision) -> CorpusPruningResult: # Process bad units found during merge. # Mapping of crash state -> CrashInfo + logs.step(6, 'triage_bad_units') crashes = pruner.process_bad_units(context.bad_units_path, context.quarantine_corpus_path) if shell.get_directory_file_count(context.quarantine_corpus_path): @@ -771,9 +776,11 @@ def do_corpus_pruning(context, revision) -> CorpusPruningResult: coverage_info.quarantine_location = context.quarantine_corpus.get_gcs_url() # Calculate remaining time to use for shared corpus merging. + step = logs.step(7, 'cross_pollinate') time_remaining = _get_time_remaining(start_time) if time_remaining <= 0: logs.warning('Not enough time for shared corpus merging.') + step.fail('Not enough time for shared corpus merging') return None cross_pollinator = CrossPollinator(runner) @@ -1064,6 +1071,7 @@ def _utask_main(uworker_input): data_types.FuzzTarget) revision = 0 # Trunk revision + logs.step(1, 'sync_bundle') if not setup.update_fuzzer_and_data_bundles(uworker_input.setup_input): logs.error(f'Failed to set up fuzzer {fuzz_target.engine}.') return uworker_msg_pb2.Output( # pylint: disable=no-member @@ -1181,6 +1189,7 @@ def _should_use_threaded_ops(fuzzer_name: str) -> bool: def _utask_preprocess(fuzzer_name, job_type, uworker_env): """Runs preprocessing for corpus pruning task.""" + step = logs.step(1, 'check_last_status') fuzz_target = data_handler.get_fuzz_target(fuzzer_name) task_name = f'corpus_pruning_{fuzzer_name}_{job_type}' @@ -1196,6 +1205,8 @@ def _utask_preprocess(fuzzer_name, job_type, uworker_env): if not data_handler.update_task_status(task_name, data_types.TaskState.STARTED): logs.info('A previous corpus pruning task is still running, exiting.') + step.complete_stage( + 'A previous corpus pruning task is still running', skip=True) return None setup_input = ( @@ -1209,6 +1220,7 @@ def _utask_preprocess(fuzzer_name, job_type, uworker_env): # to prevent corpus from growing unbounded and recurring failures when trying # to minimize it. logs.info('checking last') + logs.step(2, 'sign_urls') if last_execution_failed: logs.info('last') # TODO(metzman): Is this too expensive to do in preprocess? @@ -1290,7 +1302,9 @@ def _update_latest_backup(output): def _utask_postprocess(output): """Trusted: Handles errors and writes anything needed to the db.""" + step = logs.step(1, 'save_coverage_and_crashes') if output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + step.complete_stage(output, skip=True) _ERROR_HANDLER.handle(output) return task_name = (f'corpus_pruning_{output.uworker_input.fuzzer_name}_' diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py index 67169b2e836..d14a16eb059 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py @@ -1770,11 +1770,13 @@ def do_engine_fuzzing(self, engine_impl): self.fuzz_target.fully_qualified_name()) # Synchronize corpus files with GCS + logs.step(4, 'sync_corpus') sync_corpus_directory = builtin.get_corpus_directory( self.data_directory, self.fuzz_target.project_qualified_name()) self.sync_corpus(sync_corpus_directory) # Reset memory tool options. + step = logs.step(5, 'fuzz_and_upload_corpus') environment.reset_current_memory_tool_options( redzone_size=self.redzone, disable_ubsan=self.disable_ubsan) @@ -1802,6 +1804,7 @@ def do_engine_fuzzing(self, engine_impl): # recorded and hopefully fuzzed instead. The old targets will eventually # be garbage collected. Log this as an error to keep an eye on it. logs.error(f'{self.fuzz_target.binary} is not in the build.') + step.skip(f'{self.fuzz_target.binary} is not in the build.') return [], {} fuzzer_metadata.update(cur_fuzzer_metadata) @@ -1923,6 +1926,7 @@ def _add_crashes_from_temp_queue( def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type): """Run blackbox fuzzing. Currently also used for engine fuzzing.""" + logs.step(4, 'generate_inputs') _TrackFuzzTime.reset_session(self.fully_qualified_fuzzer_name, job_type) # Set the thread timeout values. @@ -1962,6 +1966,7 @@ def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type): self._emit_testcase_generation_time_metric( testcase_generation_start, testcase_count, fuzzer.name, job_type) + logs.step(5, 'execute_inputs') environment.set_value('FUZZER_NAME', self.fully_qualified_fuzzer_name) # Initialize a list of crashes. @@ -2116,6 +2121,9 @@ def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type): def run(self): """Run the fuzzing session.""" start_time = time.time() + is_engine = bool(engine.get(self.fuzzer_name)) + logs.set_stage_variant('cg' if is_engine else 'bb') + logs.step(1, 'sync_bundle') # Update LSAN local blacklist with global blacklist. global_blacklisted_functions = ( self.uworker_input.fuzz_task_input.global_blacklisted_functions) @@ -2146,6 +2154,7 @@ def run(self): self.test_timeout = set_test_timeout(self.fuzzer.timeout, self.timeout_multiplier) + step = logs.step(2, 'fetch_build') self.testcase_directory = environment.get_value('FUZZ_INPUTS') fuzz_target = self.fuzz_target.binary if self.fuzz_target else None @@ -2162,6 +2171,8 @@ def run(self): if not self.fuzz_task_output.fuzz_targets: logs.error('No fuzz targets.') if not self.fuzz_target: + step.fail('No fuzz targets.' if not self.fuzz_task_output. + fuzz_targets else 'No fuzz target selected.') return uworker_msg_pb2.Output( # pylint: disable=no-member fuzz_task_output=self.fuzz_task_output, error_type=uworker_msg_pb2.ErrorType.FUZZ_NO_FUZZ_TARGET_SELECTED) # pylint: disable=no-member @@ -2174,6 +2185,7 @@ def run(self): # Check if we have a bad build, i.e. one that crashes on startup. # If yes, bail out. + logs.step(3, 'check_bad_build') logs.info('Checking for bad build.') crash_revision = environment.get_value('APP_REVISION') @@ -2214,6 +2226,7 @@ def run(self): return uworker_msg_pb2.Output( # pylint: disable=no-member error_type=uworker_msg_pb2.ErrorType.UNHANDLED) # pylint: disable=no-member + logs.step(6, 'process_crashes') logs.info('Finished processing test cases.') # For Android, bring back device to a good state before analyzing crashes. @@ -2301,6 +2314,7 @@ def postprocess(self, uworker_output): logs.info('postprocess: fuzz_task_output.fully_qualified_fuzzer_name ' f'{fuzz_task_output.fully_qualified_fuzzer_name}') uworker_input = uworker_output.uworker_input + logs.step(2, 'dedup_and_file_bugs') postprocess_process_crashes(uworker_input, uworker_output) postprocess_sample_testcases(uworker_input, uworker_output) if not environment.is_engine_fuzzer_job(): @@ -2384,9 +2398,10 @@ def utask_main(uworker_input): def handle_fuzz_no_fuzz_target_selected(output): save_fuzz_targets(output) # Try again now that there are some fuzz targets. - utask_preprocess(output.uworker_input.fuzzer_name, - output.uworker_input.job_type, - output.uworker_input.uworker_env) + with logs.stage_summary_context(logs.Stage.PREPROCESS, emit_log=False): + utask_preprocess(output.uworker_input.fuzzer_name, + output.uworker_input.job_type, + output.uworker_input.uworker_env) def _make_session(uworker_input): @@ -2452,6 +2467,7 @@ def _utask_preprocess(fuzzer_name, job_type, uworker_env): uworker_env) fuzz_target = _preprocess_get_fuzz_target(fuzzer_name, job_type) fuzz_task_input = uworker_msg_pb2.FuzzTaskInput() # pylint: disable=no-member + logs.step(2, 'sign_urls') if fuzz_target: # Add the chosen fuzz target to logs context. logs.log_contexts.add_metadata('fuzz_target', fuzz_target.binary) @@ -2499,11 +2515,13 @@ def utask_preprocess(fuzzer_name, job_type, uworker_env): """Set logs context and preprocess untrusted task.""" # Delay adding the fuzz target to logs context until it is chosen in # preprocess. + step = logs.step(1, 'select_target') with logs.fuzzer_log_context(fuzzer_name, job_type, fuzz_target=None): try: return _utask_preprocess(fuzzer_name, job_type, uworker_env) except errors.InvalidFuzzerError: logs.error('Fuzzer %s is invalid or no longer exists.' % fuzzer_name) + step.fail(f'Fuzzer {fuzzer_name} is invalid or no longer exists.') return None @@ -2568,7 +2586,9 @@ def _upload_fuzzer_run_output(fuzzer_run_output, fuzzer_name): def _utask_postprocess(output): """Postprocesses fuzz_task.""" + step = logs.step(1, 'record_stats') if output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + step.complete_stage(output, skip=True) _ERROR_HANDLER.handle(output) return @@ -2580,6 +2600,7 @@ def _utask_postprocess(output): session.postprocess(output) # TODO(b/374776013): Refactor this code so the uploads happen during # utask_main. + logs.step(3, 'upload_logs') fuzzer_name = output.fuzz_task_output.fully_qualified_fuzzer_name for fuzzer_run_output in output.fuzz_task_output.fuzzer_run_outputs: _upload_fuzzer_run_output(fuzzer_run_output, fuzzer_name) diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/minimize_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/minimize_task.py index 67e7f04355f..9c5cf48c2c0 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/minimize_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/minimize_task.py @@ -372,10 +372,12 @@ def _get_minimize_task_input(testcase): def utask_preprocess(testcase_id, job_type, uworker_env): """Preprocess in a trusted bot.""" # Locate the testcase associated with the id. + step = logs.step(1, 'load_testcase') try: testcase = data_handler.get_testcase_by_id(testcase_id) - except errors.InvalidTestcaseError: + except errors.InvalidTestcaseError as e: logs.warning(f'Testcase {testcase_id} no longer exists.') + step.fail(e) return None with logs.testcase_log_context(testcase, testcase.get_fuzz_target()): # Allow setting up a different fuzzer. @@ -387,9 +389,11 @@ def utask_preprocess(testcase_id, job_type, uworker_env): if not environment.is_minimization_supported(): # TODO(ochang): More robust check for engine minimization support. _skip_minimization(testcase, 'Engine does not support minimization.') + step.complete_stage('Engine does not support minimization.', skip=True) return None # Update comments to reflect bot information. + logs.step(2, 'prepare_env') data_handler.update_testcase_comment(testcase, data_types.TaskState.STARTED) uworker_input = uworker_msg_pb2.Input( # pylint: disable=no-member @@ -409,6 +413,10 @@ def utask_main(uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-memb data_types.Testcase) with logs.testcase_log_context( testcase, testcase_manager.get_fuzz_target_from_input(uworker_input)): + job_name = uworker_input.job_type or environment.get_value('JOB_NAME') + logs.set_stage_variant( + 'lf' if job_name and environment.is_libfuzzer_job(job_name) else 'bb') + logs.step(1, 'setup_build') uworker_io.check_handling_testcase_safe(testcase) minimize_task_input = uworker_input.minimize_task_input # Setup testcase and its dependencies. @@ -448,6 +456,7 @@ def utask_main(uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-memb ' should have been detected in preprocess.') return None + logs.step(2, 'warmup_and_verify') # TODO(alhijazi): re-install multithreaded runs # max_threads = utils.maximum_parallel_processes_allowed() # Temporarily run minimization single threaded. @@ -536,6 +545,7 @@ def utask_main(uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-memb logs.info(f'Starting minimization with overall {deadline}s timeout.') + step = logs.step(3, 'minimize_gestures') if should_attempt_phase(testcase, MinimizationPhase.GESTURES): logs.info('Starting gesture minimization phase.') gestures = minimize_gestures(test_runner, testcase) @@ -568,8 +578,11 @@ def utask_main(uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-memb error_message='Timed out during gesture minimization.') logs.info('Minimized gestures.') + else: + step.skip('Phase skipped') # Minimize the main file. + step = logs.step(4, 'minimize_main_file') data = utils.get_file_contents_with_fatal_error_on_failure( testcase_file_path) if should_attempt_phase(testcase, MinimizationPhase.MAIN_FILE): @@ -589,10 +602,15 @@ def utask_main(uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-memb testcase.set_metadata('minimization_phase', MinimizationPhase.FILE_LIST, False) minimize_task_output.minimization_phase = MinimizationPhase.FILE_LIST + else: + step.skip('Phase skipped') # Minimize the file list. + step = logs.step(5, 'minimize_resources') + attempted_resources = False if should_attempt_phase(testcase, MinimizationPhase.FILE_LIST): if environment.get_value('MINIMIZE_FILE_LIST', True): + attempted_resources = True logs.info('Starting file list minimization phase.') file_list = minimize_file_list(test_runner, file_list, input_directory, testcase_file_path) @@ -616,6 +634,7 @@ def utask_main(uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-memb # Minimize any files remaining in the file list. if should_attempt_phase(testcase, MinimizationPhase.RESOURCES): if environment.get_value('MINIMIZE_RESOURCES', True): + attempted_resources = True logs.info('Starting resources minimization phase.') for dependency in file_list: minimize_resource(test_runner, dependency, input_directory, @@ -637,7 +656,10 @@ def utask_main(uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-memb testcase.set_metadata('minimization_phase', MinimizationPhase.ARGUMENTS, False) minimize_task_output.minimization_phase = MinimizationPhase.ARGUMENTS + if not attempted_resources: + step.skip('Phase skipped') + step = logs.step(6, 'minimize_arguments') if should_attempt_phase(testcase, MinimizationPhase.ARGUMENTS): logs.info('Starting arguments minimization phase.') app_arguments = minimize_arguments(test_runner, app_arguments) @@ -656,7 +678,10 @@ def utask_main(uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-memb minimize_task_output=minimize_task_output) logs.info('Minimized arguments.') + else: + step.skip('Phase skipped') + logs.step(7, 'store_minimized') logs.info('Finished minization.') command = testcase_manager.get_command_line_for_application( @@ -883,11 +908,13 @@ def finalize_testcase(testcase_id, last_crash_result_dict, flaky_stack=False): def utask_postprocess(output): """Postprocess in a trusted bot.""" # Retrive the testcase early for logs context. + step = logs.step(1, 'update_testcase') try: testcase = data_handler.get_testcase_by_id(output.uworker_input.testcase_id) except errors.InvalidTestcaseError: logs.warning( f'Testcase {output.uworker_input.testcase_id} no longer exists.') + step.skip('Testcase no longer exists') return with logs.testcase_log_context(testcase, testcase.get_fuzz_target()): testcase_utils.emit_testcase_triage_duration_metric( @@ -896,9 +923,11 @@ def utask_postprocess(output): update_testcase(output) _cleanup_unused_blobs_from_storage(output) if output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + step.complete_stage(output, skip=True) _ERROR_HANDLER.handle(output) return + logs.step(2, 'finalize_and_fan_out') finalize_testcase( output.uworker_input.testcase_id, output.minimize_task_output.last_crash_result_dict, @@ -1600,10 +1629,12 @@ def do_libfuzzer_minimization( testcase: data_types.Testcase, testcase_file_path: str) -> uworker_msg_pb2.Output: # pylint: disable=no-member """Use libFuzzer's built-in minimizer where appropriate.""" + logs.set_stage_variant('lf') timeout = environment.get_value('LIBFUZZER_MINIMIZATION_TIMEOUT', 600) rounds = environment.get_value('LIBFUZZER_MINIMIZATION_ROUNDS', 5) # Get initial crash state. + logs.step(2, 'verify_initial_crash') initial_crash_result = _run_libfuzzer_testcase( fuzz_target, testcase, testcase_file_path, crash_retries=None) # Use default retries. @@ -1622,6 +1653,7 @@ def do_libfuzzer_minimization( logs.info(f'Initial crash state: {expected_state.crash_state}\n') # Minimize *_OPTIONS env variable first. + logs.step(3, 'minimize_sanitizer_options') env = {} # A Dict[str, str] potentially containing options_env_var strings and the # corresponding minimized_options_string to be parsed and set in testcase @@ -1672,6 +1704,7 @@ def do_libfuzzer_minimization( if env: testcase.set_metadata('env', env, False) + logs.step(4, 'minimize_rounds') current_testcase_path = testcase_file_path last_crash_result = None last_minimized_keys = None @@ -1710,6 +1743,7 @@ def do_libfuzzer_minimization( logs.info('LibFuzzer minimization succeeded.') + logs.step(5, 'package_output') if utils.is_oss_fuzz(): # Scrub the testcase of non-essential data. cleansed_testcase_path, cleansed_minimized_keys = do_libfuzzer_cleanse( diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/progression_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/progression_task.py index 24cd3bb4790..469120b2b4f 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/progression_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/progression_task.py @@ -258,6 +258,8 @@ def _check_fixed_for_custom_binary(testcase: data_types.Testcase, testcase_file_path: str, uworker_input: uworker_msg_pb2.Input): # pylint: disable=no-member """Simplified fixed check for test cases using custom binaries.""" + logs.step(2, 'fetch_revisions').skip('Custom binary') + logs.step(3, 'setup_trunk_build') build_setup_result = build_manager.setup_build() # 'APP_REVISION' is set during setup_build(). revision = environment.get_value('APP_REVISION') @@ -270,6 +272,7 @@ def _check_fixed_for_custom_binary(testcase: data_types.Testcase, error_message='Build setup failed for custom binary', error_type=uworker_msg_pb2.ErrorType.PROGRESSION_BUILD_SETUP_ERROR) # pylint: disable=no-member + logs.step(4, 'test_at_trunk') test_timeout = environment.get_value('TEST_TIMEOUT', 10) fuzz_target = testcase_manager.get_fuzz_target_from_input(uworker_input) result = testcase_manager.test_for_crash_with_retries( @@ -280,6 +283,8 @@ def _check_fixed_for_custom_binary(testcase: data_types.Testcase, http_flag=testcase.http_flag) _log_output(revision, result) + logs.step(5, 'bisect_fixed_range').skip('Custom binary') + logs.step(6, 'package_output') # If this still crashes on the most recent build, it's not fixed. The task # will be rescheduled by a cron job and re-attempted eventually. if result.is_crash(): @@ -325,12 +330,18 @@ def _update_issue_metadata(testcase: data_types.Testcase, def _testcase_reproduces_in_revision( - testcase: data_types.Testcase, testcase_file_path: str, job_type: str, - revision: int, fuzz_target: data_types.FuzzTarget | None, - progression_task_output: uworker_msg_pb2.ProgressionTaskOutput): # pylint: disable=no-member + testcase: data_types.Testcase, + testcase_file_path: str, + job_type: str, + revision: int, + fuzz_target: data_types.FuzzTarget | None, + progression_task_output: uworker_msg_pb2.ProgressionTaskOutput, # pylint: disable=no-member + record_trunk_steps: bool = False): """Tests to see if a test case reproduces in the specified revision. Returns a tuple containing the (result, error) depending on whether there was an error.""" + if record_trunk_steps: + logs.step(3, 'setup_trunk_build') try: fuzz_target_binary = fuzz_target.binary if fuzz_target else None build_setup_result = build_manager.setup_build( @@ -361,6 +372,8 @@ def _testcase_reproduces_in_revision( error_message=error_message, error_type=uworker_msg_pb2.ErrorType.PROGRESSION_BAD_BUILD) # pylint: disable=no-member + if record_trunk_steps: + logs.step(4, 'test_at_trunk') test_timeout = environment.get_value('TEST_TIMEOUT', 10) result = testcase_manager.test_for_crash_with_retries( fuzz_target, @@ -450,12 +463,16 @@ def _set_regression_testcase_upload_url( def utask_preprocess(testcase_id, job_type, uworker_env): """Runs preprocessing for progression task.""" + step = logs.step(1, 'load_testcase') testcase = data_handler.get_testcase_by_id(testcase_id) with logs.testcase_log_context(testcase, testcase.get_fuzz_target()): if not testcase: + step.fail('Testcase not found') return None if testcase.fixed: logs.error(f'Fixed range is already set as {testcase.fixed}, skip.') + step.complete_stage( + f'Fixed range is already set as {testcase.fixed}', skip=True) return None # Set a flag to indicate we are running progression task. This shows pending @@ -463,6 +480,7 @@ def utask_preprocess(testcase_id, job_type, uworker_env): # triage cron. testcase.set_metadata('progression_pending', True) data_handler.update_testcase_comment(testcase, data_types.TaskState.STARTED) + logs.step(2, 'prepare_env') blob_name, blob_upload_url = blobs.get_blob_signed_upload_url() progression_input = uworker_msg_pb2.ProgressionTaskInput( # pylint: disable=no-member custom_binary=build_manager.is_custom_binary(), @@ -495,6 +513,7 @@ def find_fixed_range(uworker_input): setup_input = uworker_input.setup_input fuzz_target = testcase_manager.get_fuzz_target_from_input(uworker_input) + logs.step(1, 'setup_testcase') _, testcase_file_path, error = setup.setup_testcase(testcase, job_type, setup_input) if error: @@ -505,6 +524,7 @@ def find_fixed_range(uworker_input): return _check_fixed_for_custom_binary(testcase, testcase_file_path, uworker_input) + logs.step(2, 'fetch_revisions') build_bucket_path = build_manager.get_primary_bucket_path() bad_revisions = uworker_input.progression_task_input.bad_revisions @@ -566,8 +586,13 @@ def find_fixed_range(uworker_input): # Check to see if this testcase is still crashing now. If it is, then just # bail out. result, error = _testcase_reproduces_in_revision( - testcase, testcase_file_path, job_type, max_revision, fuzz_target, - progression_task_output) + testcase, + testcase_file_path, + job_type, + max_revision, + fuzz_target, + progression_task_output, + record_trunk_steps=True) issue_metadata = engine_common.get_fuzz_target_issue_metadata(fuzz_target) issue_metadata = issue_metadata or {} @@ -576,6 +601,8 @@ def find_fixed_range(uworker_input): return error if result.is_crash(): + logs.step(5, 'bisect_fixed_range').skip('Still crashes at trunk') + logs.step(6, 'package_output') logs.info(f'Found crash with same signature on latest' f' revision r{max_revision}.') app_path = environment.get_value('APP_PATH') @@ -603,6 +630,7 @@ def find_fixed_range(uworker_input): # Verify that we do crash in the min revision. This is assumed to be true # while we are doing the bisect. + logs.step(5, 'bisect_fixed_range') result, error = _testcase_reproduces_in_revision( testcase, testcase_file_path, job_type, min_revision, fuzz_target, progression_task_output) @@ -629,6 +657,7 @@ def find_fixed_range(uworker_input): # If the min and max revisions are one apart this is as much as we can # narrow the range. if max_index - min_index == 1: + logs.step(6, 'package_output') testcase.open = False _store_testcase_for_regression_testing( testcase, testcase_file_path, uworker_input.progression_task_input) @@ -730,6 +759,7 @@ def utask_main(uworker_input): def utask_postprocess(output: uworker_msg_pb2.Output): # pylint: disable=no-member """Trusted: Cleans up after a uworker execute_task, writing anything needed to the db.""" + step = logs.step(1, 'update_metadata') testcase = data_handler.get_testcase_by_id(output.uworker_input.testcase_id) with logs.testcase_log_context(testcase, testcase.get_fuzz_target()): _maybe_clear_progression_last_min_max_metadata(testcase, output) @@ -745,6 +775,7 @@ def utask_postprocess(output: uworker_msg_pb2.Output): # pylint: disable=no-mem task_output.build_data_list) if output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + step.complete_stage(output, skip=True) _ERROR_HANDLER.handle(output) return @@ -752,6 +783,7 @@ def utask_postprocess(output: uworker_msg_pb2.Output): # pylint: disable=no-mem # Both regression and fixed ranges are requested once. Regression is also # requested here as the bisection service may require details that # are not yet available (e.g. issue ID) at the time regress_task completes. + logs.step(2, 'save_fixed_or_open') bisection.request_bisection(testcase) if task_output and task_output.crash_on_latest: diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/regression_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/regression_task.py index 32c685254f3..f3a95695c0d 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/regression_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/regression_task.py @@ -361,6 +361,7 @@ def validate_regression_range( ) -> Optional[uworker_msg_pb2.Output]: # pylint: disable=no-member """Ensure that we found the correct min revision by testing earlier ones. Returns a uworker_msg_pb2.Output in case of error or crash, None otherwise.""" + logs.step(6, 'validate_range') earlier_revisions = revision_list[ min_index - EARLIER_REVISIONS_TO_CONSIDER_FOR_VALIDATION:min_index] revision_count = min(len(earlier_revisions), REVISIONS_TO_TEST_FOR_VALIDATION) @@ -400,6 +401,7 @@ def find_regression_range( fuzz_target = testcase_manager.get_fuzz_target_from_input(uworker_input) # Setup testcase and its dependencies. + logs.step(1, 'setup_testcase') _, testcase_file_path, error = setup.setup_testcase(testcase, job_type, uworker_input.setup_input) if error: @@ -408,6 +410,7 @@ def find_regression_range( # Help the type checker. assert isinstance(testcase_file_path, str) + logs.step(2, 'fetch_revisions') build_bucket_path = build_manager.get_primary_bucket_path() revision_list = build_manager.get_revisions_list( build_bucket_path, @@ -459,6 +462,7 @@ def find_regression_range( max_revision = revision_list[max_index] # Might be > `known_crash_revision`. # Check invariant: max revision crashes. + logs.step(3, 'reproduce_at_max') regression_task_output = uworker_msg_pb2.RegressionTaskOutput() # pylint: disable=no-member crashes_in_max_revision, error = _testcase_reproduces_in_revision( testcase, @@ -487,6 +491,7 @@ def find_regression_range( regression_task_output.is_testcase_reproducible = True regression_task_output.last_regression_max = max_revision + step = logs.step(4, 'find_min_revision') min_index = None if min_revision: min_index = revisions.find_min_revision_index(revision_list, min_revision) @@ -505,12 +510,19 @@ def find_regression_range( if output: # Either we encountered an error, or there is no good revision and the # regression range is `0:revision_list[0]`. + if output.error_type == uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + step.complete_stage('Earliest build crashes') + else: + step.fail(output) return output + else: + step.skip('Resuming from existing min_revision') # Type checker cannot figure this out. assert isinstance(min_index, int) assert isinstance(max_index, int) + logs.step(5, 'bisect_range') while time.time() < deadline: min_revision = revision_list[min_index] max_revision = revision_list[max_index] @@ -591,11 +603,18 @@ def utask_preprocess(testcase_id: str, job_type: str, Runs on a trusted worker. """ + step = logs.step(1, 'load_testcase') testcase = data_handler.get_testcase_by_id(testcase_id) + if not testcase: + step.fail('Testcase not found') + return None with logs.testcase_log_context(testcase, testcase.get_fuzz_target()): if testcase.regression: logs.error( f'Regression range is already set as {testcase.regression}, skip.') + step.complete_stage( + f'Regression range is already set as {testcase.regression}', + skip=True) return None # This task is not applicable for custom binaries. @@ -604,10 +623,12 @@ def utask_preprocess(testcase_id: str, job_type: str, data_handler.update_testcase_comment( testcase, data_types.TaskState.ERROR, 'Not applicable for custom binaries') + step.complete_stage('Not applicable for custom binaries', skip=True) return None data_handler.update_testcase_comment(testcase, data_types.TaskState.STARTED) + logs.step(2, 'prepare_env') setup_input = setup.preprocess_setup_testcase(testcase, uworker_env) task_input = uworker_msg_pb2.RegressionTaskInput( # pylint: disable=no-member @@ -729,6 +750,7 @@ def utask_postprocess(output: uworker_msg_pb2.Output) -> None: # pylint: disabl Runs on a trusted worker. """ + step = logs.step(1, 'update_metadata') testcase_id = output.uworker_input.testcase_id # Retrieve the testcase early to be used by logs context. testcase = data_handler.get_testcase_by_id(testcase_id) @@ -751,9 +773,11 @@ def utask_postprocess(output: uworker_msg_pb2.Output) -> None: # pylint: disabl False) if output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + step.complete_stage(output, skip=True) _ERROR_HANDLER.handle(output) return + logs.step(2, 'save_regression_range') save_regression_range(output) diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/symbolize_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/symbolize_task.py index de49ba3c043..d854fc156de 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/symbolize_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/symbolize_task.py @@ -51,16 +51,19 @@ def handle_build_setup_error(output): def _utask_preprocess(testcase_id, job_type, uworker_env): """Run preprocessing for symbolize task.""" + step = logs.step(1, 'load_testcase') # Locate the testcase associated with the id. testcase = data_handler.get_testcase_by_id(testcase_id) # We should atleast have a symbolized debug or release build. if not build_manager.has_symbolized_builds(): + step.complete_stage('Job has no symbolized builds', skip=True) return None data_handler.update_testcase_comment(testcase, data_types.TaskState.STARTED) # Setup testcase and its dependencies. + logs.step(2, 'prepare_env') setup_input = setup.preprocess_setup_testcase(testcase, uworker_env) old_crash_stacktrace = data_handler.get_stacktrace(testcase) @@ -91,6 +94,7 @@ def _utask_main(uworker_input): job_type = uworker_input.job_type setup_input = uworker_input.setup_input + logs.step(1, 'setup_release_build') _, testcase_file_path, error = setup.setup_testcase(testcase, job_type, setup_input) if error: @@ -129,6 +133,7 @@ def _utask_main(uworker_input): # ASAN tool settings (if the tool is used). # See if we can get better stacks with higher redzone sizes. # A UAF might actually turn out to be OOB read/write with a bigger redzone. + step = logs.step(2, 'search_asan_redzone') if environment.tool_matches('ASAN', job_type) and testcase.security_flag: redzone = MAX_REDZONE while redzone >= MIN_REDZONE: @@ -180,9 +185,12 @@ def _utask_main(uworker_input): break redzone /= 2 + else: + step.skip('Not ASAN security testcase') # We no longer need this build, delete it to save some disk space. We will # download a symbolized release build to perform the symbolization. + logs.step(3, 'setup_symbolized_build') build.delete() # We should have atleast a symbolized debug or a release build. @@ -195,6 +203,7 @@ def _utask_main(uworker_input): error_type=uworker_msg_pb2.ErrorType.SYMBOLIZE_BUILD_SETUP_ERROR) # pylint: disable=no-member # Increase malloc_context_size to get all stack frames. Default is 30. + step = logs.step(4, 'reproduce_symbolized') environment.reset_current_memory_tool_options( redzone_size=sym_redzone, malloc_context_size=STACK_FRAME_COUNT, @@ -209,7 +218,10 @@ def _utask_main(uworker_input): result, sym_crash_stacktrace = ( get_symbolized_stacktraces(testcase_file_path, testcase, old_crash_stacktrace, sym_crash_state)) + if not result: + step.skip('Unable to reproduce crash on symbolized build') + logs.step(5, 'package_output') symbolize_task_output = uworker_msg_pb2.SymbolizeTaskOutput( # pylint: disable=no-member crash_type=testcase.crash_type, crash_address=sym_crash_address, @@ -338,7 +350,9 @@ def get_symbolized_stacktraces(testcase_file_path, testcase, def _utask_postprocess(output): """Handle the output from utask_main.""" + step = logs.step(1, 'save_symbolized_stack') if output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + step.complete_stage(output, skip=True) _ERROR_HANDLER.handle(output) return diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/variant_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/variant_task.py index 44e9b8c9aaf..d9557bcf4b0 100644 --- a/src/clusterfuzz/_internal/bot/tasks/utasks/variant_task.py +++ b/src/clusterfuzz/_internal/bot/tasks/utasks/variant_task.py @@ -59,10 +59,12 @@ def _get_variant_testcase_for_job(testcase, job_type): def utask_preprocess(testcase_id, job_type, uworker_env): """Run a test case with a different job type to see if they reproduce.""" + step = logs.step(1, 'clone_testcase_and_prepare') try: testcase = data_handler.get_testcase_by_id(testcase_id) - except errors.InvalidTestcaseError: + except errors.InvalidTestcaseError as e: logs.warning(f'Testcase {testcase_id} no longer exists.') + step.fail(e) return None with logs.testcase_log_context(testcase, testcase.get_fuzz_target()): @@ -70,6 +72,7 @@ def utask_preprocess(testcase_id, job_type, uworker_env): environment.is_engine_fuzzer_job(job_type)): # We should never reach here. But in case we do, we should bail out as # otherwise we will run into exceptions. + step.complete_stage('Mismatched engine/blackbox job types', skip=True) return None # Use a cloned testcase entity with different fuzz target paramaters for @@ -97,6 +100,7 @@ def utask_preprocess(testcase_id, job_type, uworker_env): logs.warning( f'Fuzz target or fuzzer not found for variant job {job_type}: {error}' ) + step.fail(str(error)) return None return uworker_input @@ -117,6 +121,7 @@ def utask_main(uworker_input): testcase.put = lambda: None # Setup testcase and its dependencies. + logs.step(1, 'setup_build') _, testcase_file_path, error = setup.setup_testcase( testcase, uworker_input.job_type, uworker_input.setup_input) if error: @@ -141,6 +146,7 @@ def utask_main(uworker_input): # Disable gestures if we're running on a different platform from that of # the original test case. + logs.step(2, 'test_for_crash') use_gestures = testcase.platform == environment.platform().lower() # Reproduce the crash. @@ -164,6 +170,7 @@ def utask_main(uworker_input): return uworker_msg_pb2.Output( # pylint: disable=no-member error_type=uworker_msg_pb2.ErrorType.UNHANDLED) # pylint: disable=no-member + step = logs.step(3, 'check_reproducibility') if result.is_crash() and not result.should_ignore(): crash_state = result.get_state() crash_type = result.get_type() @@ -189,6 +196,7 @@ def utask_main(uworker_input): crash_stacktrace_output = utils.get_crash_stacktrace_output( command, symbolized_crash_stacktrace, unsymbolized_crash_stacktrace) else: + step.skip('Testcase did not crash or crash was ignored') status = data_types.TestcaseVariantStatus.UNREPRODUCIBLE is_similar = False crash_type = None @@ -197,6 +205,7 @@ def utask_main(uworker_input): crash_stacktrace_output = 'No crash occurred.' # Regular case of variant analysis. + logs.step(4, 'package_output') variant_task_output = uworker_msg_pb2.VariantTaskOutput() # pylint: disable=no-member variant_task_output.status = status variant_task_output.revision = int(revision) @@ -230,14 +239,17 @@ def handle_build_setup_error(output): def utask_postprocess(output): """Handle the output from utask_main.""" + step = logs.step(1, 'save_variant') try: testcase = data_handler.get_testcase_by_id(output.uworker_input.testcase_id) except errors.InvalidTestcaseError: logs.warning( f'Testcase {output.uworker_input.testcase_id} no longer exists.') + step.skip('Testcase no longer exists') return with logs.testcase_log_context(testcase, testcase.get_fuzz_target()): if output.error_type != uworker_msg_pb2.ErrorType.NO_ERROR: # pylint: disable=no-member + step.complete_stage(output, skip=True) _ERROR_HANDLER.handle(output) return diff --git a/src/clusterfuzz/_internal/metrics/logs.py b/src/clusterfuzz/_internal/metrics/logs.py index f33244e2692..cc697f24ce8 100644 --- a/src/clusterfuzz/_internal/metrics/logs.py +++ b/src/clusterfuzz/_internal/metrics/logs.py @@ -720,6 +720,10 @@ def emit(level, message, exc_info=None, **extras): # Include extras passed as an argument and default extras. all_extras = _default_extras.copy() all_extras.update(extras) + if level >= logging.ERROR and 'stage_summary' not in all_extras: + tracker = _current_stage_tracker() + if tracker is not None: + tracker.record_error(str(message)) path_name, line_number, method_name = get_source_location() @@ -1123,11 +1127,17 @@ def fuzzer_log_context(fuzzer_name: str, job_type: str, log_contexts.add_metadata('fuzz_target', fuzz_target_bin) log_contexts.add_metadata('fuzzer_name', fuzzer_name) log_contexts.add_metadata('job_type', job_type) + tracker = _current_stage_tracker() + if tracker is not None: + tracker.snapshot_log_context() yield except Exception as e: warning(message='Error during fuzzer context.') raise e finally: + tracker = _current_stage_tracker() + if tracker is not None: + tracker.snapshot_log_context() log_contexts.delete_metadata('fuzz_target') log_contexts.delete_metadata('fuzzer_name') log_contexts.delete_metadata('job_type') @@ -1160,6 +1170,9 @@ def testcase_log_context(testcase: 'Testcase | TestcaseAttributes', fuzz_target_bin = 'unknown' if not testcase else testcase.get_metadata( 'fuzzer_binary_name', 'unknown') log_contexts.add_metadata('fuzz_target', fuzz_target_bin) + tracker = _current_stage_tracker() + if tracker is not None: + tracker.snapshot_log_context() yield except Exception as e: # Logging as a warning because this error will be handled @@ -1168,6 +1181,9 @@ def testcase_log_context(testcase: 'Testcase | TestcaseAttributes', warning(message='Error during testcase context.') raise e finally: + tracker = _current_stage_tracker() + if tracker is not None: + tracker.snapshot_log_context() for field in TestcaseLogStruct._fields: log_contexts.delete_metadata(field) log_contexts.delete_metadata('fuzzer_name') @@ -1224,3 +1240,375 @@ def grouper_log_context(testcase_1: 'Testcase | TestcaseAttributes', else: log_contexts.delete_metadata(f'testcase_1_{field}') log_contexts.delete_metadata(f'testcase_2_{field}') + + +_STAGE_TRACKER_LOCAL = threading.local() + + +def _extract_step_reason(reason: Any) -> str | None: + """Extracts a concise human-readable reason string from a step outcome.""" + if reason is None: + return None + if isinstance(reason, str): + cleaned = reason.strip().splitlines()[0] if reason.strip() else '' + return cleaned[:500] or None + + error_message = getattr(reason, 'error_message', None) + if error_message: + cleaned = str(error_message).strip().splitlines()[0] + if cleaned: + return cleaned[:500] + + error_type = getattr(reason, 'error_type', None) + if isinstance(error_type, int) and error_type != 0: + try: + from clusterfuzz._internal.protos import uworker_msg_pb2 + return uworker_msg_pb2.ErrorType.Name(error_type) # pylint: disable=no-member + except Exception: + return str(error_type) + + cleaned = str(reason).strip().splitlines()[0] if str(reason).strip() else '' + return cleaned[:500] or None + + +class _StepHandle: + """Handle for a numbered stage step supporting both flat and `with` usage.""" + + def __init__(self, n: int, name: str): + self.n = n + self.name = name + self.start_time_ns = time.time_ns() + self.status = 'ok' + self.reason: str | None = None + self.duration_s = 0.0 + self.terminal = False + self.tracker: StageSummaryTracker | None = None + self.finished = False + + def skip(self, reason: Any = None) -> None: + """Marks this step as skipped with an optional reason.""" + if self.status != 'failed': + self.status = 'skipped' + resolved = _extract_step_reason(reason) + if resolved: + self.reason = resolved + + def fail(self, reason: Any = None) -> None: + """Marks this step as failed with an optional reason or Output protobuf.""" + self.status = 'failed' + resolved = _extract_step_reason(reason) + if resolved: + self.reason = resolved + + def complete_stage(self, reason: Any = None, *, skip: bool = False) -> None: + """Marks this step as a legitimate early termination of the stage.""" + self.terminal = True + if self.tracker is not None: + self.tracker.terminal = True + resolved = _extract_step_reason(reason) + if skip and self.status != 'failed': + self.status = 'skipped' + if resolved: + self.reason = resolved + + def finish(self, end_time_ns: int | None = None) -> None: + """Finalizes step timing and registers the step on the active tracker.""" + if self.finished: + return + self.finished = True + if self.tracker is not None: + self.tracker.finish_step(self, end_time_ns=end_time_ns) + + def __enter__(self) -> '_StepHandle': + return self + + def __exit__(self, exc_type, exc_value, traceback_obj) -> None: + del traceback_obj + if exc_type is not None and self.status != 'failed': + msg = str(exc_value).strip().splitlines()[0] if exc_value else '' + self.fail(f'{exc_type.__name__}: {msg}' if msg else exc_type.__name__) + self.finish() + + def to_dict(self) -> dict[str, Any]: + """Serializes the step record for `jsonPayload.extras.stage_summary`.""" + data: dict[str, Any] = { + 'n': self.n, + 'name': self.name, + 'status': self.status, + 'duration_s': self.duration_s, + } + if self.reason: + data['reason'] = self.reason + return data + + +_STATUS_PRIORITY = {'ok': 0, 'skipped': 1, 'failed': 2} + + +class StageSummaryTracker: + """Accumulates structured step telemetry for a single task stage.""" + + def __init__(self, stage: Stage | str, start_time_ns: int | None = None): + self.stage = stage.value if isinstance(stage, Stage) else str(stage) + self.start_time_ns = ( + start_time_ns if start_time_ns is not None else time.time_ns()) + self.variant: str | None = None + self.steps: list[dict[str, Any]] = [] + self.terminal = False + self.last_error: str | None = None + self.error_type: str | None = None + self.main_error_type: str | None = None + self.context_extras: dict[str, Any] = {} + self.active_step: _StepHandle | None = None + + def start_step(self, handle: _StepHandle) -> None: + """Marks a step as active, finishing any previously active step.""" + if self.active_step is not None and not self.active_step.finished: + prev_ok = self.active_step.status == 'ok' + self.active_step.finish(end_time_ns=handle.start_time_ns) + if prev_ok: + self.last_error = None + handle.tracker = self + self.active_step = handle + + def finish_step(self, + handle: _StepHandle, + end_time_ns: int | None = None) -> None: + """Records step completion and merges repeated step entries.""" + end_ns = end_time_ns if end_time_ns is not None else time.time_ns() + handle.duration_s = round(max(0, end_ns - handle.start_time_ns) / 10**9, 3) + if handle.terminal: + self.terminal = True + if handle.status == 'failed' and not handle.reason and self.last_error: + handle.reason = self.last_error + step_dict = handle.to_dict() + for existing in self.steps: + if existing['n'] == handle.n and existing['name'] == handle.name: + existing['duration_s'] = round( + existing['duration_s'] + handle.duration_s, 3) + if (_STATUS_PRIORITY.get(step_dict['status'], 0) >= + _STATUS_PRIORITY.get(existing['status'], 0)): + existing['status'] = step_dict['status'] + if 'reason' in step_dict: + existing['reason'] = step_dict['reason'] + break + else: + self.steps.append(step_dict) + if self.active_step is handle: + self.active_step = None + + def record_error(self, message: str) -> None: + """Records an error log message emitted while the stage is active.""" + first_line = str(message).strip().splitlines()[0][:500] if message else '' + if not first_line: + return + if (self.last_error is None or + first_line != 'No uworker_input returned from preprocess'): + self.last_error = first_line + + def snapshot_log_context(self) -> None: + """Snapshots active FUZZER and TESTCASE log context metadata.""" + for ctx_type in (LogContextType.FUZZER, LogContextType.TESTCASE): + if ctx_type in log_contexts.contexts: + for k, v in ctx_type.get_extras()._asdict().items(): + if v is not None and v != 'null': + self.context_extras[k] = v + + def build_summary(self, + *, + end_time_ns: int | None = None, + failed: bool = False, + error_type: str | None = None, + main_error_type: str | None = None, + failure_reason: str | None = None) -> dict[str, Any]: + """Builds the structured `stage_summary` dictionary.""" + end_ns = end_time_ns if end_time_ns is not None else time.time_ns() + resolved_error_type = error_type or self.error_type + resolved_main_error_type = main_error_type or self.main_error_type + if self.active_step is not None and not self.active_step.finished: + if failed and self.active_step.status == 'ok' and not self.terminal: + self.active_step.fail(failure_reason or self.last_error or + resolved_error_type) + self.active_step.finish(end_time_ns=end_ns) + + duration_s = round(max(0, end_ns - self.start_time_ns) / 10**9, 3) + has_failed_step = any(s['status'] == 'failed' for s in self.steps) + stage_failed = bool(failed or has_failed_step) + + if stage_failed and not has_failed_step and self.steps: + last_step = self.steps[-1] + was_skipped = last_step['status'] == 'skipped' + last_step['status'] = 'failed' + fallback_reason = ( + failure_reason or self.last_error or resolved_error_type) + if fallback_reason and (not last_step.get('reason') or was_skipped): + last_step['reason'] = fallback_reason + + status = 'failed' if stage_failed else 'ok' + summary: dict[str, Any] = { + 'stage': self.stage, + 'status': status, + 'duration_s': duration_s, + 'steps': list(self.steps), + } + if self.variant: + summary['variant'] = self.variant + if self.terminal and status == 'ok': + summary['terminal'] = True + if resolved_error_type and resolved_error_type not in ('N/A', 'NO_ERROR'): + summary['error_type'] = resolved_error_type + if (resolved_main_error_type and + resolved_main_error_type not in ('N/A', 'NO_ERROR')): + summary['main_error_type'] = resolved_main_error_type + + if stage_failed: + step_reason = next( + (s.get('reason') + for s in reversed(self.steps) + if s.get('status') == 'failed' and s.get('reason')), + None, + ) + resolved_reason = ( + step_reason or failure_reason or self.last_error or + resolved_error_type) + if resolved_reason: + summary['failure_reason'] = resolved_reason + return summary + + to_dict = build_summary + + +def _stage_tracker_stack() -> list[StageSummaryTracker]: + stack = getattr(_STAGE_TRACKER_LOCAL, 'stack', None) + if stack is None: + stack = [] + _STAGE_TRACKER_LOCAL.stack = stack + return stack + + +def _current_stage_tracker() -> StageSummaryTracker | None: + stack = _stage_tracker_stack() + return stack[-1] if stack else None + + +def _clear_stage_tracker() -> None: + """Clears all active stage trackers on the current thread.""" + _stage_tracker_stack().clear() + + +def push_stage_tracker(tracker: StageSummaryTracker) -> None: + """Pushes a StageSummaryTracker onto the thread-local stack.""" + _stage_tracker_stack().append(tracker) + + +def pop_stage_tracker(tracker: StageSummaryTracker | None = None) -> None: + """Pops the active StageSummaryTracker from the thread-local stack.""" + stack = _stage_tracker_stack() + if not stack: + return + if tracker is None or stack[-1] is tracker: + stack.pop() + elif tracker in stack: + stack.remove(tracker) + + +def set_stage_variant(variant: str) -> None: + """Sets the execution variant (`cg`, `bb`, `lf`) on the active tracker.""" + tracker = _current_stage_tracker() + if tracker is not None: + tracker.variant = variant + + +def step(n: int, name: str) -> _StepHandle: + """Starts a numbered task step, usable either flat or as a `with` block.""" + handle = _StepHandle(n, name) + tracker = _current_stage_tracker() + if tracker is not None: + tracker.start_step(handle) + return handle + + +def emit_stage_summary(tracker: StageSummaryTracker, + *, + end_time_ns: int | None = None, + failed: bool = False, + error_type: str | None = None, + main_error_type: str | None = None, + failure_reason: str | None = None) -> dict[str, Any]: + """Emits the structured `stage_summary` log entry for `tracker`.""" + tracker.snapshot_log_context() + summary = tracker.build_summary( + end_time_ns=end_time_ns, + failed=failed, + error_type=error_type, + main_error_type=main_error_type, + failure_reason=failure_reason) + info( + f'Stage {tracker.stage} finished.', + stage_summary=summary, + **tracker.context_extras) + return summary + + +class _StageSummaryContext: + """Context manager that tracks steps and emits a `stage_summary` on exit.""" + + def __init__(self, + stage: Stage | str, + *, + start_time_ns: int | None = None, + emit_log: bool = True): + self._stage_enum = stage if isinstance(stage, Stage) else None + self._tracker = StageSummaryTracker(stage, start_time_ns=start_time_ns) + self._emit_log = emit_log + self._prev_stage = None + self._had_stage = False + + def __enter__(self) -> StageSummaryTracker: + if self._stage_enum is not None: + self._had_stage = 'stage' in log_contexts.meta + self._prev_stage = log_contexts.meta.get('stage') + log_contexts.add_metadata('stage', self._stage_enum) + push_stage_tracker(self._tracker) + return self._tracker + + def __exit__(self, exc_type, exc_value, traceback_obj) -> None: + del traceback_obj + exc_failed = exc_type is not None + exc_error_type = exc_type.__name__ if exc_type is not None else None + exc_reason = None + if exc_type is not None: + lines = str(exc_value).strip().splitlines() if exc_value else [] + msg = lines[0] if lines else '' + exc_reason = f'{exc_type.__name__}: {msg}' if msg else exc_type.__name__ + if (self._tracker.active_step is not None and + not self._tracker.active_step.finished): + self._tracker.active_step.fail(exc_reason) + try: + if self._emit_log: + emit_stage_summary( + self._tracker, + failed=exc_failed, + error_type=exc_error_type, + failure_reason=exc_reason) + else: + self._tracker.build_summary( + failed=exc_failed, + error_type=exc_error_type, + failure_reason=exc_reason) + finally: + pop_stage_tracker(self._tracker) + if self._stage_enum is not None: + if self._had_stage: + log_contexts.add_metadata('stage', self._prev_stage) + else: + log_contexts.delete_metadata('stage') + + +def stage_summary_context(stage: Stage | str, + *, + start_time_ns: int | None = None, + emit_log: bool = True) -> _StageSummaryContext: + """Returns a context manager that tracks steps and emits `stage_summary`.""" + return _StageSummaryContext( + stage, start_time_ns=start_time_ns, emit_log=emit_log) diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/impact_task_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/impact_task_test.py index 9b18f04dbb9..3847e1378da 100644 --- a/src/clusterfuzz/_internal/tests/core/bot/tasks/impact_task_test.py +++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/impact_task_test.py @@ -660,3 +660,68 @@ def test_android(self): [mock.call('123:456'), mock.call('somelink')]) self.mock.get_component_range_list.assert_has_calls( [mock.call(1, 100, 'android_job')]) + + +@test_utils.with_cloud_emulators('datastore') +class ExecuteTaskStageSummaryTest(unittest.TestCase): + """Tests structured step telemetry in impact_task.execute_task.""" + + def setUp(self): + helpers.patch_environ(self) + helpers.patch(self, [ + 'clusterfuzz._internal.base.utils.is_chromium', + 'clusterfuzz._internal.build_management.build_manager.is_custom_binary', + 'clusterfuzz._internal.bot.tasks.impact_task.get_impacts_from_url', + ('clusterfuzz._internal.common.testcase_utils.' + 'emit_testcase_triage_duration_metric'), + ]) + self.testcase = test_utils.create_generic_testcase() + + def test_execute_task_records_four_steps(self): + """Tests that a normal impact_task run records all 4 canonical steps.""" + from clusterfuzz._internal.metrics import logs + + self.mock.is_chromium.return_value = True + self.mock.is_custom_binary.return_value = False + self.testcase.regression = '100:200' + self.testcase.put() + self.mock.get_impacts_from_url.return_value = impact_task.Impacts( + impact_task.Impact('100.0', False), + impact_task.Impact('101.0', False), + impact_task.Impact('99.0', False), + impact_task.Impact('102.0', False), + ) + + with logs.stage_summary_context(logs.Stage.MAIN) as tracker: + impact_task.execute_task(str(self.testcase.key.id()), 'test_job') + summary = tracker.build_summary() + + self.assertEqual(summary['status'], 'ok') + self.assertEqual([(s['n'], s['name'], s['status']) + for s in summary['steps']], [ + (1, 'validate_testcase', 'ok'), + (2, 'check_regression_range', 'ok'), + (3, 'calculate_impacts', 'ok'), + (4, 'save_impacts', 'ok'), + ]) + + def test_execute_task_missing_regression_range_marks_terminal_skip(self): + """Tests that missing regression range marks step 2 terminal skipped.""" + from clusterfuzz._internal.metrics import logs + + self.mock.is_chromium.return_value = True + self.mock.is_custom_binary.return_value = False + self.testcase.regression = '' + self.testcase.put() + + with logs.stage_summary_context(logs.Stage.MAIN) as tracker: + impact_task.execute_task(str(self.testcase.key.id()), 'test_job') + summary = tracker.build_summary() + + self.assertEqual(summary['status'], 'ok') + self.assertTrue(summary['terminal']) + self.assertEqual([(s['n'], s['name'], s['status']) + for s in summary['steps']], [ + (1, 'validate_testcase', 'ok'), + (2, 'check_regression_range', 'skipped'), + ]) diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/task_types_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/task_types_test.py index 16f064025bf..4311791d2e8 100644 --- a/src/clusterfuzz/_internal/tests/core/bot/tasks/task_types_test.py +++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/task_types_test.py @@ -541,3 +541,21 @@ def test_neither_when_preprocess_fails(self): self._execute(command) self.mock.execute_locally.assert_not_called() self.mock.add_utask_main.assert_not_called() + + def test_trusted_tasks_emit_stage_summary(self): + """Tests that TrustedTask.execute emits a stage_summary log on exit.""" + for command in self.TRUSTED_COMMANDS: + with self.subTest(command=command): + with mock.patch(f'clusterfuzz._internal.bot.tasks.{command}_task' + '.execute_task') as execute_task, \ + mock.patch('clusterfuzz._internal.metrics.logs.info') as mock_info: + self._execute(command) + execute_task.assert_called_once_with('1', 'job') + summaries = [ + call.kwargs['stage_summary'] + for call in mock_info.call_args_list + if 'stage_summary' in call.kwargs + ] + self.assertEqual(len(summaries), 1) + self.assertEqual(summaries[0]['stage'], 'main') + self.assertEqual(summaries[0]['status'], 'ok') diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/analyze_task_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/analyze_task_test.py index 499afc487f3..17b693a2dd7 100644 --- a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/analyze_task_test.py +++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/analyze_task_test.py @@ -25,6 +25,7 @@ from clusterfuzz._internal.datastore import data_handler from clusterfuzz._internal.datastore import data_types from clusterfuzz._internal.metrics import events +from clusterfuzz._internal.metrics import logs from clusterfuzz._internal.protos import uworker_msg_pb2 from clusterfuzz._internal.tests.test_libs import helpers from clusterfuzz._internal.tests.test_libs import test_utils @@ -310,3 +311,39 @@ def test_event_emission_handle_noncrash_second_attempt(self): events.TestcaseRejectionEvent( testcase=self.testcase, rejection_reason=events.RejectionReason.ANALYZE_NO_REPRO)) + + +@test_utils.with_cloud_emulators('datastore') +class AnalyzeStageSummaryStepsTest(unittest.TestCase): + """Tests that analyze_task stages record canonical steps.""" + + def setUp(self): + helpers.patch_environ(self) + helpers.patch(self, [ + 'clusterfuzz._internal.bot.tasks.setup.preprocess_setup_testcase', + 'clusterfuzz._internal.bot.testcase_manager.preprocess_testcase_manager', + 'clusterfuzz._internal.build_management.build_manager.get_job_bad_revisions', + 'clusterfuzz._internal.common.testcase_utils.emit_testcase_triage_duration_metric', + ]) + self.testcase = test_utils.create_generic_testcase() + self.upload_metadata = data_types.TestcaseUploadMetadata( + testcase_id=self.testcase.key.id()) + self.upload_metadata.put() + + def test_preprocess_records_canonical_steps(self): + """Tests that utask_preprocess records load_testcase and prepare_env.""" + self.mock.preprocess_setup_testcase.return_value = ( + uworker_msg_pb2.SetupInput()) + self.mock.get_job_bad_revisions.return_value = [] + with logs.stage_summary_context(logs.Stage.PREPROCESS) as tracker: + uworker_input = analyze_task.utask_preprocess( + str(self.testcase.key.id()), 'test_job', {}) + summary = tracker.build_summary() + + self.assertIsNotNone(uworker_input) + self.assertEqual(summary['status'], 'ok') + self.assertEqual([(s['n'], s['name'], s['status']) + for s in summary['steps']], [ + (1, 'load_testcase', 'ok'), + (2, 'prepare_env', 'ok'), + ]) diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/utasks_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/utasks_test.py index 73dda51d270..99ebd40bd83 100644 --- a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/utasks_test.py +++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/utasks_test.py @@ -27,6 +27,7 @@ from clusterfuzz._internal.bot.tasks import utasks from clusterfuzz._internal.bot.tasks.utasks import analyze_task from clusterfuzz._internal.metrics import events +from clusterfuzz._internal.metrics import logs from clusterfuzz._internal.metrics import monitor from clusterfuzz._internal.metrics import monitoring_metrics from clusterfuzz._internal.protos import uworker_msg_pb2 @@ -288,6 +289,36 @@ def test_uworker_main(self, execution_mode: utasks.Mode): # a permission denied error if running in a untrusted worker). self.mock.emit.assert_not_called() + def test_uworker_main_emits_main_stage_summary(self): + """Tests that uworker_main_no_io emits stage_summary with stage 'main'.""" + self.mock.is_running_on_swarming.return_value = False + uworker_input = uworker_msg_pb2.Input( + job_type='job_type-value', + uworker_env=self.UWORKER_ENV, + ) + + def fake_main(inp): + del inp + logs.step(1, 'setup_build') + return uworker_msg_pb2.Output() + + self.module.utask_main.side_effect = fake_main + + with mock.patch('clusterfuzz._internal.metrics.logs.info') as mock_info: + utasks.uworker_main_no_io(self.module, uworker_input.SerializeToString()) + + summaries = [ + (call.args[0], call.kwargs['stage_summary']) + for call in mock_info.call_args_list + if 'stage_summary' in call.kwargs + ] + self.assertEqual(len(summaries), 1) + msg, summary = summaries[0] + self.assertEqual(msg, 'Stage main finished.') + self.assertEqual(summary['stage'], 'main') + self.assertEqual(summary['status'], 'ok') + self.assertEqual(summary['steps'][0]['name'], 'setup_build') + class GetUtaskModuleTest(unittest.TestCase): @@ -450,3 +481,42 @@ def test_event_emit_during_exception(self, execution_mode: utasks.Mode): self.mock.emit.assert_any_call(task_finished_event) self.mock.emit.assert_any_call(task_post_event) + + @parameterized.parameterized.expand([utasks.Mode.BATCH, utasks.Mode.SWARMING]) + def test_emits_stage_summary(self, execution_mode: utasks.Mode): + """Tests that tworker_postprocess emits a structured stage_summary log.""" + self.mock._get_execution_mode.return_value = execution_mode + download_url = 'https://uworker_output_download_url' + preprocess_start_timestamp = timestamp_pb2.Timestamp() + preprocess_start_timestamp.FromNanoseconds(time.time_ns()) + uworker_output = uworker_msg_pb2.Output( + uworker_input=uworker_msg_pb2.Input( + fuzzer_name='fuzzer_test', + job_type='foo-job', + preprocess_start_time=preprocess_start_timestamp), + ) + self.mock.download_and_deserialize_uworker_output.return_value = ( + uworker_output) + module = mock.MagicMock(__name__='mock_task') + + def fake_postprocess(output): + del output + logs.step(1, 'save_coverage_and_crashes') + + module.utask_postprocess.side_effect = fake_postprocess + self.mock.get_utask_module.return_value = module + task_utils._FUZZER_BASED_TASKS.add('mock') + + with mock.patch('clusterfuzz._internal.metrics.logs.info') as mock_info: + utasks.tworker_postprocess(download_url) + + summaries = [ + call.kwargs['stage_summary'] + for call in mock_info.call_args_list + if 'stage_summary' in call.kwargs + ] + self.assertEqual(len(summaries), 1) + self.assertEqual(summaries[0]['stage'], 'postprocess') + self.assertEqual(summaries[0]['status'], 'ok') + self.assertEqual(summaries[0]['steps'][0]['name'], + 'save_coverage_and_crashes') diff --git a/src/clusterfuzz/_internal/tests/core/metrics/logs_test.py b/src/clusterfuzz/_internal/tests/core/metrics/logs_test.py index fd92f73173b..930b0a23819 100644 --- a/src/clusterfuzz/_internal/tests/core/metrics/logs_test.py +++ b/src/clusterfuzz/_internal/tests/core/metrics/logs_test.py @@ -1563,3 +1563,256 @@ def incrementer(): # Number of increments plus the common context. self.assertEqual(len(run_bot_logs.log_contexts.contexts), num_it + 1) + + +class StageSummaryTest(unittest.TestCase): + """Tests for StageSummaryTracker, logs.step, and emit_stage_summary.""" + + def setUp(self): + logs.log_contexts.clear() + logs._clear_stage_tracker() # pylint: disable=protected-access + helpers.patch(self, ['clusterfuzz._internal.metrics.logs.info']) + + def tearDown(self): + logs._clear_stage_tracker() # pylint: disable=protected-access + logs.log_contexts.clear() + + def test_flat_steps_and_summary(self): + """Tests flat logs.step phase transitions and summary emission.""" + with logs.stage_summary_context(logs.Stage.PREPROCESS): + logs.step(1, 'load_testcase') + logs.step(2, 'prepare_env') + + self.mock.info.assert_called_once() + args, kwargs = self.mock.info.call_args + self.assertEqual(args[0], 'Stage preprocess finished.') + summary = kwargs['stage_summary'] + self.assertEqual(summary['stage'], 'preprocess') + self.assertEqual(summary['status'], 'ok') + self.assertEqual(len(summary['steps']), 2) + self.assertEqual(summary['steps'][0]['n'], 1) + self.assertEqual(summary['steps'][0]['name'], 'load_testcase') + self.assertEqual(summary['steps'][0]['status'], 'ok') + self.assertEqual(summary['steps'][1]['n'], 2) + self.assertEqual(summary['steps'][1]['name'], 'prepare_env') + self.assertEqual(summary['steps'][1]['status'], 'ok') + + def test_step_skip_and_terminal_complete_stage(self): + """Tests skipping a step and marking a stage terminal.""" + with logs.stage_summary_context(logs.Stage.MAIN): + logs.set_stage_variant('lf') + logs.step(1, 'setup_build') + step2 = logs.step(2, 'verify_initial_crash') + step2.complete_stage('Did not crash', skip=True) + + summary = self.mock.info.call_args[1]['stage_summary'] + self.assertEqual(summary['variant'], 'lf') + self.assertEqual(summary['status'], 'ok') + self.assertTrue(summary['terminal']) + self.assertEqual(summary['steps'][1]['status'], 'skipped') + self.assertEqual(summary['steps'][1]['reason'], 'Did not crash') + + def test_complete_stage_without_skip_preserves_reason(self): + """Tests that complete_stage(reason, skip=False) keeps reason on step.""" + with logs.stage_summary_context(logs.Stage.MAIN): + logs.step(1, 'setup_build') + step2 = logs.step(2, 'test_for_crash') + step2.complete_stage('Testcase did not crash') + + summary = self.mock.info.call_args[1]['stage_summary'] + self.assertEqual(summary['status'], 'ok') + self.assertTrue(summary['terminal']) + self.assertEqual(summary['steps'][1]['status'], 'ok') + self.assertEqual(summary['steps'][1]['reason'], 'Testcase did not crash') + + def test_step_exception_marks_stage_failed(self): + """Tests that an unhandled exception inside a step marks it failed.""" + with self.assertRaises(RuntimeError): + with logs.stage_summary_context(logs.Stage.MAIN): + with logs.step(1, 'setup_build'): + raise RuntimeError('disk full') + + summary = self.mock.info.call_args[1]['stage_summary'] + self.assertEqual(summary['status'], 'failed') + self.assertEqual(summary['failure_reason'], 'RuntimeError: disk full') + self.assertEqual(summary['steps'][0]['status'], 'failed') + self.assertEqual(summary['steps'][0]['reason'], 'RuntimeError: disk full') + + def test_log_error_captured_as_stage_reason(self): + """Tests that logs.error records the error message on the active step.""" + with logs.stage_summary_context( + logs.Stage.POSTPROCESS, emit_log=False) as tracker: + with logs.step(1, 'update_testcase'): + tracker.record_error('Datastore write failed') + logs.emit_stage_summary(tracker, failed=True) + + summary = self.mock.info.call_args[1]['stage_summary'] + self.assertEqual(summary['status'], 'failed') + self.assertEqual(summary['failure_reason'], 'Datastore write failed') + self.assertEqual(summary['steps'][0]['status'], 'failed') + self.assertEqual(summary['steps'][0]['reason'], 'Datastore write failed') + + def test_non_fatal_error_does_not_pollute_ok_step_or_leak(self): + """Tests that non-fatal logs.error during an ok step does not pollute or leak.""" + with logs.stage_summary_context( + logs.Stage.MAIN, emit_log=False) as tracker: + logs.step(1, 'setup_build') + tracker.record_error('Transient warning') + step2 = logs.step(2, 'test_for_crash') + step2.fail() + logs.emit_stage_summary(tracker, failed=True) + + summary = self.mock.info.call_args[1]['stage_summary'] + self.assertEqual(summary['steps'][0]['status'], 'ok') + self.assertNotIn('reason', summary['steps'][0]) + self.assertEqual(summary['steps'][1]['status'], 'failed') + self.assertNotIn('reason', summary['steps'][1]) + + def test_snapshot_log_context_preserves_fuzzer_and_job_extras(self): + """Tests that fuzzer/testcase log context extras are preserved on exit.""" + fuzz_target = mock.Mock(binary='xml_fuzzer') + with logs.stage_summary_context(logs.Stage.MAIN): + with logs.fuzzer_log_context('libFuzzer', 'libfuzzer_asan_job', + fuzz_target): + logs.step(1, 'sync_bundle') + + _, kwargs = self.mock.info.call_args + self.assertEqual(kwargs.get('fuzzer'), 'libFuzzer') + self.assertEqual(kwargs.get('job'), 'libfuzzer_asan_job') + self.assertEqual(kwargs.get('fuzz_target'), 'xml_fuzzer') + + def test_all_eleven_task_modules_have_canonical_steps(self): + """Verifies that all 11 task modules contain canonical logs.step calls.""" + import ast + import pathlib + + tasks_dir = ( + pathlib.Path(__file__).resolve().parents[3] / 'bot' / 'tasks') + expected = { + 'utasks/analyze_task.py': { + (1, 'load_testcase'), + (2, 'prepare_env'), + (1, 'setup_build'), + (2, 'test_for_crash'), + (3, 'check_reproducibility'), + (4, 'package_output'), + (1, 'update_testcase'), + (2, 'create_post_analyze_tasks'), + }, + 'utasks/minimize_task.py': { + (1, 'load_testcase'), + (2, 'prepare_env'), + (1, 'setup_build'), + (2, 'verify_initial_crash'), + (3, 'minimize_sanitizer_options'), + (4, 'minimize_rounds'), + (5, 'package_output'), + (2, 'warmup_and_verify'), + (3, 'minimize_gestures'), + (4, 'minimize_main_file'), + (5, 'minimize_resources'), + (6, 'minimize_arguments'), + (7, 'store_minimized'), + (1, 'update_testcase'), + (2, 'finalize_and_fan_out'), + }, + 'utasks/fuzz_task.py': { + (1, 'select_target'), + (2, 'sign_urls'), + (1, 'sync_bundle'), + (2, 'fetch_build'), + (3, 'check_bad_build'), + (4, 'sync_corpus'), + (5, 'fuzz_and_upload_corpus'), + (4, 'generate_inputs'), + (5, 'execute_inputs'), + (6, 'process_crashes'), + (1, 'record_stats'), + (2, 'dedup_and_file_bugs'), + (3, 'upload_logs'), + }, + 'utasks/corpus_pruning_task.py': { + (1, 'check_last_status'), + (2, 'sign_urls'), + (1, 'sync_bundle'), + (2, 'fetch_build'), + (3, 'sync_and_restore_quarantine'), + (4, 'merge_corpus'), + (5, 'sync_and_backup_corpus'), + (6, 'triage_bad_units'), + (7, 'cross_pollinate'), + (1, 'save_coverage_and_crashes'), + }, + 'utasks/regression_task.py': { + (1, 'load_testcase'), + (2, 'prepare_env'), + (1, 'setup_testcase'), + (2, 'fetch_revisions'), + (3, 'reproduce_at_max'), + (4, 'find_min_revision'), + (5, 'bisect_range'), + (6, 'validate_range'), + (1, 'update_metadata'), + (2, 'save_regression_range'), + }, + 'utasks/symbolize_task.py': { + (1, 'load_testcase'), + (2, 'prepare_env'), + (1, 'setup_release_build'), + (2, 'search_asan_redzone'), + (3, 'setup_symbolized_build'), + (4, 'reproduce_symbolized'), + (5, 'package_output'), + (1, 'save_symbolized_stack'), + }, + 'utasks/variant_task.py': { + (1, 'clone_testcase_and_prepare'), + (1, 'setup_build'), + (2, 'test_for_crash'), + (3, 'check_reproducibility'), + (4, 'package_output'), + (1, 'save_variant'), + }, + 'utasks/progression_task.py': { + (1, 'load_testcase'), + (2, 'prepare_env'), + (1, 'setup_testcase'), + (2, 'fetch_revisions'), + (3, 'setup_trunk_build'), + (4, 'test_at_trunk'), + (5, 'bisect_fixed_range'), + (6, 'package_output'), + (1, 'update_metadata'), + (2, 'save_fixed_or_open'), + }, + 'impact_task.py': { + (1, 'validate_testcase'), + (2, 'check_regression_range'), + (3, 'calculate_impacts'), + (4, 'save_impacts'), + }, + 'blame_task.py': { + (1, 'validate_testcase'), + (2, 'compute_dependency_rolls'), + (3, 'publish_to_predator'), + }, + 'unpack_task.py': { + (1, 'load_bundle_metadata'), + (2, 'download_and_extract_archive'), + (3, 'create_testcases'), + (4, 'cleanup'), + }, + } + + for rel_path, expected_steps in expected.items(): + tree = ast.parse((tasks_dir / rel_path).read_text()) + found_steps = set() + for node in ast.walk(tree): + if (isinstance(node, ast.Call) and + isinstance(node.func, ast.Attribute) and + isinstance(node.func.value, ast.Name) and + node.func.value.id == 'logs' and node.func.attr == 'step' and + len(node.args) == 2 and isinstance(node.args[0], ast.Constant) and + isinstance(node.args[1], ast.Constant)): + found_steps.add((node.args[0].value, node.args[1].value)) + self.assertEqual(found_steps, expected_steps, f'Mismatch in {rel_path}')