Emit structured stage_summary telemetry across all 11 bot tasks.

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
+}
+

src/clusterfuzz/_internal/bot/tasks/blame_task.py

@@ -271,27 +271,32 @@ 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)
 
   # Make sure that predator topic is configured. If not, nothing to do here.
   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)
 

src/clusterfuzz/_internal/bot/tasks/impact_task.py

@@ -273,20 +273,24 @@ 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.
   testcase = data_handler.get_testcase_by_id(testcase_id)
 
   # 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(

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,

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()

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', '')