From 8d14b17f3344a9ca2f0c6c27200702317bfab257 Mon Sep 17 00:00:00 2001 From: Myeongjun Go Date: Fri, 3 Jul 2026 17:36:48 +0900 Subject: [PATCH] Bug 2051180 - Track which pushes are actually backfilled --- .../auto_perf_sheriffing/backfill_tracker.py | 129 ++++++++++++++++++ .../perf/auto_perf_sheriffing/sherlock.py | 14 ++ .../0078_backfillrequest_backfilledpush.py | 110 +++++++++++++++ treeherder/perf/models.py | 46 +++++++ 4 files changed, 299 insertions(+) create mode 100644 treeherder/perf/auto_perf_sheriffing/backfill_tracker.py create mode 100644 treeherder/perf/migrations/0078_backfillrequest_backfilledpush.py diff --git a/treeherder/perf/auto_perf_sheriffing/backfill_tracker.py b/treeherder/perf/auto_perf_sheriffing/backfill_tracker.py new file mode 100644 index 00000000000..0b380371dc8 --- /dev/null +++ b/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), + } diff --git a/treeherder/perf/auto_perf_sheriffing/sherlock.py b/treeherder/perf/auto_perf_sheriffing/sherlock.py index 8a3071985bb..08dd9ca3eb6 100644 --- a/treeherder/perf/auto_perf_sheriffing/sherlock.py +++ b/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,8 +201,10 @@ 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." @@ -204,6 +212,12 @@ def _backfill_record(self, record: BackfillRecord, left: int) -> tuple[int, int] 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}") diff --git a/treeherder/perf/migrations/0078_backfillrequest_backfilledpush.py b/treeherder/perf/migrations/0078_backfillrequest_backfilledpush.py new file mode 100644 index 00000000000..1ca13b92fb6 --- /dev/null +++ b/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", + }, + ), + ] diff --git a/treeherder/perf/models.py b/treeherder/perf/models.py index 3acf1fa2f98..6ba291c2611 100644 --- a/treeherder/perf/models.py +++ b/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"