feat: new TA processor implementation (#1117)

Author: joseph-sentry

conftest.py

@@ -204,6 +204,8 @@ def mock_configuration(mocker):
                 "hash_key": "88f572f4726e4971827415efa8867978",
                 "secret_access_key": "codecov-default-secret",
                 "verify_ssl": False,
+                "host": "minio",
+                "port": 9000,
             },
             "smtp": {
                 "host": "mailhog",

docker-compose.yml

@@ -7,6 +7,7 @@ services:
       - postgres
       - redis
       - timescale
+      - minio
     volumes:
       - ./:/app/apps/worker
       - ./docker/test_codecov_config.yml:/config/codecov.yml
@@ -41,6 +42,19 @@ services:
     volumes:
       - ./docker/init_db.sql:/docker-entrypoint-initdb.d/init_db.sql
 
+  minio:
+    image: minio/minio:latest
+    command: server /export
+    ports:
+      - "${MINIO_PORT:-9000}:9000"
+    environment:
+      - MINIO_ACCESS_KEY=codecov-default-key
+      - MINIO_SECRET_KEY=codecov-default-secret
+    volumes:
+      - type: tmpfs
+        target: /export
+        tmpfs:
+          size: 256M
   redis:
     image: redis:6-alpine
 

docker/test_codecov_config.yml

@@ -18,6 +18,8 @@ services:
     access_key_id: codecov-default-key
     secret_access_key: codecov-default-secret
     verify_ssl: false
+    port: 9000
+    host: minio
   smtp:
     host: mailhog
     port: 1025

services/test_analytics/ta_processing.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import sentry_sdk
+import test_results_parser
+from shared.config import get_config
+from shared.django_apps.core.models import Commit, Repository
+from shared.django_apps.reports.models import ReportSession, UploadError
+
+from services.archive import ArchiveService
+from services.test_analytics.ta_timeseries import get_flaky_tests_set, insert_testrun
+from services.yaml import UserYaml, read_yaml_field
+
+
+@dataclass
+class TAProcInfo:
+    repository: Repository
+    branch: str | None
+    user_yaml: UserYaml
+
+
+def handle_file_not_found(upload: ReportSession):
+    upload.state = "processed"
+    upload.save()
+    UploadError.objects.create(
+        report_session=upload,
+        error_code="file_not_in_storage",
+        error_params={},
+    )
+
+
+def handle_parsing_error(upload: ReportSession, exc: Exception):
+    sentry_sdk.capture_exception(exc, tags={"upload_state": upload.state})
+    upload.state = "processed"
+    upload.save()
+    UploadError.objects.create(
+        report_session=upload,
+        error_code="unsupported_file_format",
+        error_params={"error_message": str(exc)},
+    )
+
+
+def get_ta_processing_info(
+    repoid: int,
+    commitid: str,
+    commit_yaml: dict[str, Any],
+) -> TAProcInfo:
+    repository = Repository.objects.get(repoid=repoid)
+
+    commit = Commit.objects.get(repository=repository, commitid=commitid)
+    branch = commit.branch
+    if branch is None:
+        raise ValueError("Branch is None")
+
+    user_yaml: UserYaml = UserYaml(commit_yaml)
+    return TAProcInfo(
+        repository,
+        branch,
+        user_yaml,
+    )
+
+
+def should_delete_archive_settings(user_yaml: UserYaml) -> bool:
+    if get_config("services", "minio", "expire_raw_after_n_days"):
+        return True
+    return not read_yaml_field(user_yaml, ("codecov", "archive", "uploads"), _else=True)
+
+
+def rewrite_or_delete_upload(
+    archive_service: ArchiveService,
+    user_yaml: UserYaml,
+    upload: ReportSession,
+    readable_file: bytes,
+):
+    if should_delete_archive_settings(user_yaml):
+        archive_url = upload.storage_path
+        if archive_url and not archive_url.startswith("http"):
+            archive_service.delete_file(archive_url)
+    else:
+        archive_service.write_file(upload.storage_path, bytes(readable_file))
+
+
+def insert_testruns_timeseries(
+    repoid: int,
+    commitid: str,
+    branch: str | None,
+    upload: ReportSession,
+    parsing_infos: list[test_results_parser.ParsingInfo],
+):
+    flaky_test_set = get_flaky_tests_set(repoid)
+
+    for parsing_info in parsing_infos:
+        insert_testrun(
+            timestamp=upload.created_at,
+            repo_id=repoid,
+            commit_sha=commitid,
+            branch=branch,
+            upload_id=upload.id,
+            flags=upload.flag_names,
+            parsing_info=parsing_info,
+            flaky_test_ids=flaky_test_set,
+        )

services/test_analytics/ta_processor.py

@@ -0,0 +1,80 @@
+import logging
+from typing import Any
+
+from shared.django_apps.reports.models import ReportSession
+from shared.storage.exceptions import FileNotInStorageError
+from test_results_parser import parse_raw_upload
+
+from services.archive import ArchiveService
+from services.processing.types import UploadArguments
+from services.test_analytics.ta_processing import (
+    get_ta_processing_info,
+    handle_file_not_found,
+    handle_parsing_error,
+    insert_testruns_timeseries,
+    rewrite_or_delete_upload,
+)
+
+log = logging.getLogger(__name__)
+
+
+def ta_processor_impl(
+    repoid: int,
+    commitid: str,
+    commit_yaml: dict[str, Any],
+    argument: UploadArguments,
+    update_state: bool = False,
+) -> bool:
+    log.info(
+        "Processing single TA argument",
+        extra=dict(
+            upload_id=argument.get("upload_id"),
+            repoid=repoid,
+            commitid=commitid,
+        ),
+    )
+
+    upload_id = argument.get("upload_id")
+    if upload_id is None:
+        return False
+
+    upload = ReportSession.objects.get(id=upload_id)
+    if upload.state == "processed":
+        # don't need to process again because the intermediate result should already be in redis
+        return False
+
+    if upload.storage_path is None:
+        if update_state:
+            handle_file_not_found(upload)
+        return False
+
+    ta_proc_info = get_ta_processing_info(repoid, commitid, commit_yaml)
+
+    archive_service = ArchiveService(ta_proc_info.repository)
+
+    try:
+        payload_bytes = archive_service.read_file(upload.storage_path)
+    except FileNotInStorageError:
+        if update_state:
+            handle_file_not_found(upload)
+        return False
+
+    try:
+        parsing_infos, readable_file = parse_raw_upload(payload_bytes)
+    except RuntimeError as exc:
+        if update_state:
+            handle_parsing_error(upload, exc)
+        return False
+
+    insert_testruns_timeseries(
+        repoid, commitid, ta_proc_info.branch, upload, parsing_infos
+    )
+
+    if update_state:
+        upload.state = "processed"
+        upload.save()
+
+        rewrite_or_delete_upload(
+            archive_service, ta_proc_info.user_yaml, upload, readable_file
+        )
+    return True

services/test_analytics/tests/conftest.py

@@ -0,0 +1,14 @@
+import pytest
+from shared.config import get_config
+from shared.storage import get_appropriate_storage_service
+from shared.storage.exceptions import BucketAlreadyExistsError
+
+
+@pytest.fixture
+def storage(mock_configuration):
+    storage_service = get_appropriate_storage_service()
+    try:
+        storage_service.create_root_storage(get_config("services", "minio", "bucket"))
+    except BucketAlreadyExistsError:
+        pass
+    return storage_service

services/test_analytics/tests/samples/sample_test.json

@@ -0,0 +1,11 @@
+{
+    "test_results_files": [
+        {
+            "filename": "codecov-demo/temp.junit.xml",
+            "format": "base64+compressed",
+            "data": "eJy1VMluwjAQvfMVI1dCoBbHZiklJEFVS4V66Kkqx8okBqw6i2KHwt/XWSChnCrRXDLjefNm8Uuc2T6UsOOpEnHkIooJAh75cSCijYsyve49oJnnaK60yoR5NWyIWMhdlBzyE5OWpnGqXGQY1kzILOXGoQjUl0gSHhSBItdFQ2OJPJdgMuqXjtIsTFzUJ/1Bj9IeuX+n1KZjmwwxoZSMDWwbK13W/HhZvC1fn5eLCZaxzyQq2/KZ4uBLplQJY4nAmocJNhA/k0zHKc5xn7WPqimKYxYEjc6Iad66DrHKVjplvv4f9jCTWiTy0GQnV2MPxE4E/Lxzz6muGMzFKbbJaZXiqQajIHBdIHjUvqFkCrcA31tugEUA2lJP11nkayM3eKobKIsA00D2lAz9CV9NSHujpx16B/3uievI9mceU/sChryAr6ExZKdrtwpw+VQjXeSVPVVjtubn6HoBp8iVlnDGd9VFtFpGFFYuCqsWgfVLFDg52ANiw2Mxpyk3zz94x6qU4DnWUW2VWfwkmrbyfgBbcXMH",
+            "labels": ""
+        }
+    ],
+    "metadata": {}
+}

services/test_analytics/tests/snapshots/ta_processing__insert_testruns_timeseries__0.json

@@ -0,0 +1,38 @@
+[
+  {
+    "timestamp": "2025-01-01T00:00:00+00:00",
+    "test_id": "7a44f8a4b65ee2abd9617fc99a63fc2e",
+    "name": "test_1_name",
+    "classname": "test_1_classname",
+    "testsuite": "test_1_testsuite",
+    "computed_name": "test_1_computed_name",
+    "outcome": "pass",
+    "duration_seconds": 1.0,
+    "failure_message": null,
+    "framework": "Pytest",
+    "filename": "test_1_file",
+    "repo_id": 1,
+    "commit_sha": "123",
+    "branch": "main",
+    "flags": [],
+    "upload_id": 1
+  },
+  {
+    "timestamp": "2025-01-01T00:00:00+00:00",
+    "test_id": "25ce6e22db03ef4f4230eb999c776f99",
+    "name": "test_2_name",
+    "classname": "test_2_classname",
+    "testsuite": "test_2_testsuite",
+    "computed_name": "test_2",
+    "outcome": "failure",
+    "duration_seconds": 1.0,
+    "failure_message": "test_2_failure_message",
+    "framework": "Pytest",
+    "filename": "test_2_file",
+    "repo_id": 1,
+    "commit_sha": "123",
+    "branch": "main",
+    "flags": [],
+    "upload_id": 1
+  }
+]