Skip to content

Commit f7e1fff

Browse files
traskCopilot
andauthored
Fix collector completion gaps (#472)
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 64d32fdf-1825-4d3f-8f47-63bafb80a01a
1 parent 7c3d47e commit f7e1fff

3 files changed

Lines changed: 477 additions & 22 deletions

File tree

‎.github/scripts/github-actions-queue/collect.py‎

Lines changed: 41 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import re
1010
import threading
1111
import time
12+
from collections import Counter
1213
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
1314
from dataclasses import dataclass, field
1415
from datetime import UTC, datetime, timedelta
@@ -553,6 +554,7 @@ class CollectionResult:
553554
pause_reason: str | None
554555
completed_windows: int
555556
completed_repositories: int
557+
exclusions: Counter[str] = field(default_factory=Counter)
556558

557559

558560
@dataclass(frozen=True)
@@ -629,6 +631,7 @@ def _collect(
629631
raise ValueError("max_workers must be positive")
630632

631633
records: list[dict[str, Any]] = []
634+
exclusions: Counter[str] = Counter()
632635
completed_windows = 0
633636
completed_repositories = 0
634637

@@ -656,7 +659,7 @@ def _collect(
656659
if item["repository"] in current_repositories
657660
]
658661
pending_to_retry = _pending_runs_to_retry(state.pending_runs)
659-
self._collect_pending_runs(state, pending_to_retry, records)
662+
self._collect_pending_runs(state, pending_to_retry, records, exclusions)
660663

661664
while (
662665
_has_window(_parse_instant(state.cursor), boundary, direction)
@@ -686,8 +689,9 @@ def _collect(
686689
outcome = outcomes.get(repository)
687690
if outcome is None:
688691
continue
689-
repository_records, pending = outcome
692+
repository_records, pending, repository_exclusions = outcome
690693
records.extend(repository_records)
694+
exclusions.update(repository_exclusions)
691695
_merge_pending_runs(state, pending)
692696
completed.add(repository)
693697
completed_repositories += 1
@@ -704,6 +708,7 @@ def _collect(
704708
pause_reason=pause_reason,
705709
completed_windows=completed_windows,
706710
completed_repositories=completed_repositories,
711+
exclusions=exclusions,
707712
)
708713

709714
state.cursor = _format_instant(
@@ -720,6 +725,7 @@ def _collect(
720725
pause_reason=_pause_reason(error),
721726
completed_windows=completed_windows,
722727
completed_repositories=completed_repositories,
728+
exclusions=exclusions,
723729
)
724730

725731
return CollectionResult(
@@ -728,6 +734,7 @@ def _collect(
728734
pause_reason=None,
729735
completed_windows=completed_windows,
730736
completed_repositories=completed_repositories,
737+
exclusions=exclusions,
731738
)
732739

733740
def _collect_window_repositories(
@@ -737,12 +744,12 @@ def _collect_window_repositories(
737744
end: datetime,
738745
max_workers: int,
739746
) -> tuple[
740-
dict[str, tuple[list[dict[str, Any]], list[dict[str, Any]]]],
747+
dict[str, tuple[list[dict[str, Any]], list[dict[str, Any]], Counter[str]]],
741748
str | None,
742749
]:
743750
outcomes: dict[
744751
str,
745-
tuple[list[dict[str, Any]], list[dict[str, Any]]],
752+
tuple[list[dict[str, Any]], list[dict[str, Any]], Counter[str]],
746753
] = {}
747754
pause_reason = None
748755
with ThreadPoolExecutor(max_workers=max_workers) as executor:
@@ -772,6 +779,7 @@ def _collect_pending_runs(
772779
state: CollectionState,
773780
pending: list[dict[str, Any]],
774781
records: list[dict[str, Any]],
782+
exclusions: Counter[str],
775783
) -> None:
776784
for item in pending:
777785
try:
@@ -784,11 +792,12 @@ def _collect_pending_runs(
784792
[_pending_run_attempt(item, self._now())],
785793
)
786794
continue
787-
run_records, terminal = self._collect_completed_run(
795+
run_records, terminal, run_exclusions = self._collect_completed_run(
788796
item["repository"], run
789797
)
790798
if terminal:
791799
records.extend(run_records)
800+
exclusions.update(run_exclusions)
792801
_remove_pending_run(state, item)
793802
else:
794803
_merge_pending_runs(
@@ -810,16 +819,19 @@ def _collect_repository_window(
810819
repository: str,
811820
start: datetime,
812821
end: datetime,
813-
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
822+
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], Counter[str]]:
814823
records: list[dict[str, Any]] = []
815824
pending: list[dict[str, Any]] = []
825+
exclusions: Counter[str] = Counter()
816826
runs = self._client.list_workflow_runs(self._org, repository, start, end)
817827
for run in runs:
818828
if run.get("status") != "completed":
819829
pending.append(_pending_run(repository, run))
820830
continue
821831
try:
822-
run_records, terminal = self._collect_completed_run(repository, run)
832+
run_records, terminal, run_exclusions = self._collect_completed_run(
833+
repository, run
834+
)
823835
except ApiError as error:
824836
pending.append(
825837
_pending_run_failure(
@@ -831,26 +843,33 @@ def _collect_repository_window(
831843
continue
832844
if terminal:
833845
records.extend(run_records)
846+
exclusions.update(run_exclusions)
834847
else:
835848
pending.append(_pending_run(repository, run))
836-
return records, pending
849+
return records, pending, exclusions
837850

838851
def _collect_completed_run(
839852
self,
840853
repository: str,
841854
run: dict[str, Any],
842-
) -> tuple[list[dict[str, Any]], bool]:
843-
jobs = self._client.list_jobs(self._org, repository, run["id"])
855+
) -> tuple[list[dict[str, Any]], bool, Counter[str]]:
856+
all_jobs = self._client.list_jobs(self._org, repository, run["id"])
857+
jobs = [job for job in all_jobs if job.get("runner_name")]
844858
if any(not _job_is_terminal(job) for job in jobs):
845-
return [], False
846-
jobs = [job for job in jobs if job.get("runner_name")]
859+
return [], False, Counter()
860+
exclusions = Counter(
861+
runnerless_jobs=len(all_jobs) - len(jobs),
862+
runnerless_runs=int(bool(all_jobs) and not jobs),
863+
)
847864
collected_at = _format_instant(self._now())
848865
records = []
849866
for job in jobs:
850867
record = _job_record(self._org, repository, run, job, collected_at)
851868
if record["queue_seconds"] >= 0:
852869
records.append(record)
853-
return records, True
870+
else:
871+
exclusions["negative_queue_jobs"] += 1
872+
return records, True, exclusions
854873

855874

856875
def collect_live_and_backfill(
@@ -900,7 +919,7 @@ def collect_live_and_backfill(
900919

901920
def _job_is_terminal(job: dict[str, Any]) -> bool:
902921
return job.get("status") == "completed" or (
903-
job.get("status") == "in_progress"
922+
job.get("status") in ("in_progress", "queued")
904923
and bool(job.get("completed_at"))
905924
and job.get("conclusion") in TERMINAL_JOB_CONCLUSIONS
906925
)
@@ -1365,6 +1384,14 @@ def main() -> None:
13651384
),
13661385
"completed_windows": sum(result.completed_windows for result in results),
13671386
"cursor": state.cursor,
1387+
"exclusions": {
1388+
reason: sum(result.exclusions[reason] for result in results)
1389+
for reason in (
1390+
"runnerless_jobs",
1391+
"runnerless_runs",
1392+
"negative_queue_jobs",
1393+
)
1394+
},
13681395
"live": {
13691396
"completed_repositories": live_result.completed_repositories,
13701397
"completed_windows": live_result.completed_windows,

0 commit comments

Comments
 (0)