Add test analytics breadcrumbs

apps/worker/tasks/base.py

@@ -38,6 +38,7 @@
     BreadcrumbData,
     Errors,
     Milestones,
+    ReportTypes,
 )
 from shared.metrics import Counter, Histogram
 from shared.torngit.base import TorngitBaseAdapter
@@ -541,6 +542,7 @@ def _call_upload_breadcrumb_task(
         upload_ids: list[int] = [],
         error: Errors | None = None,
         error_text: str | None = None,
+        report_type: ReportTypes | None = None,
     ):
         """
         Queue a task to create an upload breadcrumb.
@@ -551,7 +553,10 @@ def _call_upload_breadcrumb_task(
                     "commit_sha": commit_sha,
                     "repo_id": repo_id,
                     "breadcrumb_data": BreadcrumbData(
-                        milestone=milestone, error=error, error_text=error_text
+                        milestone=milestone,
+                        error=error,
+                        error_text=error_text,
+                        report_type=report_type,
                     ),
                     "upload_ids": upload_ids,
                     "sentry_trace_id": current_sentry_trace_id(),
@@ -567,5 +572,6 @@ def _call_upload_breadcrumb_task(
                     "upload_ids": upload_ids,
                     "error": error,
                     "error_text": error_text,
+                    "report_type": report_type,
                 },
             )

apps/worker/tasks/test_results_finisher.py

@@ -10,6 +10,7 @@
 from services.lock_manager import LockManager, LockRetry, LockType
 from services.test_analytics.ta_finish_upload import ta_finish_upload
 from shared.celery_config import test_results_finisher_task_name
+from shared.django_apps.upload_breadcrumbs.models import Errors, Milestones, ReportTypes
 from shared.yaml import UserYaml
 from tasks.base import BaseCodecovTask
 from tasks.notify import notify_task_name
@@ -56,6 +57,14 @@ def run_impl(
                     commit_yaml=UserYaml.from_dict(commit_yaml),
                     **kwargs,
                 )
+
+            self._call_upload_breadcrumb_task(
+                commit_sha=commitid,
+                repo_id=repoid,
+                milestone=Milestones.UPLOAD_COMPLETE,
+                report_type=ReportTypes.TEST_RESULTS,
+            )
+
             if finisher_result["queue_notify"]:
                 self.app.tasks[notify_task_name].apply_async(
                     args=None,
@@ -65,12 +74,44 @@ def run_impl(
                         "current_yaml": commit_yaml,
                     },
                 )
+                self._call_upload_breadcrumb_task(
+                    commit_sha=commitid,
+                    repo_id=repoid,
+                    milestone=Milestones.NOTIFICATIONS_TRIGGERED,
+                    report_type=ReportTypes.TEST_RESULTS,
+                )
 
             return finisher_result
 
         except LockRetry as retry:
+            self._call_upload_breadcrumb_task(
+                commit_sha=commitid,
+                repo_id=repoid,
+                error=Errors.INTERNAL_LOCK_ERROR,
+                report_type=ReportTypes.TEST_RESULTS,
+            )
+            self._call_upload_breadcrumb_task(
+                commit_sha=commitid,
+                repo_id=repoid,
+                error=Errors.INTERNAL_RETRYING,
+                report_type=ReportTypes.TEST_RESULTS,
+            )
             self.retry(max_retries=5, countdown=retry.countdown)
 
+        except Exception as e:
+            log.exception(
+                "Error in test results finisher",
+                extra={"repoid": repoid, "commitid": commitid},
+            )
+            self._call_upload_breadcrumb_task(
+                commit_sha=commitid,
+                repo_id=repoid,
+                error=Errors.UNKNOWN,
+                error_text=repr(e),
+                report_type=ReportTypes.TEST_RESULTS,
+            )
+            raise
+
     def process_impl_within_lock(
         self,
         *,

apps/worker/tasks/test_results_processor.py

@@ -6,6 +6,7 @@
 from services.processing.types import UploadArguments
 from services.test_analytics.ta_processor import ta_processor
 from shared.celery_config import test_results_processor_task_name
+from shared.django_apps.upload_breadcrumbs.models import Errors, Milestones, ReportTypes
 from tasks.base import BaseCodecovTask
 
 log = logging.getLogger(__name__)
@@ -25,15 +26,44 @@ def run_impl(
         arguments_list: list[UploadArguments],
         **kwargs,
     ) -> bool:
-        for argument in arguments_list:
-            ta_processor(
-                repoid=repoid,
-                commitid=commitid,
-                commit_yaml=commit_yaml,
-                argument=argument,
-                update_state=True,
+        upload_ids = [
+            arg.get("upload_id") for arg in arguments_list if arg.get("upload_id")
+        ]
+
+        # Log breadcrumb for processing start
+        self._call_upload_breadcrumb_task(
+            commit_sha=commitid,
+            repo_id=repoid,
+            milestone=Milestones.PROCESSING_UPLOAD,
+            upload_ids=upload_ids,
+            report_type=ReportTypes.TEST_RESULTS,
+        )
+
+        try:
+            for argument in arguments_list:
+                ta_processor(
+                    repoid=repoid,
+                    commitid=commitid,
+                    commit_yaml=commit_yaml,
+                    argument=argument,
+                    update_state=True,
+                )
+            return True
+        except Exception as e:
+            log.exception(
+                "Error processing test results",
+                extra={"repoid": repoid, "commitid": commitid},
+            )
+            self._call_upload_breadcrumb_task(
+                commit_sha=commitid,
+                repo_id=repoid,
+                milestone=Milestones.PROCESSING_UPLOAD,
+                upload_ids=upload_ids,
+                error=Errors.UNKNOWN,
+                error_text=repr(e),
+                report_type=ReportTypes.TEST_RESULTS,
             )
-        return True
+            raise
 
 
 RegisteredTestResultsProcessorTask = celery_app.register_task(