ci_analyzer_balance.py

import argparse
import json
import os
import re
import sys
import time
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional, Tuple

import requests


class SGLangTestBalanceAnalyzer:

    def __init__(self, token: str):
        self.token = token
        self.base_url = "https://api.github.com"
        self.repo = "sgl-project/sglang"
        self.headers = {
            "Authorization": f"token {token}",
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "SGLang-Test-Balance-Analyzer/1.0",
        }
        self.session = requests.Session()
        self.session.headers.update(self.headers)

        self.test_time_pattern = re.compile(
            r"filename='([^']+)',\s*elapsed=(\d+),\s*estimated_time=(\d+)"
        )

    def get_recent_runs(self, limit: int = 1000) -> List[Dict]:
        print(f"Fetching {limit} recent CI runs...")

        all_runs = []
        page = 1
        per_page = 100

        while len(all_runs) < limit:
            url = f"{self.base_url}/repos/{self.repo}/actions/runs"
            params = {"per_page": min(per_page, limit - len(all_runs)), "page": page}

            try:
                response = self.session.get(url, params=params)
                response.raise_for_status()
                data = response.json()

                if not data.get("workflow_runs"):
                    break

                all_runs.extend(data["workflow_runs"])
                print(f"Fetched {len(all_runs)} runs so far...")

                if len(data["workflow_runs"]) < per_page:
                    break

                page += 1
                time.sleep(0.1)

            except requests.exceptions.RequestException as e:
                print(f"Error fetching CI data: {e}")
                break

        return all_runs[:limit]

    def get_job_logs(self, run_id: int, job_name: str) -> Optional[str]:
        try:
            jobs_url = f"{self.base_url}/repos/{self.repo}/actions/runs/{run_id}/jobs"
            response = self.session.get(jobs_url)
            response.raise_for_status()
            jobs_data = response.json()

            target_job = None
            for job in jobs_data.get("jobs", []):
                if job.get("name", "") == job_name:
                    target_job = job
                    break

            if not target_job:
                return None

            logs_url = f"{self.base_url}/repos/{self.repo}/actions/jobs/{target_job['id']}/logs"
            response = self.session.get(logs_url)
            response.raise_for_status()

            return response.text

        except Exception as e:
            if "404" not in str(e):
                print(f"Failed to get job {job_name} logs: {e}")
            return None

    def get_all_jobs_for_run(self, run_id: int) -> List[Dict]:
        try:
            jobs_url = f"{self.base_url}/repos/{self.repo}/actions/runs/{run_id}/jobs"
            response = self.session.get(jobs_url)
            response.raise_for_status()
            jobs_data = response.json()
            return jobs_data.get("jobs", [])
        except Exception as e:
            print(f"Failed to get jobs for run {run_id}: {e}")
            return []

    def get_job_logs_by_id(self, job_id: int) -> Optional[str]:
        try:
            logs_url = f"{self.base_url}/repos/{self.repo}/actions/jobs/{job_id}/logs"
            response = self.session.get(logs_url)
            response.raise_for_status()
            return response.text
        except Exception as e:
            if "404" not in str(e):
                print(f"Failed to get job {job_id} logs: {e}")
            return None

    def parse_test_times(self, log_content: str) -> List[Dict]:
        if not log_content:
            return []

        test_times = []
        matches = self.test_time_pattern.findall(log_content)
        filtered_count = 0

        for match in matches:
            filename, elapsed_str, estimated_str = match
            try:
                elapsed = int(elapsed_str)
                estimated = int(estimated_str)
                gap = elapsed - estimated

                if self._is_abnormal_test_data(
                    elapsed, estimated, log_content, filename
                ):
                    filtered_count += 1
                    continue

                test_times.append(
                    {
                        "filename": filename,
                        "elapsed": elapsed,
                        "estimated": estimated,
                        "gap": gap,
                    }
                )
            except ValueError:
                continue

        return test_times

    def _is_abnormal_test_data(
        self, elapsed: int, estimated: int, log_content: str, filename: str
    ) -> bool:

        # To avoid collect retry data
        if elapsed % estimated == 0:
            return True

        return False

    def collect_test_balance_data(self, runs: List[Dict]) -> Dict[str, Dict]:
        print("Starting test balance data collection...")

        test_gaps = defaultdict(
            lambda: {
                "max_gap": 0,
                "max_elapsed": 0,
                "max_estimated": 0,
                "max_gap_run_info": {},
                "total_runs": 0,
                "all_gaps": [],
            }
        )

        total_tests_parsed = 0
        abnormal_tests_filtered = 0

        target_job_prefixes = [
            "unit-test-frontend",
            "unit-test-backend-1-gpu",
            "unit-test-backend-2-gpu",
            "unit-test-backend-4-gpu",
            "unit-test-backend-8-gpu-h200",
            "unit-test-backend-8-gpu-h20",
            "unit-test-backend-4-gpu-b200",
            "unit-test-backend-4-gpu-gb200",
            "unit-test-deepep-4-gpu",
            "unit-test-deepep-8-gpu",
            "unit-test-backend-8-gpu-deepseek-v32",
            "performance-test-1-gpu-part-1",
            "performance-test-1-gpu-part-2",
            "performance-test-1-gpu-part-3",
            "performance-test-2-gpu",
            "accuracy-test-1-gpu",
            "accuracy-test-2-gpu",
        ]

        total_runs = len(runs)
        for i, run in enumerate(runs, 1):
            if i % 10 == 0 or i == total_runs:
                print(f"Processing run {i}/{total_runs}: #{run.get('run_number')}")

            workflow_name = run.get("name", "")
            if "AMD" in workflow_name or "amd" in workflow_name.lower():
                continue

            run_info = {
                "run_number": run.get("run_number"),
                "created_at": run.get("created_at"),
                "head_sha": run.get("head_sha", "")[:8],
                "author": run.get("head_commit", {})
                .get("author", {})
                .get("name", "Unknown"),
                "url": f"https://github.com/{self.repo}/actions/runs/{run.get('id')}",
            }

            pull_requests = run.get("pull_requests", [])
            if pull_requests:
                run_info["pr_number"] = pull_requests[0].get("number")

            all_jobs = self.get_all_jobs_for_run(run.get("id"))

            for job in all_jobs:
                job_name = job.get("name", "")
                job_id = job.get("id")

                matches_prefix = False
                for prefix in target_job_prefixes:
                    if job_name.startswith(prefix):
                        matches_prefix = True
                        break

                if not matches_prefix:
                    continue

                logs = self.get_job_logs_by_id(job_id)
                if not logs:
                    continue

                test_times = self.parse_test_times(logs)
                total_tests_parsed += len(test_times)

                for test_data in test_times:
                    filename = test_data["filename"]
                    elapsed = test_data["elapsed"]
                    estimated = test_data["estimated"]
                    gap = test_data["gap"]

                    test_stats = test_gaps[filename]
                    test_stats["total_runs"] += 1
                    test_stats["all_gaps"].append(gap)

                    if gap > test_stats["max_gap"]:
                        test_stats["max_gap"] = gap
                        test_stats["max_elapsed"] = elapsed
                        test_stats["max_estimated"] = estimated
                        test_stats["max_gap_run_info"] = {
                            **run_info,
                            "job_name": job_name,
                            "job_url": f"https://github.com/{self.repo}/actions/runs/{run.get('id')}/job/{job_id}",
                        }

            time.sleep(0.1)

        return dict(test_gaps)

    def generate_balance_report(
        self, test_data: Dict[str, Dict], output_file: str = "test_balance_report.json"
    ):
        print("\n" + "=" * 80)
        print("SGLang Test Balance Analysis Report (PR Test GPU Jobs)")
        print("=" * 80)

        sorted_tests = sorted(
            test_data.items(), key=lambda x: x[1]["max_gap"], reverse=True
        )

        print(f"\nTotal tests analyzed: {len(sorted_tests)}")
        print(
            f"Tests with significant gaps (>100s): {len([t for t in sorted_tests if t[1]['max_gap'] > 100])}"
        )
        print(
            f"Tests with large gaps (>300s): {len([t for t in sorted_tests if t[1]['max_gap'] > 300])}"
        )
        print(
            f"Note: Abnormal test data (due to failures/retries) has been filtered out"
        )

        report_data = {
            "summary": {
                "total_tests": len(sorted_tests),
                "tests_with_gaps_over_100s": len(
                    [t for t in sorted_tests if t[1]["max_gap"] > 100]
                ),
                "tests_with_gaps_over_300s": len(
                    [t for t in sorted_tests if t[1]["max_gap"] > 300]
                ),
                "analysis_timestamp": datetime.now().isoformat(),
            },
            "test_balance_table": [],
        }

        print(f"\nTop 50 PR Test GPU Jobs with Largest Time Gaps:")
        print("-" * 100)
        print(
            f"{'Rank':<4} {'Test File':<40} {'Max Gap':<8} {'Max Elapsed':<12} {'Max Estimated':<15} {'Job Name':<25}"
        )
        print("-" * 100)

        for i, (filename, stats) in enumerate(sorted_tests[:50], 1):
            test_name = filename.split("/")[-1] if "/" in filename else filename
            job_name = (
                stats["max_gap_run_info"].get("job_name", "Unknown")
                if stats["max_gap_run_info"]
                else "Unknown"
            )

            print(
                f"{i:<4} {test_name:<40} {stats['max_gap']:<8} {stats['max_elapsed']:<12} {stats['max_estimated']:<15} {job_name:<25}"
            )

            report_data["test_balance_table"].append(
                {
                    "rank": i,
                    "filename": filename,
                    "test_name": test_name,
                    "max_gap": stats["max_gap"],
                    "max_elapsed": stats["max_elapsed"],
                    "max_estimated": stats["max_estimated"],
                    "max_gap_run_info": stats["max_gap_run_info"],
                    "total_runs": stats["total_runs"],
                }
            )

        with open(output_file, "w", encoding="utf-8") as f:
            json.dump(report_data, f, ensure_ascii=False, indent=2)
        print(f"\nDetailed report saved to: {output_file}")

        return report_data

    def generate_github_summary(self, report_data: Dict):
        try:
            github_step_summary = os.environ.get("GITHUB_STEP_SUMMARY")
            if not github_step_summary:
                print("Not running in GitHub Actions, skipping summary generation")
                return

            print("Generating GitHub Actions summary for Test Balance Analysis...")

            summary_lines = []
            summary_lines.append(
                "# SGLang Test Balance Analysis Report (PR Test GPU Jobs)"
            )
            summary_lines.append("")
            summary_lines.append(
                f"**Analysis Timestamp:** {report_data['summary']['analysis_timestamp']}"
            )
            summary_lines.append("")

            summary_lines.append("## Summary Statistics")
            summary_lines.append("")
            summary_lines.append("| Metric | Count |")
            summary_lines.append("|--------|-------|")
            summary_lines.append(
                f"| Total Tests Analyzed | {report_data['summary']['total_tests']} |"
            )
            summary_lines.append(
                f"| Tests with Gaps > 100s | {report_data['summary']['tests_with_gaps_over_100s']} |"
            )
            summary_lines.append(
                f"| Tests with Gaps > 300s | {report_data['summary']['tests_with_gaps_over_300s']} |"
            )
            summary_lines.append("")

            summary_lines.append("## Top 30 PR Test GPU Jobs with Largest Time Gaps")
            summary_lines.append("")
            summary_lines.append(
                "| Rank | Test File | Max Gap (s) | Max Elapsed (s) | Max Estimated (s) | Job Name | Job Link | Total Runs |"
            )
            summary_lines.append(
                "|------|-----------|-------------|----------------|------------------|---------|----------|------------|"
            )

            for test in report_data["test_balance_table"][:30]:
                test_name = test["test_name"]
                if len(test_name) > 30:
                    test_name = test_name[:27] + "..."

                job_name = (
                    test["max_gap_run_info"].get("job_name", "Unknown")
                    if test["max_gap_run_info"]
                    else "Unknown"
                )
                job_url = (
                    test["max_gap_run_info"].get("job_url", "")
                    if test["max_gap_run_info"]
                    else ""
                )
                job_link = f"[{job_name}]({job_url})" if job_url else job_name

                summary_lines.append(
                    f"| {test['rank']} | `{test_name}` | {test['max_gap']} | {test['max_elapsed']} | {test['max_estimated']} | {job_name} | [{job_name}]({job_url}) | {test['total_runs']} |"
                )

            summary_lines.append("")
            summary_lines.append("## Recommendations")
            summary_lines.append("")
            summary_lines.append(
                "Based on the analysis above, consider adjusting estimated times for tests with large gaps:"
            )
            summary_lines.append("")

            top_5_tests = report_data["test_balance_table"][:5]
            for test in top_5_tests:
                test_name = test["test_name"]
                if len(test_name) > 40:
                    test_name = test_name[:37] + "..."
                suggested_estimated = test["max_elapsed"] + 50
                summary_lines.append(
                    f"- **{test_name}**: Current max elapsed: {test['max_elapsed']}s, suggested estimated: {suggested_estimated}s"
                )

            summary_lines.append("")
            summary_lines.append(
                "Set estimated times to be slightly higher than the maximum observed elapsed time to avoid CI timeouts."
            )

            with open(github_step_summary, "w", encoding="utf-8") as f:
                f.write("\n".join(summary_lines))

            print("GitHub Actions summary generated successfully")

        except Exception as e:
            print(f"Failed to generate GitHub Actions summary: {e}")

    def save_csv_report(
        self, report_data: Dict, output_file: str = "test_balance_report.csv"
    ):
        import csv

        with open(output_file, "w", encoding="utf-8", newline="") as f:
            writer = csv.writer(f)

            writer.writerow(
                [
                    "Rank",
                    "Test File",
                    "Test Name",
                    "Max Gap (s)",
                    "Max Elapsed (s)",
                    "Max Estimated (s)",
                    "Job Name",
                    "Max Gap Job URL",
                    "Total Runs",
                ]
            )

            for test in report_data["test_balance_table"]:
                max_job_url = (
                    test["max_gap_run_info"].get("job_url", "")
                    if test["max_gap_run_info"]
                    else ""
                )
                job_name = (
                    test["max_gap_run_info"].get("job_name", "Unknown")
                    if test["max_gap_run_info"]
                    else "Unknown"
                )

                writer.writerow(
                    [
                        test["rank"],
                        test["filename"],
                        test["test_name"],
                        test["max_gap"],
                        test["max_elapsed"],
                        test["max_estimated"],
                        job_name,
                        max_job_url,
                        test["total_runs"],
                    ]
                )

        print(f"CSV report saved to: {output_file}")


def main():
    parser = argparse.ArgumentParser(description="SGLang Test Balance Analyzer")
    parser.add_argument("--token", required=True, help="GitHub Personal Access Token")
    parser.add_argument(
        "--limit",
        type=int,
        default=1000,
        help="Number of runs to analyze (default: 1000)",
    )
    parser.add_argument(
        "--output",
        default="test_balance_report.json",
        help="Output file (default: test_balance_report.json)",
    )

    args = parser.parse_args()

    analyzer = SGLangTestBalanceAnalyzer(args.token)

    try:
        runs = analyzer.get_recent_runs(args.limit)

        if not runs:
            print("No CI run data found")
            return

        test_data = analyzer.collect_test_balance_data(runs)

        if not test_data:
            print("No test balance data found")
            return

        report_data = analyzer.generate_balance_report(test_data, args.output)

        csv_output = args.output.replace(".json", ".csv")
        analyzer.save_csv_report(report_data, csv_output)

        analyzer.generate_github_summary(report_data)

    except Exception as e:
        print(f"Error during analysis: {e}")
        import traceback

        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()