[WIP] Bug 2051180 - Track which pushes are actually backfilled

treeherder/perf/auto_perf_sheriffing/backfill_tracker.py

@@ -0,0 +1,129 @@
+import ast
+import json
+import logging
+import re
+
+import requests
+
+from treeherder.model.models import JobLog, TaskclusterMetadata
+from treeherder.perf.models import BackfilledPush, BackfillRequest
+
+logger = logging.getLogger(__name__)
+
+
+class BackfillTracker:
+    RE_BACKFILL_DATA = re.compile(r".*?BACKFILL_DATA:\s+({.*})")
+
+    def sync(self):
+        pending_requests = BackfillRequest.objects.filter(status=BackfillRequest.PENDING)
+        for request in pending_requests:
+            try:
+                self._sync_one(request)
+            except Exception as ex:
+                logger.warning(f"Failed to sync BackfillRequest {request.id}: {ex}")
+
+    def _sync_one(self, request: BackfillRequest):
+        try:
+            tc_metadata = TaskclusterMetadata.objects.get(task_id=request.bk_task_id)
+        except TaskclusterMetadata.DoesNotExist:
+            logger.info(
+                f"BackfillRequest {request.id}: bk_task {request.bk_task_id} not yet ingested, skipping."
+            )
+            return
+
+        job = tc_metadata.job
+        try:
+            job_log = job.job_log.get(name="live_backing_log")
+        except JobLog.DoesNotExist:
+            logger.warning(
+                f"BackfillRequest {request.id}: live_backing_log not found for job {job.id}."
+            )
+            request.status = BackfillRequest.FAILED
+            request.save()
+            return
+
+        if job_log.status != JobLog.PARSED:
+            logger.info(
+                f"BackfillRequest {request.id}: live_backing_log not yet parsed for job {job.id}, skipping."
+            )
+            return
+
+        try:
+            response = requests.get(job_log.url, timeout=30)
+            response.raise_for_status()
+        except requests.RequestException as ex:
+            logger.warning(
+                f"BackfillRequest {request.id}: Failed to fetch log from {job_log.url}: {ex}"
+            )
+            return
+
+        for line in response.text.splitlines():
+            if "Backfill started:" in line:
+                parsed = self._parse_backfill_started(line)
+                request.label = parsed["label"]
+                request.strategy = parsed["strategy"]
+                request.slices = parsed["slices"]
+                request.target_hg_push_ids = parsed["target_hg_push_ids"]
+                request.scanned_push_count = parsed["scanned_push_count"]
+            elif "BACKFILL_DATA:" in line:
+                match = self.RE_BACKFILL_DATA.match(line)
+                if match:
+                    try:
+                        entry = json.loads(match.group(1))
+                    except json.JSONDecodeError as ex:
+                        logger.warning(
+                            f"BackfillRequest {request.id}: Failed to parse BACKFILL_DATA line: {ex}"
+                        )
+                        continue
+                try:
+                    push_id = TaskclusterMetadata.objects.get(
+                        task_id=entry["decision_task_id"]
+                    ).job.push_id
+                except TaskclusterMetadata.DoesNotExist:
+                    logger.warning(
+                        f"BackfillRequest {request.id}: decision task {entry['decision_task_id']} not yet ingested, skipping line."
+                    )
+                    continue
+                BackfilledPush.objects.get_or_create(
+                    backfill_request=request,
+                    push_id=push_id,
+                    defaults={"decision_task_id": entry["decision_task_id"]},
+                )
+
+        if job.state == "completed":
+            request.status = BackfillRequest.SYNCED
+
+        request.save()
+
+    @staticmethod
+    def _parse_backfill_started(line: str) -> dict:
+        content = line[len("Backfill started: ") :]
+
+        label_match = re.search(r"label=([^,]+)", content)
+        strategy_match = re.search(r"strategy=([^,]+)", content)
+        slices_match = re.search(r"slices=(\d+)", content)
+        target_pushes_match = re.search(r"target_pushes=(\[.*?\])", content)
+
+        label = label_match.group(1).strip() if label_match else ""
+        strategy_str = strategy_match.group(1).strip() if strategy_match else "standard"
+        slices = int(slices_match.group(1)) if slices_match else 0
+
+        target_hg_push_ids = []
+        if target_pushes_match:
+            try:
+                raw = ast.literal_eval(target_pushes_match.group(1))
+                target_hg_push_ids = [str(p) for p in raw]
+            except (ValueError, SyntaxError) as ex:
+                logger.warning(f"Failed to parse target_pushes: {ex}")
+
+        strategy = (
+            BackfillRequest.STANDARD if strategy_str == "standard" else BackfillRequest.SLICED
+        )
+
+        return {
+            "label": label,
+            "strategy": strategy,
+            "slices": slices,
+            "target_hg_push_ids": target_hg_push_ids,
+            "scanned_push_count": len(target_hg_push_ids),
+        }

treeherder/perf/auto_perf_sheriffing/sherlock.py

@@ -14,6 +14,7 @@
     BackfillReportMaintainer,
 )
 from treeherder.perf.auto_perf_sheriffing.backfill_tool import BackfillTool
+from treeherder.perf.auto_perf_sheriffing.backfill_tracker import BackfillTracker
 from treeherder.perf.auto_perf_sheriffing.performance_alerting.alert_manager import (
     PerformanceAlertManager,
 )
@@ -33,6 +34,7 @@
     BackfillNotificationRecord,
     BackfillRecord,
     BackfillReport,
+    BackfillRequest,
     PerformanceDatum,
     PerformanceFramework,
     PerformanceTelemetryAlert,
@@ -84,6 +86,10 @@ def sheriff(self, since: datetime, frameworks: list[str], repositories: list[str
         self.secretary.mark_reports_for_backfill()
         self.assert_can_run()
 
+        logger.info("Sherlock: Syncing backfill tracking data...")
+        BackfillTracker().sync()
+        self.assert_can_run()
+
         # secretary checks the status of all backfilled jobs
         self.secretary.check_outcome()
         self.assert_can_run()
@@ -195,15 +201,23 @@ def _backfill_record(self, record: BackfillRecord, left: int) -> tuple[int, int]
             try:
                 if isinstance(data_point, dict):
                     using_job_id = data_point["job_id"]
+                    push_id = data_point["push_id"]
                 else:
                     using_job_id = data_point.job_id
+                    push_id = data_point.push_id
                 if not using_job_id:
                     logger.warning(
                         f"Failed to backfill [alert_id={record.alert.id}]: invalid job id."
                     )
                     continue
                 task_id = self.backfill_tool.backfill_job(using_job_id, alert_id=record.alert.id)
                 task_ids[using_job_id] = task_id
+                BackfillRequest.objects.create(
+                    job_id=using_job_id,
+                    push_id=push_id,
+                    repository=record.repository,
+                    bk_task_id=task_id,
+                )
                 left, consumed = left - 1, consumed + 1
             except (KeyError, CannotBackfillError, Exception) as ex:
                 logger.warning(f"Failed to backfill [alert_id={record.alert.id}]: {ex}")

treeherder/perf/migrations/0078_backfillrequest_backfilledpush.py

@@ -0,0 +1,110 @@
+# Generated by Django 6.0.3 on 2026-07-03 07:23
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("model", "0053_push_branch"),
+        ("perf", "0077_add_infra_status"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="BackfillRequest",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("bk_task_id", models.CharField(db_index=True, max_length=255)),
+                ("label", models.CharField(blank=True, max_length=255, null=True)),
+                (
+                    "strategy",
+                    models.IntegerField(
+                        blank=True, choices=[(0, "Standard"), (1, "Sliced")], null=True
+                    ),
+                ),
+                ("slices", models.IntegerField(blank=True, null=True)),
+                ("target_hg_push_ids", models.JSONField(default=list)),
+                ("scanned_push_count", models.IntegerField(default=0)),
+                (
+                    "status",
+                    models.IntegerField(
+                        choices=[(0, "Pending"), (1, "Synced"), (2, "Failed")],
+                        default=0,
+                    ),
+                ),
+                ("created", models.DateTimeField(auto_now_add=True)),
+                ("last_updated", models.DateTimeField(auto_now=True)),
+                (
+                    "job",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="backfill_requests",
+                        to="model.job",
+                    ),
+                ),
+                (
+                    "push",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="backfill_requests",
+                        to="model.push",
+                    ),
+                ),
+                (
+                    "repository",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="model.repository",
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "backfill_request",
+            },
+        ),
+        migrations.CreateModel(
+            name="BackfilledPush",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("decision_task_id", models.CharField(max_length=255)),
+                ("created", models.DateTimeField(auto_now_add=True)),
+                (
+                    "push",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="+",
+                        to="model.push",
+                    ),
+                ),
+                (
+                    "backfill_request",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="backfilled_pushes",
+                        to="perf.backfillrequest",
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "backfilled_push",
+            },
+        ),
+    ]

treeherder/perf/models.py

@@ -1257,3 +1257,49 @@ def deepgetattr(obj: object, attr_chain: str) -> object | None:
             f"Failed to access deeply nested attribute `{attr_chain}` on object of type {type(obj)}."
         )
         return None
+
+
+class BackfillRequest(models.Model):
+    job = models.ForeignKey(Job, on_delete=models.CASCADE, related_name="backfill_requests")
+    repository = models.ForeignKey(Repository, on_delete=models.CASCADE)
+    push = models.ForeignKey(Push, on_delete=models.CASCADE, related_name="backfill_requests")
+    bk_task_id = models.CharField(max_length=255, db_index=True)
+
+    STANDARD = 0
+    SLICED = 1
+    STRATEGY_STATUSES = (
+        (STANDARD, "Standard"),
+        (SLICED, "Sliced"),
+    )
+    label = models.CharField(max_length=255, null=True, blank=True)
+    strategy = models.IntegerField(choices=STRATEGY_STATUSES, null=True, blank=True)
+    slices = models.IntegerField(null=True, blank=True)
+    target_hg_push_ids = models.JSONField(default=list)
+    scanned_push_count = models.IntegerField(default=0)  # Standard: Depth size, Sliced: Gap Size
+
+    PENDING = 0
+    SYNCED = 1
+    FAILED = 2
+    STATUSES = (
+        (PENDING, "Pending"),
+        (SYNCED, "Synced"),
+        (FAILED, "Failed"),
+    )
+    status = models.IntegerField(choices=STATUSES, default=PENDING)
+    created = models.DateTimeField(auto_now_add=True)
+    last_updated = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        db_table = "backfill_request"
+
+
+class BackfilledPush(models.Model):
+    backfill_request = models.ForeignKey(
+        BackfillRequest, on_delete=models.CASCADE, related_name="backfilled_pushes"
+    )
+    push = models.ForeignKey(Push, on_delete=models.CASCADE, related_name="+")
+    decision_task_id = models.CharField(max_length=255)
+    created = models.DateTimeField(auto_now_add=True)
+
+    class Meta:
+        db_table = "backfilled_push"