diff --git a/.github/workflows/amd-ci-job-monitor.yml b/.github/workflows/amd-ci-job-monitor.yml index 87a195470..cbb8798b1 100644 --- a/.github/workflows/amd-ci-job-monitor.yml +++ b/.github/workflows/amd-ci-job-monitor.yml @@ -20,10 +20,8 @@ on: type: string jobs: - # Single job filter mode - custom-report: - name: Custom Job Report - if: ${{ inputs.job_filter }} + fetch-actions-data: + name: Fetch Actions Snapshot runs-on: ubuntu-latest env: GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} @@ -39,6 +37,55 @@ jobs: - name: Install dependencies run: pip install tabulate + - name: Select workflows for snapshot + id: select-workflows + run: | + if [[ -n "${{ inputs.job_filter }}" ]]; then + echo "workflows=pr-test-amd.yml" >> "$GITHUB_OUTPUT" + else + echo "workflows=pr-test-amd.yml,nightly-test-amd.yml,pr-test-amd-rocm720.yml,nightly-test-amd-rocm720.yml" >> "$GITHUB_OUTPUT" + fi + + - name: Fetch Actions data snapshot + timeout-minutes: 30 + run: | + python scripts/ci/utils/query_job_status.py \ + --repo ${{ github.repository }} \ + --workflow "${{ steps.select-workflows.outputs.workflows }}" \ + --hours ${{ inputs.hours || '24' }} \ + --dump-data-file actions-job-snapshot.json + + - name: Upload Actions data snapshot + uses: actions/upload-artifact@v4 + with: + name: actions-job-snapshot + path: actions-job-snapshot.json + if-no-files-found: error + + # Single job filter mode + custom-report: + name: Custom Job Report + if: ${{ inputs.job_filter }} + needs: fetch-actions-data + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.10' + + - name: Install dependencies + run: pip install tabulate + + - name: Download Actions data snapshot + uses: actions/download-artifact@v4 + with: + name: actions-job-snapshot + path: ci-data + - name: Generate Custom Job Report timeout-minutes: 30 run: | @@ -47,6 +94,7 @@ jobs: --job "${{ inputs.job_filter }}" \ --workflow "pr-test-amd.yml" \ --hours ${{ inputs.hours || '24' }} \ + --input-data-file ci-data/actions-job-snapshot.json \ --summary # Parse workflow files to get job names dynamically @@ -57,6 +105,8 @@ jobs: outputs: pr_jobs: ${{ steps.parse.outputs.pr_jobs }} nightly_jobs: ${{ steps.parse.outputs.nightly_jobs }} + pr_rocm720_jobs: ${{ steps.parse.outputs.pr_rocm720_jobs }} + nightly_rocm720_jobs: ${{ steps.parse.outputs.nightly_rocm720_jobs }} steps: - name: Checkout code uses: actions/checkout@v4 @@ -80,18 +130,32 @@ jobs: echo "nightly_jobs=$nightly_jobs" >> $GITHUB_OUTPUT echo "Nightly jobs: $nightly_jobs" + # Parse pr-test-amd-rocm720.yml (exclude utility jobs) + # Excluded: call-gate, check-changes, pr-test-amd-finish, cancel, check-all-jobs + pr_rocm720_jobs=$(yq -r '.jobs | keys | .[]' .github/workflows/pr-test-amd-rocm720.yml | \ + grep -v -E '^(call-gate|check-changes|pr-test-amd-finish|cancel|check-all-jobs)$' | \ + jq -R -s -c 'split("\n") | map(select(length > 0))') + echo "pr_rocm720_jobs=$pr_rocm720_jobs" >> $GITHUB_OUTPUT + echo "PR ROCm 7.2 jobs: $pr_rocm720_jobs" + + # Parse nightly-test-amd-rocm720.yml (exclude utility jobs) + # Excluded: check-all-jobs + nightly_rocm720_jobs=$(yq -r '.jobs | keys | .[]' .github/workflows/nightly-test-amd-rocm720.yml | \ + grep -v -E '^(check-all-jobs)$' | \ + jq -R -s -c 'split("\n") | map(select(length > 0))') + echo "nightly_rocm720_jobs=$nightly_rocm720_jobs" >> $GITHUB_OUTPUT + echo "Nightly ROCm 7.2 jobs: $nightly_rocm720_jobs" + # PR CI reports using dynamic matrix pr-ci-reports: name: PR - ${{ matrix.job_name }} - needs: parse-workflows + needs: [parse-workflows, fetch-actions-data] if: ${{ !inputs.job_filter }} runs-on: ubuntu-latest strategy: fail-fast: false matrix: job_name: ${{ fromJson(needs.parse-workflows.outputs.pr_jobs) }} - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} steps: - name: Checkout code uses: actions/checkout@v4 @@ -104,6 +168,12 @@ jobs: - name: Install dependencies run: pip install tabulate + - name: Download Actions data snapshot + uses: actions/download-artifact@v4 + with: + name: actions-job-snapshot + path: ci-data + - name: Generate Report timeout-minutes: 15 run: | @@ -112,20 +182,19 @@ jobs: --job "${{ matrix.job_name }}" \ --workflow "pr-test-amd.yml" \ --hours ${{ inputs.hours || '24' }} \ + --input-data-file ci-data/actions-job-snapshot.json \ --summary # Nightly AMD test reports using dynamic matrix nightly-reports: name: Nightly - ${{ matrix.job_name }} - needs: parse-workflows + needs: [parse-workflows, fetch-actions-data] if: ${{ !inputs.job_filter }} runs-on: ubuntu-latest strategy: fail-fast: false matrix: job_name: ${{ fromJson(needs.parse-workflows.outputs.nightly_jobs) }} - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} steps: - name: Checkout code uses: actions/checkout@v4 @@ -138,6 +207,12 @@ jobs: - name: Install dependencies run: pip install tabulate + - name: Download Actions data snapshot + uses: actions/download-artifact@v4 + with: + name: actions-job-snapshot + path: ci-data + - name: Generate Nightly Report timeout-minutes: 15 run: | @@ -146,4 +221,118 @@ jobs: --job "${{ matrix.job_name }}" \ --workflow "nightly-test-amd.yml" \ --hours ${{ inputs.hours || '24' }} \ + --input-data-file ci-data/actions-job-snapshot.json \ + --summary + + # PR ROCm 7.2 CI reports using dynamic matrix + pr-rocm720-ci-reports: + name: PR ROCm720 - ${{ matrix.job_name }} + needs: [parse-workflows, fetch-actions-data] + if: ${{ !inputs.job_filter }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + job_name: ${{ fromJson(needs.parse-workflows.outputs.pr_rocm720_jobs) }} + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.10' + + - name: Install dependencies + run: pip install tabulate + + - name: Download Actions data snapshot + uses: actions/download-artifact@v4 + with: + name: actions-job-snapshot + path: ci-data + + - name: Generate PR ROCm 7.2 Report + timeout-minutes: 15 + run: | + python scripts/ci/utils/query_job_status.py \ + --repo ${{ github.repository }} \ + --job "${{ matrix.job_name }}" \ + --workflow "pr-test-amd-rocm720.yml" \ + --hours ${{ inputs.hours || '24' }} \ + --input-data-file ci-data/actions-job-snapshot.json \ + --summary + + # Nightly ROCm 7.2 reports using dynamic matrix + nightly-rocm720-reports: + name: Nightly ROCm720 - ${{ matrix.job_name }} + needs: [parse-workflows, fetch-actions-data] + if: ${{ !inputs.job_filter }} + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + job_name: ${{ fromJson(needs.parse-workflows.outputs.nightly_rocm720_jobs) }} + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.10' + + - name: Install dependencies + run: pip install tabulate + + - name: Download Actions data snapshot + uses: actions/download-artifact@v4 + with: + name: actions-job-snapshot + path: ci-data + + - name: Generate Nightly ROCm 7.2 Report + timeout-minutes: 15 + run: | + python scripts/ci/utils/query_job_status.py \ + --repo ${{ github.repository }} \ + --job "${{ matrix.job_name }}" \ + --workflow "nightly-test-amd-rocm720.yml" \ + --hours ${{ inputs.hours || '24' }} \ + --input-data-file ci-data/actions-job-snapshot.json \ + --summary + + # Runner fleet report - cross-workflow runner analytics in a single pass + runner-fleet-report: + name: Runner Fleet Report + if: ${{ !inputs.job_filter }} + needs: fetch-actions-data + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.10' + + - name: Install dependencies + run: pip install tabulate + + - name: Download Actions data snapshot + uses: actions/download-artifact@v4 + with: + name: actions-job-snapshot + path: ci-data + + - name: Generate Runner Fleet Report + timeout-minutes: 30 + run: | + python scripts/ci/utils/query_job_status.py \ + --repo ${{ github.repository }} \ + --runner-report \ + --workflow "pr-test-amd.yml,nightly-test-amd.yml,pr-test-amd-rocm720.yml,nightly-test-amd-rocm720.yml" \ + --hours ${{ inputs.hours || '24' }} \ + --input-data-file ci-data/actions-job-snapshot.json \ --summary diff --git a/scripts/ci/utils/query_job_status.py b/scripts/ci/utils/query_job_status.py index 9a5e8a128..39028af2d 100755 --- a/scripts/ci/utils/query_job_status.py +++ b/scripts/ci/utils/query_job_status.py @@ -1,11 +1,17 @@ #!/usr/bin/env python3 """ -Query GitHub Actions job status for specific jobs. +Query GitHub Actions job status for specific jobs or generate runner fleet reports. Usage: + # Per-job reports (original mode) python scripts/ci/utils/query_job_status.py --job "stage-c-test-large-8-gpu-amd-mi35x" python scripts/ci/utils/query_job_status.py --job "stage-c-test-large-8-gpu-amd-mi35x" --hours 48 - python scripts/ci/utils/query_job_status.py --job "AMD" --workflow pr-test-amd.yml + python scripts/ci/utils/query_job_status.py --job "stage-c-test-large-8-gpu-amd-mi35x" --workflow "pr-test-amd.yml" --input-data-file actions-job-snapshot.json --summary + + # Runner fleet report (cross-workflow runner analytics) + python scripts/ci/utils/query_job_status.py --runner-report --workflow "pr-test-amd.yml,nightly-test-amd.yml" --hours 24 + python scripts/ci/utils/query_job_status.py --runner-report --workflow "pr-test-amd.yml,nightly-test-amd.yml,pr-test-amd-rocm720.yml,nightly-test-amd-rocm720.yml" --summary + python scripts/ci/utils/query_job_status.py --workflow "pr-test-amd.yml,nightly-test-amd.yml,pr-test-amd-rocm720.yml,nightly-test-amd-rocm720.yml" --dump-data-file actions-job-snapshot.json Requirements: pip install tabulate @@ -17,7 +23,7 @@ import os import subprocess import sys from datetime import datetime, timedelta, timezone -from typing import Optional +from typing import Any, Optional try: from tabulate import tabulate @@ -76,6 +82,77 @@ def run_gh_command(args: list[str]) -> dict: return json.loads(result.stdout) +def is_rate_limit_error(error: str) -> bool: + """Check whether an API error was caused by GitHub rate limiting.""" + return "rate limit exceeded" in error.lower() + + +def _new_workflow_fetch_stats(workflow: str) -> dict[str, Any]: + """Create an empty metadata bucket for a workflow snapshot.""" + return { + "workflow": workflow, + "total_runs_seen": 0, + "runs_with_jobs": 0, + "skipped_runs": 0, + "skipped_runs_rate_limit": 0, + "jobs_collected": 0, + } + + +def _new_fetch_metadata(repo: str, workflows: list[str], hours: int) -> dict[str, Any]: + """Create the fetch metadata container stored alongside snapshot jobs.""" + return { + "repo": repo, + "hours": hours, + "requested_workflows": workflows, + "total_runs_seen": 0, + "runs_with_jobs": 0, + "jobs_collected": 0, + "skipped_runs": [], + "workflow_fetch_failures": [], + "workflow_stats": { + workflow: _new_workflow_fetch_stats(workflow) for workflow in workflows + }, + } + + +def _record_workflow_fetch_failure( + fetch_metadata: dict[str, Any], workflow: str, error: str +) -> None: + """Record a workflow-level failure while listing workflow runs.""" + fetch_metadata["workflow_fetch_failures"].append( + { + "workflow": workflow, + "error": error.strip(), + "reason": "rate_limit" if is_rate_limit_error(error) else "api_error", + } + ) + + +def _record_skipped_run( + fetch_metadata: dict[str, Any], workflow: str, run: dict, error: str +) -> None: + """Record a run whose jobs could not be fetched.""" + workflow_stats = fetch_metadata["workflow_stats"].setdefault( + workflow, _new_workflow_fetch_stats(workflow) + ) + workflow_stats["skipped_runs"] += 1 + if is_rate_limit_error(error): + workflow_stats["skipped_runs_rate_limit"] += 1 + + fetch_metadata["skipped_runs"].append( + { + "workflow": workflow, + "run_id": run["id"], + "created_at": run.get("created_at", ""), + "status": run.get("status", "unknown"), + "conclusion": run.get("conclusion") or "-", + "reason": "rate_limit" if is_rate_limit_error(error) else "api_error", + "error": error.strip(), + } + ) + + def parse_time(time_str: str) -> Optional[datetime]: """Parse ISO timestamp to datetime.""" if not time_str: @@ -150,66 +227,131 @@ def get_pr_number_from_run(run: dict) -> Optional[int]: return None -def query_jobs( - repo: str, +def _job_name_matches_filter(job_name: str, job_filter: str) -> bool: + """Check whether a job name matches the report filter prefix.""" + job_name_lower = job_name.lower() + filter_lower = job_filter.lower() + if not job_name_lower.startswith(filter_lower): + return False + if len(job_name_lower) > len(filter_lower): + next_char = job_name_lower[len(filter_lower)] + if next_char not in (" ", "("): + return False + return True + + +def filter_jobs( + jobs: list[dict], job_filter: str, workflow: str = None, - hours: int = 24, status_filter: str = None, ) -> list[dict]: - """Query jobs matching the filter.""" + """Filter a prefetched job list for a specific report target.""" + results = [] + for job in jobs: + if workflow and job.get("workflow") != workflow: + continue + if not _job_name_matches_filter(job.get("job_name", ""), job_filter): + continue + if status_filter and job.get("status") != status_filter: + continue + results.append(job) + return results - print(f"Fetching workflow runs from last {hours} hours...", file=sys.stderr) - runs = get_workflow_runs(repo, workflow, hours) - print(f"Found {len(runs)} workflow runs", file=sys.stderr) + +def save_snapshot(path: str, snapshot: dict[str, Any]) -> None: + """Persist a prefetched Actions snapshot to disk.""" + with open(path, "w") as f: + json.dump(snapshot, f, indent=2) + + +def load_snapshot(path: str) -> dict[str, Any]: + """Load a previously saved Actions snapshot from disk.""" + with open(path) as f: + snapshot = json.load(f) + if "jobs" not in snapshot: + raise ValueError(f"Snapshot file {path} is missing the 'jobs' field") + return snapshot + + +def fetch_all_jobs_snapshot( + repo: str, + workflows: list[str], + hours: int = 24, +) -> dict[str, Any]: + """Fetch jobs once and store enough metadata to detect incomplete data.""" + fetch_metadata = _new_fetch_metadata(repo, workflows, hours) + all_runs = [] + + for workflow in workflows: + print(f"Fetching runs for {workflow}...", file=sys.stderr) + try: + runs = get_workflow_runs(repo, workflow, hours) + except Exception as e: + error = str(e) + print( + f"Warning: Failed to list runs for workflow {workflow}: {error}", + file=sys.stderr, + ) + _record_workflow_fetch_failure(fetch_metadata, workflow, error) + continue + + print(f" Found {len(runs)} runs for {workflow}", file=sys.stderr) + fetch_metadata["workflow_stats"][workflow]["total_runs_seen"] = len(runs) + for run in runs: + run["_workflow"] = workflow + all_runs.extend(runs) + + seen_run_ids = set() + unique_runs = [] + for run in all_runs: + if run["id"] not in seen_run_ids: + seen_run_ids.add(run["id"]) + unique_runs.append(run) + + fetch_metadata["total_runs_seen"] = len(unique_runs) + print(f"Total unique workflow runs: {len(unique_runs)}", file=sys.stderr) results = [] - total_runs = len(runs) + total_runs = len(unique_runs) - for i, run in enumerate(runs): + for i, run in enumerate(unique_runs): if (i + 1) % 20 == 0: print(f"Processing run {i+1}/{total_runs}...", file=sys.stderr) + workflow_name = run.get("_workflow", "-") try: jobs = get_jobs_for_run(repo, run["id"]) except Exception as e: + error = str(e) print( - f"Warning: Failed to get jobs for run {run['id']}: {e}", file=sys.stderr + f"Warning: Failed to get jobs for run {run['id']}: {error}", + file=sys.stderr, ) + _record_skipped_run(fetch_metadata, workflow_name, run, error) continue + workflow_stats = fetch_metadata["workflow_stats"].setdefault( + workflow_name, _new_workflow_fetch_stats(workflow_name) + ) + workflow_stats["runs_with_jobs"] += 1 + fetch_metadata["runs_with_jobs"] += 1 + pr_number = get_pr_number_from_run(run) branch = run.get("head_branch", "") run_status = run.get("status", "unknown") run_conclusion = run.get("conclusion") or "-" + jobs_added = 0 for job in jobs: job_name = job.get("name", "") - - # Filter by job name - # Use prefix matching to avoid e.g. "stage-c-test-large-8-gpu-amd" - # also matching "stage-c-test-large-8-gpu-amd-mi35x" - job_name_lower = job_name.lower() - filter_lower = job_filter.lower() - if not job_name_lower.startswith(filter_lower): - continue - # If there are characters after the filter, ensure it's not a - # continuation of the base job name (e.g., "-mi35x") - if len(job_name_lower) > len(filter_lower): - next_char = job_name_lower[len(filter_lower)] - if next_char not in (" ", "("): - continue - - # Filter by status if specified - if status_filter and job.get("status") != status_filter: - continue - job_status = job.get("status", "unknown") runner_name = job.get("runner_name") or "-" + labels = job.get("labels", []) + + if len(labels) == 1 and labels[0] == "ubuntu-latest": + continue - # Detect stuck/ghost jobs: - # - Job is in_progress but no runner assigned - # - Job is in_progress but workflow run is cancelled/completed is_stuck = False if job_status == "in_progress": if runner_name == "-": @@ -229,6 +371,8 @@ def query_jobs( "started_at": job.get("started_at", ""), "completed_at": job.get("completed_at", ""), "runner_name": runner_name, + "labels": labels, + "runner_group_name": job.get("runner_group_name") or "-", "run_id": run["id"], "run_status": run_status, "run_conclusion": run_conclusion, @@ -236,10 +380,49 @@ def query_jobs( "branch": branch, "html_url": job.get("html_url", ""), "is_stuck": is_stuck, + "workflow": workflow_name, } ) + jobs_added += 1 - return results + workflow_stats["jobs_collected"] += jobs_added + + fetch_metadata["jobs_collected"] = len(results) + return { + "snapshot_version": 1, + "repo": repo, + "hours": hours, + "workflows": workflows, + "generated_at": datetime.now(timezone.utc).isoformat(), + "jobs": results, + "fetch_metadata": fetch_metadata, + } + + +def query_jobs( + repo: str, + job_filter: str, + workflow: str = None, + hours: int = 24, + status_filter: str = None, +) -> list[dict]: + """Query jobs matching the filter.""" + snapshot = fetch_all_jobs_snapshot(repo, [workflow], hours) + return filter_jobs(snapshot["jobs"], job_filter, workflow, status_filter) + + +def query_all_jobs( + repo: str, + workflows: list[str], + hours: int = 24, +) -> list[dict]: + """Query all jobs across multiple workflows for fleet-level analysis. + + Unlike query_jobs(), this does NOT filter by job name and collects + everything in a single pass -- ideal for runner-centric analytics. + Jobs on ubuntu-latest are excluded since those are utility jobs. + """ + return fetch_all_jobs_snapshot(repo, workflows, hours)["jobs"] def calculate_duration(started_at: str, completed_at: str) -> str: @@ -320,6 +503,253 @@ def calculate_queue_time( return f"{minutes}m{seconds}s" +# --------------------------------------------------------------------------- +# Runner fleet analytics functions +# --------------------------------------------------------------------------- + + +def _format_duration_seconds(seconds: Optional[float]) -> str: + """Format seconds into human-readable duration string.""" + if seconds is None or seconds < 0: + return "-" + total_seconds = int(seconds) + minutes = total_seconds // 60 + secs = total_seconds % 60 + if minutes >= 60: + hours = minutes // 60 + minutes = minutes % 60 + return f"{hours}h{minutes}m" + return f"{minutes}m{secs}s" + + +def _get_runner_label(job: dict) -> str: + """Extract the primary runner label from a job's labels list.""" + labels = job.get("labels", []) + if not labels: + return "unknown" + for label in labels: + if label.startswith("linux-mi"): + return label + return labels[0] + + +def _percentile(data: list[float], p: int) -> Optional[float]: + """Return a percentile from an already sorted or unsorted numeric list.""" + if not data: + return None + sorted_data = sorted(data) + idx = min(int(len(sorted_data) * p / 100), len(sorted_data) - 1) + return sorted_data[idx] + + +def _average(data: list[float]) -> Optional[float]: + """Return the average of a numeric list when samples exist.""" + if not data: + return None + return sum(data) / len(data) + + +def _queue_time_seconds(job: dict) -> Optional[float]: + """Extract queue time in seconds for a job if both timestamps exist.""" + created = parse_time(job.get("created_at", "")) + started = parse_time(job.get("started_at", "")) + if not (created and started): + return None + + queue_seconds = (started - created).total_seconds() + if queue_seconds < 0: + return None + return queue_seconds + + +def _build_queue_distribution(queue_times: list[float]) -> dict[str, Any]: + """Build queue time buckets and percentile stats for one sample set.""" + if not queue_times: + return {"buckets": [], "p50": None, "p90": None, "p99": None, "total": 0} + + sorted_queue_times = sorted(queue_times) + bucket_defs = [ + ("< 1 min", 0, 60), + ("1-5 min", 60, 300), + ("5-15 min", 300, 900), + ("15-30 min", 900, 1800), + ("30-60 min", 1800, 3600), + ("> 60 min", 3600, float("inf")), + ] + + total = len(sorted_queue_times) + buckets = [] + for label, lo, hi in bucket_defs: + count = sum(1 for qt in sorted_queue_times if lo <= qt < hi) + pct = count / total * 100 if total > 0 else 0 + buckets.append({"range": label, "count": count, "percentage": round(pct, 1)}) + + return { + "buckets": buckets, + "p50": _percentile(sorted_queue_times, 50), + "p90": _percentile(sorted_queue_times, 90), + "p99": _percentile(sorted_queue_times, 99), + "total": total, + } + + +def analyze_concurrency(jobs: list[dict], report_time: datetime = None) -> dict: + """Analyze concurrent runner usage per runner label. + + Uses an event-sweep algorithm: for each job that ran, create +1 event + at started_at and -1 event at completed_at, then sweep through sorted + events tracking the concurrent count. + """ + if report_time is None: + report_time = datetime.now(timezone.utc) + + label_jobs: dict[str, list[dict]] = {} + for job in jobs: + label = _get_runner_label(job) + label_jobs.setdefault(label, []).append(job) + + results = {} + for label in sorted(label_jobs): + pool_jobs = label_jobs[label] + events: list[tuple[datetime, int]] = [] + queue_times: list[float] = [] + durations: list[float] = [] + + for job in pool_jobs: + started = parse_time(job.get("started_at", "")) + completed = parse_time(job.get("completed_at", "")) + + if started and completed: + events.append((started, +1)) + events.append((completed, -1)) + durations.append((completed - started).total_seconds()) + elif started: + events.append((started, +1)) + events.append((report_time, -1)) + durations.append((report_time - started).total_seconds()) + + qt = _queue_time_seconds(job) + if qt is not None: + queue_times.append(qt) + + if not events: + results[label] = { + "peak": 0, + "avg_concurrent": 0.0, + "total_jobs": len(pool_jobs), + "avg_queue_seconds": _average(queue_times), + "p50_queue_seconds": _percentile(queue_times, 50), + "p99_queue_seconds": _percentile(queue_times, 99), + "avg_duration_seconds": _average(durations), + } + continue + + events.sort(key=lambda x: (x[0], x[1])) + concurrent = 0 + peak = 0 + time_weighted_sum = 0.0 + total_time = 0.0 + prev_time = events[0][0] + + for ts, delta in events: + if prev_time and concurrent > 0: + dt = (ts - prev_time).total_seconds() + time_weighted_sum += concurrent * dt + total_time += dt + concurrent += delta + peak = max(peak, concurrent) + prev_time = ts + + avg_concurrent = time_weighted_sum / total_time if total_time > 0 else 0 + avg_queue = _average(queue_times) + avg_duration = _average(durations) + + results[label] = { + "peak": peak, + "avg_concurrent": round(avg_concurrent, 1), + "total_jobs": len(pool_jobs), + "avg_queue_seconds": avg_queue, + "p50_queue_seconds": _percentile(queue_times, 50), + "p99_queue_seconds": _percentile(queue_times, 99), + "avg_duration_seconds": avg_duration, + } + + return results + + +def analyze_busy_periods(jobs: list[dict]) -> list[dict]: + """Analyze job activity by hour of day (UTC). + + Buckets jobs by the UTC hour they started and computes avg queue time. + Classifies each hour as Quiet / Moderate / Busy / Peak relative to the + busiest hour. + """ + hourly: dict[int, dict] = { + h: {"jobs_started": 0, "queue_times": []} for h in range(24) + } + + for job in jobs: + started = parse_time(job.get("started_at", "")) + created = parse_time(job.get("created_at", "")) + + if started: + hour = started.astimezone(timezone.utc).hour + hourly[hour]["jobs_started"] += 1 + + if created: + qt = (started - created).total_seconds() + if qt >= 0: + hourly[hour]["queue_times"].append(qt) + + max_jobs = max((v["jobs_started"] for v in hourly.values()), default=1) or 1 + + results = [] + for hour in range(24): + data = hourly[hour] + avg_queue = ( + sum(data["queue_times"]) / len(data["queue_times"]) + if data["queue_times"] + else 0 + ) + ratio = data["jobs_started"] / max_jobs + if ratio >= 0.75: + load = "Peak" + elif ratio >= 0.5: + load = "Busy" + elif ratio >= 0.25: + load = "Moderate" + else: + load = "Quiet" + + results.append( + { + "hour": hour, + "hour_label": f"{hour:02d}:00-{(hour + 1) % 24:02d}:00", + "jobs_started": data["jobs_started"], + "avg_queue_seconds": avg_queue, + "load": load, + } + ) + + return results + + +def analyze_queue_distribution(jobs: list[dict]) -> dict: + """Analyze queue time distribution per runner label.""" + queue_times_by_label: dict[str, list[float]] = {} + for job in jobs: + queue_seconds = _queue_time_seconds(job) + if queue_seconds is None: + continue + label = _get_runner_label(job) + queue_times_by_label.setdefault(label, []).append(queue_seconds) + + return { + label: _build_queue_distribution(queue_times) + for label, queue_times in sorted(queue_times_by_label.items()) + } + + def process_results( results: list[dict], repo: str, report_time: datetime = None ) -> dict: @@ -433,6 +863,104 @@ def process_results( } +def summarize_fetch_metadata( + fetch_metadata: Optional[dict[str, Any]], workflows: list[str] = None +) -> Optional[dict[str, Any]]: + """Summarize snapshot completeness for the workflows relevant to a report.""" + if not fetch_metadata: + return None + + workflow_filter = ( + set(workflows) + if workflows + else set(fetch_metadata.get("requested_workflows", [])) + ) + workflow_stats = fetch_metadata.get("workflow_stats", {}) + if not workflow_filter: + workflow_filter = set(workflow_stats) + + relevant_stats = [ + workflow_stats[workflow] + for workflow in workflow_filter + if workflow in workflow_stats + ] + relevant_skipped_runs = [ + run + for run in fetch_metadata.get("skipped_runs", []) + if run.get("workflow") in workflow_filter + ] + relevant_workflow_failures = [ + failure + for failure in fetch_metadata.get("workflow_fetch_failures", []) + if failure.get("workflow") in workflow_filter + ] + + skipped_run_rate_limit = sum( + 1 for run in relevant_skipped_runs if run.get("reason") == "rate_limit" + ) + workflow_failure_rate_limit = sum( + 1 + for failure in relevant_workflow_failures + if failure.get("reason") == "rate_limit" + ) + + return { + "known_runs": sum(stat.get("total_runs_seen", 0) for stat in relevant_stats), + "runs_with_jobs": sum(stat.get("runs_with_jobs", 0) for stat in relevant_stats), + "jobs_collected": sum(stat.get("jobs_collected", 0) for stat in relevant_stats), + "skipped_runs": relevant_skipped_runs, + "workflow_failures": relevant_workflow_failures, + "skipped_run_rate_limit": skipped_run_rate_limit, + "workflow_failure_rate_limit": workflow_failure_rate_limit, + "incomplete": bool(relevant_skipped_runs or relevant_workflow_failures), + } + + +def append_fetch_metadata_notice( + lines: list[str], + fetch_metadata: Optional[dict[str, Any]], + workflows: list[str] = None, +) -> None: + """Append a markdown notice when the report is based on incomplete data.""" + summary = summarize_fetch_metadata(fetch_metadata, workflows) + if not summary or not summary["incomplete"]: + return + + skipped_runs = summary["skipped_runs"] + workflow_failures = summary["workflow_failures"] + other_skipped = len(skipped_runs) - summary["skipped_run_rate_limit"] + other_workflow_failures = ( + len(workflow_failures) - summary["workflow_failure_rate_limit"] + ) + + lines.append( + "> **Data completeness:** Incomplete. GitHub API rate limit and/or fetch errors prevented a full dataset." + ) + if summary["known_runs"] > 0: + lines.append( + f"> Successfully fetched jobs for **{summary['runs_with_jobs']}/{summary['known_runs']}** known runs in scope. Missing runs: **{len(skipped_runs)}** (rate limit: {summary['skipped_run_rate_limit']}, other API errors: {other_skipped})." + ) + + if workflow_failures: + workflow_names = ", ".join( + f"`{failure['workflow']}`" for failure in workflow_failures + ) + lines.append( + f"> Could not list workflow runs for {workflow_names}. Missing run count is unknown for those workflows (rate limit: {summary['workflow_failure_rate_limit']}, other API errors: {other_workflow_failures})." + ) + + if skipped_runs: + skipped_ids = ", ".join(f"`{run['run_id']}`" for run in skipped_runs[:10]) + remaining = len(skipped_runs) - 10 + suffix = f", and {remaining} more" if remaining > 0 else "" + lines.append(f"> Missing run IDs: {skipped_ids}{suffix}.") + + lines.append( + "> Missing job counts inside skipped runs are unknown because GitHub did not return those run job lists." + ) + lines.append("") + + def print_table( results: list[dict], repo: str, generated_time: str, report_time: datetime = None ): @@ -621,6 +1149,8 @@ def format_markdown( hours: int, generated_time: str, report_time: datetime = None, + fetch_metadata: dict[str, Any] = None, + workflow: str = None, ) -> str: """Format results as markdown for GitHub Actions summary.""" lines = [] @@ -634,6 +1164,9 @@ def format_markdown( lines.append("") lines.append("> **Note:** All times are displayed in UTC") lines.append("") + append_fetch_metadata_notice( + lines, fetch_metadata, [workflow] if workflow else None + ) if not results: lines.append("> No jobs found matching the filter.") @@ -795,11 +1328,192 @@ def format_markdown( return "\n".join(lines) -def main(): - # Check gh CLI availability before proceeding - if not check_gh_cli_available(): - sys.exit(1) +def format_runner_report_markdown( + jobs: list[dict], + workflows: list[str], + hours: int, + generated_time: str, + report_time: datetime = None, + fetch_metadata: dict[str, Any] = None, +) -> str: + """Format runner fleet analytics as markdown for GitHub Actions summary.""" + if report_time is None: + report_time = datetime.now(timezone.utc) + lines: list[str] = [] + + # Header + lines.append("# CI Runner Fleet Report") + lines.append("") + lines.append(f"**Workflows:** {', '.join(f'`{w}`' for w in workflows)}") + lines.append(f"**Time window:** Last {hours} hours") + lines.append(f"**Generated:** {generated_time} UTC") + lines.append(f"**Total jobs analyzed:** {len(jobs)}") + lines.append("") + lines.append("> All times are in UTC. Jobs on `ubuntu-latest` are excluded.") + lines.append("") + append_fetch_metadata_notice(lines, fetch_metadata, workflows) + + if not jobs: + lines.append("> No self-hosted runner jobs found in the time window.") + return "\n".join(lines) + + # --- Fleet Overview --- + unique_labels = {_get_runner_label(j) for j in jobs} + completed_jobs = [j for j in jobs if j.get("status") == "completed"] + lines.append("## Fleet Overview") + lines.append("") + lines.append("| Metric | Value |") + lines.append("|--------|-------|") + lines.append(f"| Total runner labels seen | {len(unique_labels)} |") + lines.append(f"| Total jobs analyzed | {len(jobs)} |") + lines.append(f"| Completed jobs | {len(completed_jobs)} |") + lines.append(f"| Time window | {hours}h |") + lines.append("") + + # --- Concurrency by Runner Label --- + concurrency = analyze_concurrency(jobs, report_time) + if concurrency: + lines.append("## Concurrency by Runner Label") + lines.append("") + lines.append( + "| Runner Label | Peak Concurrent | Avg Concurrent | Total Jobs | Avg Queue | P50 Queue | P99 Queue | Avg Duration |" + ) + lines.append( + "|-------------|----------------|---------------|-----------|-----------|-----------|-----------|-------------|" + ) + for label in sorted(concurrency, key=lambda k: -concurrency[k]["peak"]): + c = concurrency[label] + lines.append( + f"| `{label}` | **{c['peak']}** | {c['avg_concurrent']} " + f"| {c['total_jobs']} " + f"| {_format_duration_seconds(c['avg_queue_seconds'])} " + f"| {_format_duration_seconds(c['p50_queue_seconds'])} " + f"| {_format_duration_seconds(c['p99_queue_seconds'])} " + f"| {_format_duration_seconds(c['avg_duration_seconds'])} |" + ) + lines.append("") + + # --- Busy Periods --- + busy_periods = analyze_busy_periods(jobs) + if busy_periods: + lines.append("## Busy Periods (UTC)") + lines.append("") + lines.append("| Hour (UTC) | Jobs Started | Avg Queue Time | Load |") + lines.append("|-----------|-------------|---------------|------|") + for bp in busy_periods: + if bp["jobs_started"] == 0: + continue + load_display = ( + f"**{bp['load']}**" if bp["load"] in ("Peak", "Busy") else bp["load"] + ) + lines.append( + f"| {bp['hour_label']} | {bp['jobs_started']} " + f"| {_format_duration_seconds(bp['avg_queue_seconds'])} " + f"| {load_display} |" + ) + lines.append("") + + peak_hours = [bp for bp in busy_periods if bp["load"] == "Peak"] + quiet_hours = [ + bp + for bp in busy_periods + if bp["load"] == "Quiet" and bp["jobs_started"] > 0 + ] + if peak_hours: + labels = ", ".join(bp["hour_label"] for bp in peak_hours) + lines.append(f"> **Peak hours:** {labels}") + lines.append("") + if quiet_hours: + labels = ", ".join(bp["hour_label"] for bp in quiet_hours) + lines.append(f"> **Quiet hours:** {labels}") + lines.append("") + + # --- Queue Time Distribution --- + queue_dist = analyze_queue_distribution(jobs) + if queue_dist: + lines.append("## Queue Time Distribution by Runner Label") + lines.append("") + for label in sorted(queue_dist, key=lambda k: -queue_dist[k]["total"]): + dist = queue_dist[label] + lines.append(f"### `{label}`") + lines.append("") + lines.append( + f"> **Samples:** {dist['total']} | **P50:** {_format_duration_seconds(dist['p50'])} | **P90:** {_format_duration_seconds(dist['p90'])} | **P99:** {_format_duration_seconds(dist['p99'])}" + ) + lines.append("") + lines.append("| Queue Time Range | Count | Percentage |") + lines.append("|-----------------|-------|------------|") + for b in dist["buckets"]: + bar = "#" * int(b["percentage"] / 3) + lines.append( + f"| {b['range']} | {b['count']} | {b['percentage']}% {bar} |" + ) + lines.append("") + + # --- Failed Jobs Detail (collapsible) --- + failed_jobs = [ + j + for j in jobs + if j.get("conclusion") == "failure" and not j.get("is_stuck", False) + ] + if failed_jobs: + lines.append("
") + lines.append( + f"Failed Jobs ({len(failed_jobs)} total) - Click to expand" + ) + lines.append("") + lines.append( + "| Job Name | Runner | Workflow | Queue | Duration | PR/Branch | Link |" + ) + lines.append( + "|----------|--------|---------|-------|----------|-----------|------|" + ) + for j in sorted(failed_jobs, key=lambda x: x["created_at"], reverse=True): + queue = calculate_queue_time( + j["created_at"], j["started_at"], j["status"], report_time + ) + dur = calculate_duration(j["started_at"], j["completed_at"]) + pr_info = ( + f"PR#{j['pr_number']}" if j.get("pr_number") else j.get("branch", "-") + ) + url = j.get("html_url", "") + wf = j.get("workflow", "-") + lines.append( + f"| `{j['job_name']}` | `{j['runner_name']}` | `{wf}` " + f"| {queue} | {dur} | {pr_info} | [View]({url}) |" + ) + lines.append("") + lines.append("
") + lines.append("") + + # --- Stuck Jobs --- + stuck_jobs = [j for j in jobs if j.get("is_stuck", False)] + if stuck_jobs: + lines.append("## Stuck/Ghost Jobs") + lines.append("") + lines.append( + "> Jobs showing `in_progress` but have no runner assigned or workflow run is cancelled" + ) + lines.append("") + lines.append( + "| Job Name | Job Status | Run Status | Runner | Workflow | Link |" + ) + lines.append("|----------|-----------|-----------|--------|---------|------|") + for j in sorted(stuck_jobs, key=lambda x: x["created_at"], reverse=True): + run_info = f"{j.get('run_status', '-')}/{j.get('run_conclusion', '-')}" + url = j.get("html_url", "") + wf = j.get("workflow", "-") + lines.append( + f"| `{j['job_name']}` | {j['status']} | {run_info} " + f"| `{j['runner_name']}` | `{wf}` | [View]({url}) |" + ) + lines.append("") + + return "\n".join(lines) + + +def main(): # Capture the time when the command is run (both datetime and formatted string) report_time = datetime.now(timezone.utc) report_generated_time = report_time.strftime("%Y-%m-%d %H:%M:%S") @@ -812,13 +1526,14 @@ def main(): ) parser.add_argument( "--job", - required=True, - help="Job name filter (e.g., 'stage-c-test-large-8-gpu-amd-mi35x')", + required=False, + default=None, + help="Job name filter (required unless --runner-report is used)", ) parser.add_argument( "--workflow", default="pr-test-amd.yml", - help="Workflow file name (default: pr-test-amd.yml)", + help="Workflow file name, or comma-separated list for --runner-report (default: pr-test-amd.yml)", ) parser.add_argument( "--hours", @@ -847,20 +1562,117 @@ def main(): type=str, help="Write output to file", ) + parser.add_argument( + "--runner-report", + action="store_true", + help="Generate runner fleet analytics report across all jobs (no --job filter needed)", + ) + parser.add_argument( + "--input-data-file", + type=str, + help="Load a prefetched Actions snapshot JSON instead of calling gh api", + ) + parser.add_argument( + "--dump-data-file", + type=str, + help="Fetch Actions data once and save it as a snapshot JSON file", + ) args = parser.parse_args() - results = query_jobs( - args.repo, - args.job, - args.workflow, - args.hours, - args.status, - ) + if args.input_data_file and args.dump_data_file: + parser.error("--input-data-file and --dump-data-file cannot be used together") + + if not args.runner_report and not args.job and not args.dump_data_file: + parser.error( + "--job is required unless --runner-report or --dump-data-file is specified" + ) + + workflows = [w.strip() for w in args.workflow.split(",") if w.strip()] + + if not args.input_data_file and not check_gh_cli_available(): + sys.exit(1) + + snapshot = None + repo = args.repo + fetch_metadata = None + + if args.input_data_file: + snapshot = load_snapshot(args.input_data_file) + repo = snapshot.get("repo", args.repo) + fetch_metadata = snapshot.get("fetch_metadata") + + if args.dump_data_file: + snapshot = fetch_all_jobs_snapshot(repo, workflows, args.hours) + save_snapshot(args.dump_data_file, snapshot) + summary = summarize_fetch_metadata(snapshot.get("fetch_metadata"), workflows) + print(f"Snapshot written to {args.dump_data_file}", file=sys.stderr) + if summary and summary["incomplete"]: + print( + "Warning: Snapshot is incomplete due to rate limit/API fetch failures.", + file=sys.stderr, + ) + if summary["known_runs"] > 0: + print( + f"Known runs fetched successfully: {summary['runs_with_jobs']}/{summary['known_runs']}", + file=sys.stderr, + ) + print( + f"Skipped runs with unknown job counts: {len(summary['skipped_runs'])}", + file=sys.stderr, + ) + return + + # --- Runner fleet report mode --- + if args.runner_report: + if snapshot is None: + snapshot = fetch_all_jobs_snapshot(repo, workflows, args.hours) + fetch_metadata = snapshot.get("fetch_metadata") + + jobs = [ + job for job in snapshot["jobs"] if job.get("workflow") in set(workflows) + ] + + md_content = format_runner_report_markdown( + jobs, + workflows, + args.hours, + report_generated_time, + report_time, + fetch_metadata, + ) + + print(md_content) + + if args.output_file: + with open(args.output_file, "w") as f: + f.write(md_content) + print(f"\nOutput written to {args.output_file}", file=sys.stderr) + + if args.summary: + summary_file = os.environ.get("GITHUB_STEP_SUMMARY") + if summary_file: + with open(summary_file, "a") as f: + f.write(md_content) + f.write("\n") + print("Summary written to GITHUB_STEP_SUMMARY", file=sys.stderr) + else: + print( + "Warning: GITHUB_STEP_SUMMARY not set, markdown printed above.", + file=sys.stderr, + ) + return + + # --- Original per-job report mode --- + if snapshot is None: + snapshot = fetch_all_jobs_snapshot(repo, [args.workflow], args.hours) + fetch_metadata = snapshot.get("fetch_metadata") + + results = filter_jobs(snapshot["jobs"], args.job, args.workflow, args.status) output_content = None if args.output == "table": - print_table(results, args.repo, report_generated_time, report_time) + print_table(results, repo, report_generated_time, report_time) elif args.output == "csv": lines = [ "job_name,status,is_stuck,conclusion,created_at,started_at,queue_time,duration,runner,run_status,run_conclusion,pr_number,branch,url" @@ -877,7 +1689,6 @@ def main(): output_content = "\n".join(lines) print(output_content) elif args.output == "json": - # Add calculated fields to JSON output for consistency json_results = [] for r in sorted(results, key=lambda x: x["created_at"], reverse=True): r_copy = r.copy() @@ -892,27 +1703,39 @@ def main(): print(output_content) elif args.output == "markdown": output_content = format_markdown( - results, args.repo, args.job, args.hours, report_generated_time, report_time + results, + repo, + args.job, + args.hours, + report_generated_time, + report_time, + fetch_metadata, + args.workflow, ) print(output_content) - # Write to file if specified if args.output_file and output_content: with open(args.output_file, "w") as f: f.write(output_content) print(f"\nOutput written to {args.output_file}", file=sys.stderr) - # Write to GITHUB_STEP_SUMMARY if requested if args.summary: md_content = format_markdown( - results, args.repo, args.job, args.hours, report_generated_time, report_time + results, + repo, + args.job, + args.hours, + report_generated_time, + report_time, + fetch_metadata, + args.workflow, ) summary_file = os.environ.get("GITHUB_STEP_SUMMARY") if summary_file: with open(summary_file, "a") as f: f.write(md_content) f.write("\n") - print(f"Summary written to GITHUB_STEP_SUMMARY", file=sys.stderr) + print("Summary written to GITHUB_STEP_SUMMARY", file=sys.stderr) else: print( "Warning: GITHUB_STEP_SUMMARY not set, printing markdown instead:",