From 193bbf9b66032f3e3b9e2d17231bd1db139558a3 Mon Sep 17 00:00:00 2001 From: Lianmin Zheng Date: Fri, 20 Mar 2026 00:05:22 -0700 Subject: [PATCH] chore(ci): remove deprecated CI Monitor workflow (#20993) --- .github/workflows/ci-monitor.yml | 111 -- scripts/ci_monitor/README.md | 313 +--- scripts/ci_monitor/ci_analyzer.py | 1213 --------------- scripts/ci_monitor/ci_analyzer_balance.py | 534 ------- scripts/ci_monitor/ci_analyzer_perf.py | 1375 ----------------- .../ci_monitor/post_ci_failures_to_slack.py | 8 +- 6 files changed, 22 insertions(+), 3532 deletions(-) delete mode 100644 .github/workflows/ci-monitor.yml delete mode 100755 scripts/ci_monitor/ci_analyzer.py delete mode 100755 scripts/ci_monitor/ci_analyzer_balance.py delete mode 100755 scripts/ci_monitor/ci_analyzer_perf.py diff --git a/.github/workflows/ci-monitor.yml b/.github/workflows/ci-monitor.yml deleted file mode 100644 index 28a198a32..000000000 --- a/.github/workflows/ci-monitor.yml +++ /dev/null @@ -1,111 +0,0 @@ -name: CI Monitor - -on: - schedule: - - cron: '0 */12 * * *' # Every 12 hours for main analysis - workflow_dispatch: - inputs: - limit: - description: 'Number of CI runs to analyze' - required: false - default: '1000' - type: string - -concurrency: - group: ci-monitor-${{ github.ref }} - cancel-in-progress: true - -permissions: - contents: write - actions: read - -jobs: - ci-monitor: - if: github.repository == 'sgl-project/sglang'|| github.event_name == 'pull_request' - 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.9' - - - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install requests matplotlib pandas - - - name: Run CI Analysis - env: - GITHUB_TOKEN: ${{ secrets.GH_PAT_FOR_NIGHTLY_CI_DATA }} - PYTHONUNBUFFERED: 1 - PYTHONIOENCODING: utf-8 - run: | - cd scripts/ci_monitor - python ci_analyzer.py --token $GITHUB_TOKEN --limit ${{ inputs.limit || '1000' }} --output ci_analysis_$(date +%Y%m%d_%H%M%S).json - - - name: Run Nightly Test Analysis - env: - GITHUB_TOKEN: ${{ secrets.GH_PAT_FOR_NIGHTLY_CI_DATA }} - PYTHONUNBUFFERED: 1 - PYTHONIOENCODING: utf-8 - run: | - cd scripts/ci_monitor - python ci_analyzer.py --token $GITHUB_TOKEN --mode nightly --days 2 --output nightly_analysis_$(date +%Y%m%d_%H%M%S).json - - - name: Run Performance Analysis - env: - GITHUB_TOKEN: ${{ secrets.GH_PAT_FOR_NIGHTLY_CI_DATA }} - PYTHONUNBUFFERED: 1 - PYTHONIOENCODING: utf-8 - run: | - cd scripts/ci_monitor - python ci_analyzer_perf.py --token $GITHUB_TOKEN --limit ${{ inputs.limit || '1000' }} --output-dir performance_tables_$(date +%Y%m%d_%H%M%S) --upload-to-github - - - name: Upload Analysis Results - uses: actions/upload-artifact@v4 - with: - name: ci-analysis-results-${{ github.run_number }} - path: | - scripts/ci_monitor/ci_analysis_*.json - scripts/ci_monitor/nightly_analysis_*.json - scripts/ci_monitor/performance_tables_* - retention-days: 30 - - ci-monitor-balance: - needs: ci-monitor - if: github.repository == 'sgl-project/sglang' || github.event_name == 'pull_request' - 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.9' - - - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install requests - - - name: Run Test Balance Analysis - env: - GITHUB_TOKEN: ${{ secrets.GH_PAT_FOR_NIGHTLY_CI_DATA }} - PYTHONUNBUFFERED: 1 - PYTHONIOENCODING: utf-8 - run: | - cd scripts/ci_monitor - python ci_analyzer_balance.py --token $GITHUB_TOKEN --limit ${{ inputs.limit || '1000' }} --output test_balance_report_$(date +%Y%m%d_%H%M%S).json - - - name: Upload Balance Analysis Results - uses: actions/upload-artifact@v4 - with: - name: test-balance-results-${{ github.run_number }} - path: | - scripts/ci_monitor/test_balance_report_*.json - scripts/ci_monitor/test_balance_report_*.csv - retention-days: 30 diff --git a/scripts/ci_monitor/README.md b/scripts/ci_monitor/README.md index 4c0f953dd..009c7b1c3 100644 --- a/scripts/ci_monitor/README.md +++ b/scripts/ci_monitor/README.md @@ -1,334 +1,55 @@ -# SGLang CI Monitor +# SGLang CI failure monitoring -> **Note**: This README.md is primarily generated by Claude 4 with some manual adjustments. +Scripts used by [.github/workflows/ci-failure-monitor.yml](../../.github/workflows/ci-failure-monitor.yml): scheduled failure analysis and optional Slack notifications. -A comprehensive toolkit to analyze CI failures and performance trends for the SGLang project. This toolkit includes four main tools: +## Tools -1. **CI Analyzer** (`ci_analyzer.py`): Analyzes CI failures and provides detailed failure pattern analysis -2. **Performance Analyzer** (`ci_analyzer_perf.py`): Tracks performance metrics over time and generates trend charts -3. **Test Balance Analyzer** (`ci_analyzer_balance.py`): Analyzes test time gaps between elapsed and estimated times to help balance CI -4. **Failures Analyzer** (`ci_failures_analysis.py`): Tracks consecutive failures, identifies flaky jobs, and monitors runner health +1. **Failures Analyzer** (`ci_failures_analysis.py`): Tracks consecutive failures, identifies flaky jobs, and monitors runner health across PR Test / Nightly workflows (Nvidia, AMD, Intel, XPU, NPU). -## Features - -### CI Analyzer (`ci_analyzer.py`) -- **Simple Analysis**: Analyze recent CI runs and identify failure patterns -- **Category Classification**: Automatically categorize failures by type (unit-test, performance, etc.) -- **Pattern Recognition**: Identify common failure patterns (timeouts, build failures, etc.) -- **CI Links**: Direct links to recent failed CI runs for detailed investigation -- **Last Success Tracking**: Track the last successful run for each failed job with PR information -- **JSON Export**: Export detailed analysis data to JSON format - -### Performance Analyzer (`ci_analyzer_perf.py`) -- **Performance Tracking**: Monitor performance metrics across CI runs over time -- **Automated Chart Generation**: Generate time-series charts for each performance metric -- **Multi-Test Support**: Track performance for all test types (throughput, latency, accuracy) -- **CSV Export**: Export performance data in structured CSV format -- **Trend Analysis**: Visualize performance trends with interactive charts -- **Comprehensive Metrics**: Track output throughput, E2E latency, TTFT, accept length, and more -- **Time-Based Sampling**: Intelligent sampling strategy to cover extended time periods (up to 30 days) with limited API calls - -### Test Balance Analyzer (`ci_analyzer_balance.py`) -- **Time Gap Analysis**: Identify GPU tests with large gaps between elapsed and estimated times -- **CI Balancing**: Help optimize CI by identifying tests that need time adjustments -- **Gap Tracking**: Track maximum time gaps for each test across multiple CI runs -- **PR Test Focus**: Only analyzes GPU jobs from pr-test.yml workflow (excludes AMD and other workflows) -- **Ranking System**: Sort tests by time gap severity to prioritize adjustments -- **CSV Export**: Export analysis results in CSV format for easy review -- **GitHub Integration**: Generate GitHub Actions summaries with recommendations - -### Failures Analyzer (`ci_failures_analysis.py`) -- **Consecutive Failure Tracking**: Identify jobs currently failing -- **Runner Health Monitoring**: Track runner failure rates and identify problematic infrastructure -- **Multi-Workflow Support**: Monitors PR Test (Nvidia), PR Test (AMD), and PR Test (Xeon) workflows -- **Queue Time Tracking**: Monitor average and P90 queue times per runner type -- **Alert System**: Automatic alerts for consecutive failures and runner problems -- **Instance Tracking**: Monitor specific runner instances for targeted remediation -- **Slack Notifications**: Send condensed alerts to Slack (top 3 jobs/runners by consecutive failures and failure rates) -- **GitHub Integration**: Generate comprehensive summaries with actionable recommendations -- **JSON Export**: Export detailed analysis data for further processing - -### Common Features -- **Automated Monitoring**: GitHub Actions workflow for continuous CI and performance monitoring +2. **Slack poster** (`post_ci_failures_to_slack.py`): Sends a condensed summary from a failure-analysis JSON to Slack (invoked by the workflow when `SGLANG_DIFFUSION_SLACK_TOKEN` is set). ## Installation -### For CI Analyzer -No additional dependencies required beyond Python standard library and `requests`: - ```bash -pip install requests +pip install requests slack_sdk ``` -### For Performance Analyzer -Additional dependencies required for chart generation: - -```bash -pip install requests matplotlib pandas -``` - -### For Test Balance Analyzer -No additional dependencies required beyond Python standard library and `requests`: - -```bash -pip install requests -``` +(`slack_sdk` is only required for `post_ci_failures_to_slack.py`.) ## Usage -### CI Analyzer - -#### Basic Usage - -```bash -# Replace YOUR_GITHUB_TOKEN with your actual token from https://github.com/settings/tokens -python ci_analyzer.py --token YOUR_GITHUB_TOKEN -``` - -#### Advanced Usage - -```bash -# Analyze last 1000 runs -python ci_analyzer.py --token YOUR_GITHUB_TOKEN --limit 1000 - -# Custom output file -python ci_analyzer.py --token YOUR_GITHUB_TOKEN --limit 500 --output my_analysis.json -``` - -### Performance Analyzer - -#### Basic Usage - -```bash -# Analyze performance trends from recent CI runs -python ci_analyzer_perf.py --token YOUR_GITHUB_TOKEN -``` - -#### Advanced Usage - -```bash -# Analyze last 1000 PR Test runs (auto-enables uniform sampling for ~30 days coverage) -python ci_analyzer_perf.py --token YOUR_GITHUB_TOKEN --limit 1000 - -# Custom output directory -python ci_analyzer_perf.py --token YOUR_GITHUB_TOKEN --limit 500 --output-dir my_performance_data - -# Use sampling with 500 runs (will use sequential mode since < 500 threshold) -python ci_analyzer_perf.py --token YOUR_GITHUB_TOKEN --limit 500 - -# Get ALL performance data within a specific date range (recommended for historical analysis) -python ci_analyzer_perf.py --token YOUR_GITHUB_TOKEN --start-date 2024-12-01 --end-date 2024-12-31 - -# Get complete data for the last week -python ci_analyzer_perf.py --token YOUR_GITHUB_TOKEN --start-date $(date -d '7 days ago' +%Y-%m-%d) --end-date $(date +%Y-%m-%d) - -# Upload results to GitHub repository for sharing -python ci_analyzer_perf.py --token YOUR_GITHUB_TOKEN --limit 1000 --upload-to-github -``` - -### Test Balance Analyzer - -#### Basic Usage - -```bash -# Analyze PR Test GPU job time gaps from recent CI runs -python ci_analyzer_balance.py --token YOUR_GITHUB_TOKEN -``` - -#### Advanced Usage - -```bash -# Analyze last 1000 PR Test GPU CI runs for comprehensive test balance analysis -python ci_analyzer_balance.py --token YOUR_GITHUB_TOKEN --limit 1000 - -# Custom output file -python ci_analyzer_balance.py --token YOUR_GITHUB_TOKEN --limit 500 --output my_balance_analysis.json -``` - ### Failures Analyzer -#### Quick Start - ```bash -# Set token as environment variable (recommended for security) export GITHUB_TOKEN="your_token_here" -# Quick test with recent runs python ci_failures_analysis.py --token $GITHUB_TOKEN --limit 50 --threshold 2 - -# Standard analysis (same as automated workflow) python ci_failures_analysis.py --token $GITHUB_TOKEN --limit 300 --threshold 2 - -# Deep analysis python ci_failures_analysis.py --token $GITHUB_TOKEN --limit 500 --threshold 3 ``` -#### Monitored Workflows +### Slack notifications -The Failures Analyzer monitors the following workflows: - -- **PR Test** - Nvidia GPU tests (self-hosted runners: 1-gpu-runner, 4-gpu-h100-runner, etc.) -- **PR Test (AMD)** - AMD GPU tests (AMD-specific runners) -- **PR Test (Xeon)** - Intel Xeon CPU tests (Xeon-specific runners) - -All three workflows are analyzed together, with runner statistics tracked separately by runner type. - -#### Slack Notifications - -The Failures Analyzer can send condensed alerts to Slack. See [SLACK_SETUP.md](SLACK_SETUP.md) for complete setup instructions. - -**What gets sent:** -- Top 3 jobs with consecutive failures -- Top 3 runners with consecutive failures -- Top 3 jobs with highest total failure rate -- Top 3 runners with highest total failure rate -- Queue time summary +From the `scripts/ci_monitor` directory, after generating a report: ```bash -# Send Slack notification from analysis JSON -export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/YOUR/WEBHOOK/URL" -python slack_notifier.py --json ci_failure_analysis.json +export SGLANG_DIFFUSION_SLACK_TOKEN="xoxb-..." +python post_ci_failures_to_slack.py --report-file ci_failure_analysis_YYYYMMDD_HHMMSS.json ``` -#### Understanding the Output +## Token permissions -The script generates a **2-section report**: +The GitHub token needs `repo` and `workflow` scopes to read CI run data; otherwise API calls may return 404. -**Section 1: Currently Broken Jobs (Active Consecutive Failures)** -- Shows consecutive failure streaks -- These need immediate attention - -**Section 2: Runner Health Analysis** -- Shows which runners have high failure rates -- Includes queue time metrics (average and P90) -- Helps identify infrastructure vs code issues - -#### Alert Types - -**Job Alerts (Consecutive Failures):** -- Triggered when a job fails ≥ threshold times in a row -- Example: threshold=2, job fails 3 times → ALERT - -**Runner Alerts:** -- **Runner Health**: Runner has >30% failure rate with ≥2 different jobs failing -- **Runner Instance**: Specific instance has >50% failure rate with ≥3 jobs - -#### Output Files - -- **Console**: Human-readable 3-section report (always generated) -- **JSON**: Detailed data (optional, only if `--output` is specified) -- **GitHub Summary**: Markdown (automatically generated in GitHub Actions) - -**Important**: Make sure your GitHub token has `repo` and `workflow` permissions, otherwise you'll get 404 errors. - -## Data Collection Strategies - -The Performance Analyzer offers multiple strategies for collecting performance data to suit different analysis needs. - -### 1. Uniform Sampling Strategy - -**When to use**: Daily monitoring and trend analysis over extended periods. - -- **Automatically enabled** when `--limit >= 500` -- **Disabled** for smaller limits (< 500) to maintain backward compatibility - -#### How it works: -- Collects data uniformly across a 30-day period -- Ensures even time distribution of samples -- Provides consistent coverage for trend analysis - -#### Example with 1000 Runs: -- **Time Range**: Last 30 days -- **Distribution**: 1000 samples evenly distributed across the period -- **Coverage**: ~33 samples per day on average - -### 2. Date Range Collection - -**When to use**: Historical analysis, specific period investigation, or complete data collection. - -Use `--start-date` and `--end-date` parameters to get **ALL** CI runs within a specific time range. - -#### Features: -- **Complete Data**: Gets every CI run in the specified range (no sampling) -- **No Limit**: Ignores the `--limit` parameter -- **Flexible Range**: Specify any date range you need -- **Historical Analysis**: Perfect for investigating specific time periods - -#### Date Format: -- Use `YYYY-MM-DD` format (e.g., `2024-12-01`) -- Both parameters are optional: - - Only `--start-date`: Gets all runs from that date to now - - Only `--end-date`: Gets all runs from 30 days ago to that date - - Both: Gets all runs in the specified range - -### 3. Sequential Collection (Traditional) - -**When to use**: Quick checks or when you only need recent data. - -- **Default behavior** for `--limit < 500` -- Gets the most recent CI runs in chronological order -- Fast and simple for immediate analysis - -### Comparison - -| Strategy | Use Case | Time Coverage | Data Completeness | API Efficiency | -|----------|----------|---------------|-------------------|----------------| -| **Uniform Sampling** | Daily monitoring, trends | ~30 days | Sampled | High | -| **Date Range** | Historical analysis | Any range | Complete | Variable | -| **Sequential** | Quick checks | 3-4 days | Complete (recent) | High | - -### Benefits - -- **Flexible Analysis**: Choose the right strategy for your needs -- **Extended Coverage**: Up to 30 days with sampling, unlimited with date ranges -- **Complete Data**: Get every run in a specific period when needed -- **API Efficiency**: Optimized for different use patterns - -## Parameters - -### CI Analyzer Parameters - -| Parameter | Default | Description | -|-----------|---------|-------------| -| `--token` | Required | GitHub Personal Access Token | -| `--limit` | 100 | Number of CI runs to analyze | -| `--output` | ci_analysis.json | Output JSON file for detailed data | - -### Performance Analyzer Parameters - -| Parameter | Default | Description | -|-----------|---------|-------------| -| `--token` | Required | GitHub Personal Access Token | -| `--limit` | 100 | Number of PR Test runs to analyze (ignored when using date range) | -| `--output-dir` | performance_tables | Output directory for CSV tables and PNG charts | -| `--start-date` | None | Start date for date range query (YYYY-MM-DD format) | -| `--end-date` | None | End date for date range query (YYYY-MM-DD format) | -| `--upload-to-github` | False | Upload results to sglang-bot/sglang-ci-data repository | - -### Test Balance Analyzer Parameters - -| Parameter | Default | Description | -|-----------|---------|-------------| -| `--token` | Required | GitHub Personal Access Token | -| `--limit` | 1000 | Number of CI runs to analyze | -| `--output` | test_balance_report.json | Output JSON file for detailed analysis data | - -### Failures Analyzer Parameters +### Failures Analyzer parameters | Parameter | Default | Description | |-----------|---------|-------------| | `--token` | Required | GitHub Personal Access Token | | `--limit` | 500 | Number of workflow runs to analyze | | `--threshold` | 3 | Alert threshold for consecutive failures | -| `--output` | None | Output JSON file (optional, only writes if specified) | +| `--output` | None | Output JSON file (optional) | -## Getting GitHub Token +## Historical note -1. Go to [GitHub Settings > Personal Access Tokens](https://github.com/settings/tokens) -2. Click "Generate new token" > "Generate new token (classic)" -3. **Important**: Select the following permissions: - - `repo` (Full control of private repositories) - **Required for accessing repository data** - - `workflow` (Update GitHub Action workflows) - **Required for reading CI/CD data** -4. Copy the generated token and use it as `YOUR_GITHUB_TOKEN` - -**Note**: Without the `repo` and `workflow` permissions, the tool will not be able to access CI run data and will return 404 errors. +The former **CI Monitor** workflow (`ci-monitor.yml`) and its analyzers (`ci_analyzer.py`, `ci_analyzer_perf.py`, `ci_analyzer_balance.py`) were removed as redundant; use this failure monitor workflow and scripts for ongoing CI health alerts. diff --git a/scripts/ci_monitor/ci_analyzer.py b/scripts/ci_monitor/ci_analyzer.py deleted file mode 100755 index 274e74842..000000000 --- a/scripts/ci_monitor/ci_analyzer.py +++ /dev/null @@ -1,1213 +0,0 @@ -#!/usr/bin/env python3 - -import argparse -import base64 -import json -import os -import re -import sys -import time -from collections import Counter, defaultdict -from datetime import datetime, timedelta -from typing import Dict, List, Optional - -import requests - - -class SGLangCIAnalyzer: - - 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-CI-Analyzer/1.0", - } - self.session = requests.Session() - self.session.headers.update(self.headers) - - # Nightly workflow files to monitor - self.nightly_workflows = [ - "nightly-test-nvidia.yml", - "nightly-test-amd.yml", - "nightly-test-intel.yml", - ] - - # Performance metric patterns for parsing logs - self.perf_patterns = { - "output_throughput": re.compile( - r"Output token throughput \(tok/s\):\s*([\d.]+)" - ), - "input_throughput": re.compile( - r"Input token throughput \(tok/s\):\s*([\d.]+)" - ), - "latency": re.compile(r"Median E2E Latency \(ms\):\s*([\d.]+)"), - "ttft": re.compile(r"Median TTFT \(ms\):\s*([\d.]+)"), - "accept_length": re.compile(r"Accept length:\s*([\d.]+)"), - "accuracy": re.compile(r"Accuracy:\s*([\d.]+)"), - "gsm8k_score": re.compile(r"GSM8K Score:\s*([\d.]+)"), - } - - # Historical data repository - self.data_repo = "sglang-bot/sglang-ci-data" - self.data_branch = "main" - - def get_recent_runs(self, limit: int = 100, branch: str = None) -> List[Dict]: - branch_info = f" from branch '{branch}'" if branch else "" - print(f"Fetching {limit} recent CI runs{branch_info}...") - - 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} - if branch: - params["branch"] = branch - - 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 analyze_ci_failures(self, runs: List[Dict]) -> Dict: - print( - "Analyzing CI failure data (pr-test.yml, quantization-test.yml, nightly-test.yml jobs only)..." - ) - - job_categories = { - "build": [ - "build-test", - "sgl-kernel-build-wheels", - ], - "unit-test": [ - "stage-a-test-1", - "unit-test-backend-1-gpu", - "unit-test-backend-2-gpu", - "stage-b-test-4-gpu-b200", - "unit-test-backend-4-gpu", - "unit-test-backend-8-gpu", - ], - "performance": [ - "performance-test-1-gpu-part-1", - "performance-test-1-gpu-part-2", - "performance-test-1-gpu-part-3", - "performance-test-2-gpu", - ], - "accuracy": [ - "accuracy-test-1-gpu", - "accuracy-test-2-gpu", - ], - "mla-test": [ - "sgl-kernel-mla-test", - ], - "deepep": [ - "unit-test-deepep-4-gpu", - "unit-test-deepep-8-gpu", - ], - "per-commit": [ - "per-commit-8-gpu-h20", - ], - "nightly": [ - # NVIDIA job names (nightly-test-nvidia.yml) - "nightly-test-general-1-gpu-runner", - "nightly-test-general-4-gpu-h100", - "nightly-test-general-8-gpu-h200", - "nightly-test-general-8-gpu-h20", - "nightly-test-general-8-gpu-b200", - "nightly-test-text-accuracy-2-gpu-runner", - "nightly-test-text-perf-2-gpu-runner", - "nightly-test-vlm-accuracy-2-gpu-runner", - "nightly-test-vlm-perf-2-gpu-runner", - "nightly-test-perf-4-gpu-b200", - "nightly-test-perf-8-gpu-b200", - # AMD job names (nightly-test-amd.yml) - "nightly-test", # AMD uses this generic name with matrix - ], - "integration": [ - "run-all-notebooks", - "quantization-test", - "test-disaggregation", - ], - "b200": [ - "unit-test-backend-4-gpu-b200", - ], - "gb200": [ - "unit-test-backend-4-gpu-gb200", - ], - } - - stats = { - "total_runs": len(runs), - "failed_runs": 0, - "successful_runs": 0, - "cancelled_runs": 0, - "skipped_runs": 0, - "category_failures": defaultdict(int), - "job_failures": defaultdict(int), - "failure_patterns": defaultdict(int), - "job_failure_links": defaultdict( - list - ), # Store recent failure links for each job - "job_last_success": {}, # Store last successful run for each job - "performance_metrics": defaultdict( - lambda: defaultdict(list) - ), # Track performance metrics for nightly jobs - } - - total_runs = len(runs) - for i, run in enumerate(runs, 1): - if i % max(1, min(50, total_runs // 10)) == 0 or i == total_runs: - progress = (i / total_runs) * 100 - print(f"Progress: {i}/{total_runs} ({progress:.1f}%)") - - run_status = run.get("conclusion", "unknown") - workflow_name = run.get("name", "Unknown") - run_id = run.get("id") - run_number = run.get("run_number") - created_at = run.get("created_at") - - if run_status == "failure": - stats["failed_runs"] += 1 - elif run_status == "success": - stats["successful_runs"] += 1 - elif run_status == "cancelled": - stats["cancelled_runs"] += 1 - elif run_status == "skipped": - stats["skipped_runs"] += 1 - - jobs = self._get_job_details(run_id) - run_url = f"https://github.com/{self.repo}/actions/runs/{run_id}" - pr_info = self._get_pr_info(run) - - for job in jobs: - job_name = job.get("name", "Unknown") - job_conclusion = job.get("conclusion", "unknown") - - target_jobs = [ - "check-changes", - "sgl-kernel-build-wheels", - "sgl-kernel-unit-test", - "sgl-kernel-mla-test", - "sgl-kernel-benchmark-test", - "stage-a-test-1", - "unit-test-backend-1-gpu", - "unit-test-backend-2-gpu", - "stage-b-test-4-gpu-b200", - "unit-test-backend-4-gpu", - "unit-test-backend-8-gpu-h200", - "unit-test-backend-8-gpu-h20", - "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", - "unit-test-deepep-4-gpu", - "unit-test-deepep-8-gpu", - "unit-test-backend-8-gpu-deepseek-v32", - "unit-test-backend-4-gpu-b200", - "unit-test-backend-4-gpu-gb200", - "quantization-test", - # NVIDIA job names (nightly-test-nvidia.yml) - "nightly-test-general-1-gpu-runner", - "nightly-test-general-4-gpu-h100", - "nightly-test-general-8-gpu-h200", - "nightly-test-general-8-gpu-h20", - "nightly-test-general-8-gpu-b200", - "nightly-test-text-accuracy-2-gpu-runner", - "nightly-test-text-perf-2-gpu-runner", - "nightly-test-vlm-accuracy-2-gpu-runner", - "nightly-test-vlm-perf-2-gpu-runner", - "nightly-test-perf-4-gpu-b200", - "nightly-test-perf-8-gpu-b200", - # AMD job names (nightly-test-amd.yml) - "nightly-test", - ] - - if job_name in target_jobs: - if job_conclusion == "success": - stats["job_last_success"][job_name] = { - "url": run_url, - "run_number": run_number, - "created_at": created_at, - "pr_info": pr_info, - } - - # Parse performance metrics from successful nightly jobs - if job_name in job_categories["nightly"] and ( - "perf" in job_name.lower() - or "accuracy" in job_name.lower() - or "eval" in job_name.lower() - ): - job_id = job.get("id") - logs = self.get_job_logs(job_id) - if logs: - metrics = self.parse_metrics_from_logs(logs, job_name) - for metric_name, values in metrics.items(): - if values: - for value in values: - stats["performance_metrics"][job_name][ - metric_name - ].append( - { - "value": value, - "timestamp": created_at, - "run_id": run_id, - "run_url": run_url, - } - ) - - elif job_conclusion == "failure": - stats["job_failures"][job_name] += 1 - - if len(stats["job_failure_links"][job_name]) < 3: - stats["job_failure_links"][job_name].append( - { - "url": run_url, - "run_number": run_number, - "created_at": created_at, - "pr_info": pr_info, - } - ) - - for category, jobs_list in job_categories.items(): - if any( - job_pattern in job_name for job_pattern in jobs_list - ): - stats["category_failures"][category] += 1 - break - - self._analyze_failure_pattern(job, stats) - - time.sleep(0.1) - - return stats - - def _get_job_details(self, run_id: int) -> List[Dict]: - url = f"{self.base_url}/repos/{self.repo}/actions/runs/{run_id}/jobs" - try: - response = self.session.get(url) - response.raise_for_status() - return response.json().get("jobs", []) - except: - return [] - - def _get_pr_info(self, run: Dict) -> Dict: - pr_info = { - "pr_number": None, - "author": run.get("head_commit", {}) - .get("author", {}) - .get("name", "Unknown"), - "head_sha": run.get("head_sha", ""), - "head_branch": run.get("head_branch", ""), - } - - pull_requests = run.get("pull_requests", []) - if pull_requests: - pr_info["pr_number"] = pull_requests[0].get("number") - - return pr_info - - def _analyze_failure_pattern(self, job: Dict, stats: Dict): - job_name = job.get("name", "") - steps = job.get("steps", []) - - for step in steps: - if step.get("conclusion") == "failure": - step_name = step.get("name", "") - - if "timeout" in step_name.lower(): - stats["failure_patterns"]["Timeout"] += 1 - elif "build" in step_name.lower() or "build" in job_name.lower(): - stats["failure_patterns"]["Build Failure"] += 1 - elif "install" in step_name.lower() or "dependency" in job_name.lower(): - stats["failure_patterns"]["Dependency Installation Failure"] += 1 - elif "unit" in job_name.lower() or "unit-test" in job_name.lower(): - stats["failure_patterns"]["Unit Test Failure"] += 1 - elif "performance" in job_name.lower() or "perf" in job_name.lower(): - stats["failure_patterns"]["Performance Test Failure"] += 1 - elif "accuracy" in job_name.lower(): - stats["failure_patterns"]["Accuracy Test Failure"] += 1 - elif "mla" in job_name.lower(): - stats["failure_patterns"]["MLA Test Failure"] += 1 - elif "deepep" in job_name.lower(): - stats["failure_patterns"]["DeepEP Test Failure"] += 1 - elif "nightly" in job_name.lower(): - stats["failure_patterns"]["Nightly Test Failure"] += 1 - elif "notebook" in job_name.lower(): - stats["failure_patterns"]["Notebook Test Failure"] += 1 - elif "disaggregation" in job_name.lower(): - stats["failure_patterns"]["Disaggregation Test Failure"] += 1 - elif "h20" in job_name.lower() or "h200" in job_name.lower(): - stats["failure_patterns"]["H20/H200 GPU Failure"] += 1 - elif "b200" in job_name.lower(): - stats["failure_patterns"]["B200 GPU Failure"] += 1 - elif "gpu" in job_name.lower(): - stats["failure_patterns"]["GPU Related Failure"] += 1 - else: - stats["failure_patterns"]["Other"] += 1 - - def generate_report(self, stats: Dict): - print("\n" + "=" * 60) - print("SGLang CI Analysis Report (Target Workflows Only)") - print("=" * 60) - - total = stats["total_runs"] - failed = stats["failed_runs"] - success = stats["successful_runs"] - cancelled = stats["cancelled_runs"] - skipped = stats["skipped_runs"] - success_rate = (success / total * 100) if total > 0 else 0 - - print(f"\nOverall Statistics:") - print(f" Total runs: {total}") - print(f" Successful: {success}") - print(f" Failed: {failed}") - print(f" Cancelled: {cancelled}") - print(f" Skipped: {skipped}") - print(f" Success rate: {success_rate:.1f}%") - - if stats["category_failures"]: - print(f"\nCategory Failure Statistics:") - for category, count in sorted( - stats["category_failures"].items(), key=lambda x: x[1], reverse=True - ): - print(f" {category}: {count} failures") - - if stats["job_failures"]: - print(f"\nMost Frequently Failed Jobs (Top 50):") - for i, (job, count) in enumerate( - sorted(stats["job_failures"].items(), key=lambda x: x[1], reverse=True)[ - :50 - ], - 1, - ): - print(f" {i:2d}. {job}: {count} times") - - if job in stats["job_last_success"]: - last_success = stats["job_last_success"][job] - success_date = datetime.fromisoformat( - last_success["created_at"].replace("Z", "+00:00") - ) - pr_info = last_success["pr_info"] - - pr_text = "" - if pr_info["pr_number"]: - pr_text = ( - f" (PR #{pr_info['pr_number']} by {pr_info['author']})" - ) - else: - pr_text = f" by {pr_info['author']}" - - print( - f" Last Success: Run #{last_success['run_number']} ({success_date.strftime('%Y-%m-%d %H:%M')}){pr_text}: {last_success['url']}" - ) - - if ( - job in stats["job_failure_links"] - and stats["job_failure_links"][job] - ): - print(" Recent Failures:") - for link_info in stats["job_failure_links"][job]: - created_at = datetime.fromisoformat( - link_info["created_at"].replace("Z", "+00:00") - ) - - pr_info = link_info.get("pr_info", {}) - pr_text = "" - if pr_info.get("pr_number"): - pr_text = f" (PR #{pr_info['pr_number']} by {pr_info.get('author', 'Unknown')})" - else: - pr_text = f" by {pr_info.get('author', 'Unknown')}" - - print( - f" - Run #{link_info['run_number']} ({created_at.strftime('%Y-%m-%d %H:%M')}){pr_text}: {link_info['url']}" - ) - - if stats["failure_patterns"]: - print(f"\nFailure Pattern Analysis:") - for pattern, count in sorted( - stats["failure_patterns"].items(), key=lambda x: x[1], reverse=True - ): - print(f" {pattern}: {count} times") - - print("\n" + "=" * 60) - - def save_detailed_report(self, stats: Dict, output_file: str = "ci_analysis.json"): - with open(output_file, "w", encoding="utf-8") as f: - json.dump(stats, f, ensure_ascii=False, indent=2) - print(f"\nDetailed report saved to: {output_file}") - - def generate_github_summary(self, stats: 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 CI Analysis...") - - summary_lines = [] - summary_lines.append("# SGLang CI Analysis Report (Target Workflows Only)") - summary_lines.append("") - - total = stats["total_runs"] - failed = stats["failed_runs"] - success = stats["successful_runs"] - cancelled = stats["cancelled_runs"] - skipped = stats["skipped_runs"] - success_rate = (success / total * 100) if total > 0 else 0 - - summary_lines.append("## Overall Statistics") - summary_lines.append("") - summary_lines.append("| Metric | Count | Percentage |") - summary_lines.append("|--------|-------|------------|") - summary_lines.append(f"| Total Runs | {total} | 100% |") - summary_lines.append( - f"| Successful | {success} | {success/total*100:.1f}% |" - ) - summary_lines.append(f"| Failed | {failed} | {failed/total*100:.1f}% |") - summary_lines.append( - f"| Cancelled | {cancelled} | {cancelled/total*100:.1f}% |" - ) - summary_lines.append(f"| Skipped | {skipped} | {skipped/total*100:.1f}% |") - summary_lines.append(f"| **Success Rate** | **{success_rate:.1f}%** | - |") - summary_lines.append("") - - if stats["category_failures"]: - summary_lines.append("## Category Failure Statistics") - summary_lines.append("") - summary_lines.append("| Category | Failures |") - summary_lines.append("|----------|----------|") - for category, count in sorted( - stats["category_failures"].items(), key=lambda x: x[1], reverse=True - ): - summary_lines.append(f"| {category} | {count} |") - summary_lines.append("") - - if stats["job_failures"]: - summary_lines.append("## Most Frequently Failed Jobs (Top 20)") - summary_lines.append("") - - top_failures = sorted( - stats["job_failures"].items(), key=lambda x: x[1], reverse=True - )[:20] - - for i, (job, count) in enumerate(top_failures, 1): - summary_lines.append(f"### {i}. `{job}` ({count} failures)") - summary_lines.append("") - - if job in stats["job_last_success"]: - last_success = stats["job_last_success"][job] - success_date = datetime.fromisoformat( - last_success["created_at"].replace("Z", "+00:00") - ) - pr_info = last_success["pr_info"] - - pr_text = "" - if pr_info["pr_number"]: - pr_text = ( - f" (PR #{pr_info['pr_number']} by {pr_info['author']})" - ) - else: - pr_text = f" by {pr_info['author']}" - - summary_lines.append( - f"**Last Success:** [Run #{last_success['run_number']}]({last_success['url']}) ({success_date.strftime('%Y-%m-%d %H:%M')}){pr_text}" - ) - summary_lines.append("") - - if ( - job in stats["job_failure_links"] - and stats["job_failure_links"][job] - ): - summary_lines.append("**Recent Failures:**") - for link_info in stats["job_failure_links"][job]: - created_at = datetime.fromisoformat( - link_info["created_at"].replace("Z", "+00:00") - ) - - pr_info = link_info.get("pr_info", {}) - pr_text = "" - if pr_info.get("pr_number"): - pr_text = f" (PR #{pr_info['pr_number']} by {pr_info.get('author', 'Unknown')})" - else: - pr_text = f" by {pr_info.get('author', 'Unknown')}" - - summary_lines.append( - f"- [Run #{link_info['run_number']}]({link_info['url']}) ({created_at.strftime('%Y-%m-%d %H:%M')}){pr_text}" - ) - summary_lines.append("") - - if stats["failure_patterns"]: - summary_lines.append("## Failure Pattern Analysis") - summary_lines.append("") - summary_lines.append("| Pattern | Count |") - summary_lines.append("|---------|-------|") - for pattern, count in sorted( - stats["failure_patterns"].items(), key=lambda x: x[1], reverse=True - ): - summary_lines.append(f"| {pattern} | {count} |") - summary_lines.append("") - - # Performance metrics section for nightly jobs - if stats.get("performance_metrics"): - summary_lines.append("## Nightly Test Performance Metrics") - summary_lines.append("") - summary_lines.append("| Job | Metric | Latest Value | Count | Trend |") - summary_lines.append("|-----|--------|--------------|-------|-------|") - - for job_name in sorted(stats["performance_metrics"].keys()): - job_metrics = stats["performance_metrics"][job_name] - for metric_name in sorted(job_metrics.keys()): - metric_data = job_metrics[metric_name] - if metric_data: - # Calculate average of recent values - values = [m["value"] for m in metric_data] - avg_value = sum(values) / len(values) - count = len(values) - - # Simple trend: compare first half vs second half - trend_indicator = "➡️" - if len(values) >= 4: - first_half = values[: len(values) // 2] - second_half = values[len(values) // 2 :] - first_avg = sum(first_half) / len(first_half) - second_avg = sum(second_half) / len(second_half) - - if first_avg > 0: - change_pct = ( - (second_avg - first_avg) / first_avg - ) * 100 - - # For throughput metrics, up is good - # For latency/ttft metrics, down is good - if "throughput" in metric_name.lower(): - if change_pct > 10: - trend_indicator = f"📈 +{change_pct:.1f}%" - elif change_pct < -10: - trend_indicator = f"⚠️ 📉 {change_pct:.1f}%" - else: - trend_indicator = f"➡️ {change_pct:+.1f}%" - elif ( - "latency" in metric_name.lower() - or "ttft" in metric_name.lower() - ): - if change_pct < -10: - trend_indicator = f"📈 {change_pct:.1f}%" - elif change_pct > 10: - trend_indicator = ( - f"⚠️ 📉 +{change_pct:.1f}%" - ) - else: - trend_indicator = f"➡️ {change_pct:+.1f}%" - else: - trend_indicator = f"➡️ {change_pct:+.1f}%" - - summary_lines.append( - f"| {job_name} | {metric_name} | {avg_value:.2f} | {count} | {trend_indicator} |" - ) - - summary_lines.append("") - - with open(github_step_summary, "w", encoding="utf-8") as f: - f.write("\n".join(summary_lines)) - f.write("\n\n---\n\n") - - print("GitHub Actions summary generated successfully") - - except Exception as e: - print(f"Failed to generate GitHub Actions summary: {e}") - - def get_nightly_runs(self, days: int = 2) -> List[Dict]: - """Get nightly test workflow runs from the last N days""" - print(f"Fetching nightly test runs from the last {days} days...") - - since_date = (datetime.now() - timedelta(days=days)).isoformat() - all_runs = [] - - for workflow_file in self.nightly_workflows: - print(f" Fetching from {workflow_file}...") - page = 1 - per_page = 10 # Nightly runs once per day, so 10 runs covers ~10 days max - workflow_runs = [] - max_runs_per_workflow = days * 5 # Allow up to 5 runs per day per workflow - - while len(workflow_runs) < max_runs_per_workflow: - url = f"{self.base_url}/repos/{self.repo}/actions/runs" - params = { - "workflow_id": workflow_file, - "per_page": per_page, - "page": page, - "created": f">={since_date}", - } - - try: - response = self.session.get(url, params=params) - response.raise_for_status() - data = response.json() - - if not data.get("workflow_runs"): - break - - runs = data["workflow_runs"] - workflow_runs.extend(runs) - - if len(runs) < per_page: - break - - page += 1 - time.sleep(0.1) - - except requests.exceptions.RequestException as e: - print(f" Warning: Error fetching from {workflow_file}: {e}") - break - - print(f" Fetched {len(workflow_runs)} runs from {workflow_file}") - all_runs.extend(workflow_runs) - - print(f"Total nightly runs fetched: {len(all_runs)}") - return all_runs - - def get_job_logs(self, job_id: int) -> Optional[str]: - """Get logs for a specific job""" - url = f"{self.base_url}/repos/{self.repo}/actions/jobs/{job_id}/logs" - try: - response = self.session.get(url) - response.raise_for_status() - return response.text - except requests.exceptions.RequestException as e: - print(f" Warning: Could not fetch logs for job {job_id}: {e}") - return None - - def parse_metrics_from_logs( - self, logs: str, job_name: str - ) -> Dict[str, List[float]]: - """Parse performance metrics from job logs""" - metrics = defaultdict(list) - - if not logs: - return metrics - - for line in logs.split("\n"): - for metric_name, pattern in self.perf_patterns.items(): - match = pattern.search(line) - if match: - try: - value = float(match.group(1)) - metrics[metric_name].append(value) - except (ValueError, IndexError): - continue - - return dict(metrics) - - def analyze_nightly_with_metrics(self, runs: List[Dict]) -> Dict: - """Analyze nightly test runs including performance metrics""" - print("Analyzing nightly test data with performance metrics...") - - # Get nightly job names from the existing job categories - nightly_jobs = [ - # NVIDIA job names (nightly-test-nvidia.yml) - "nightly-test-general-1-gpu-runner", - "nightly-test-general-4-gpu-h100", - "nightly-test-general-8-gpu-h200", - "nightly-test-general-8-gpu-h20", - "nightly-test-general-8-gpu-b200", - "nightly-test-text-accuracy-2-gpu-runner", - "nightly-test-text-perf-2-gpu-runner", - "nightly-test-vlm-accuracy-2-gpu-runner", - "nightly-test-vlm-perf-2-gpu-runner", - "nightly-test-perf-4-gpu-b200", - "nightly-test-perf-8-gpu-b200", - # AMD job names (nightly-test-amd.yml) - "nightly-test", - # Intel job names (nightly-test-intel.yml) - "placeholder", - ] - - stats = { - "total_runs": len(runs), - "successful_runs": 0, - "failed_runs": 0, - "cancelled_runs": 0, - "job_stats": defaultdict( - lambda: { - "total": 0, - "success": 0, - "failure": 0, - "recent_failures": [], - "avg_duration_minutes": 0, - "durations": [], - "performance_metrics": defaultdict(list), - } - ), - "daily_stats": defaultdict( - lambda: { - "total": 0, - "success": 0, - "failure": 0, - } - ), - } - - for i, run in enumerate(runs, 1): - if i % 10 == 0: - print(f"Processed {i}/{len(runs)} runs...") - - run_status = run.get("conclusion", "unknown") - run_id = run.get("id") - run_number = run.get("run_number") - created_at = run.get("created_at") - run_url = f"https://github.com/{self.repo}/actions/runs/{run_id}" - - # Track daily stats - date_str = created_at.split("T")[0] if created_at else "unknown" - stats["daily_stats"][date_str]["total"] += 1 - - if run_status == "success": - stats["successful_runs"] += 1 - stats["daily_stats"][date_str]["success"] += 1 - elif run_status == "failure": - stats["failed_runs"] += 1 - stats["daily_stats"][date_str]["failure"] += 1 - elif run_status == "cancelled": - stats["cancelled_runs"] += 1 - - # Analyze individual jobs - jobs = self._get_job_details(run_id) - for job in jobs: - job_name = job.get("name", "Unknown") - job_conclusion = job.get("conclusion", "unknown") - job_id = job.get("id") - started_at = job.get("started_at") - completed_at = job.get("completed_at") - - # Only track nightly test jobs - if job_name not in nightly_jobs: - continue - - job_stat = stats["job_stats"][job_name] - job_stat["total"] += 1 - - if job_conclusion == "success": - job_stat["success"] += 1 - - # For successful performance/accuracy jobs, fetch metrics - if ( - "perf" in job_name.lower() - or "accuracy" in job_name.lower() - or "eval" in job_name.lower() - ): - logs = self.get_job_logs(job_id) - if logs: - metrics = self.parse_metrics_from_logs(logs, job_name) - for metric_name, values in metrics.items(): - if values: - job_stat["performance_metrics"][metric_name].extend( - [ - { - "value": v, - "timestamp": created_at, - "run_id": run_id, - "job_name": job_name, - } - for v in values - ] - ) - - elif job_conclusion == "failure": - job_stat["failure"] += 1 - - if len(job_stat["recent_failures"]) < 5: - job_stat["recent_failures"].append( - { - "run_url": run_url, - "run_number": run_number, - "created_at": created_at, - "job_url": job.get("html_url"), - } - ) - - # Track duration - if started_at and completed_at: - try: - start = datetime.fromisoformat( - started_at.replace("Z", "+00:00") - ) - end = datetime.fromisoformat( - completed_at.replace("Z", "+00:00") - ) - duration_minutes = (end - start).total_seconds() / 60 - job_stat["durations"].append(duration_minutes) - except: - pass - - time.sleep(0.1) - - # Calculate average durations - for job_name, job_stat in stats["job_stats"].items(): - if job_stat["durations"]: - job_stat["avg_duration_minutes"] = sum(job_stat["durations"]) / len( - job_stat["durations"] - ) - del job_stat["durations"] - - return stats - - def generate_nightly_report(self, stats: Dict, output_file: str = None): - """Generate a report for nightly test analysis""" - print("\n" + "=" * 80) - print("NIGHTLY TEST MONITOR REPORT") - print("=" * 80) - print(f"Report Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") - print(f"Total Runs Analyzed: {stats['total_runs']}") - print( - f"Successful: {stats['successful_runs']} " - f"({stats['successful_runs']/max(1, stats['total_runs'])*100:.1f}%)" - ) - print( - f"Failed: {stats['failed_runs']} " - f"({stats['failed_runs']/max(1, stats['total_runs'])*100:.1f}%)" - ) - print(f"Cancelled: {stats['cancelled_runs']}") - print("=" * 80) - - # Daily trend - print("\nDAILY TRENDS:") - print("-" * 80) - daily_stats = sorted(stats["daily_stats"].items(), reverse=True)[:7] - for date, day_stats in daily_stats: - success_rate = (day_stats["success"] / max(1, day_stats["total"])) * 100 - print( - f"{date}: {day_stats['total']} runs, {day_stats['success']} success " - f"({success_rate:.1f}%), {day_stats['failure']} failed" - ) - - # Job statistics - print("\nJOB STATISTICS:") - print("-" * 80) - print( - f"{'Job Name':<50} {'Total':<8} {'Success':<8} {'Failed':<8} " - f"{'Rate':<8} {'Avg Duration'}" - ) - print("-" * 80) - - job_stats_sorted = sorted( - stats["job_stats"].items(), key=lambda x: x[1]["failure"], reverse=True - ) - - for job_name, job_stat in job_stats_sorted: - total = job_stat["total"] - success = job_stat["success"] - failure = job_stat["failure"] - success_rate = (success / max(1, total)) * 100 - avg_duration = job_stat["avg_duration_minutes"] - - print( - f"{job_name:<50} {total:<8} {success:<8} {failure:<8} " - f"{success_rate:>6.1f}% {avg_duration:>7.1f}m" - ) - - # Show performance metrics if available - if job_stat.get("performance_metrics"): - perf_metrics = job_stat["performance_metrics"] - print(f" Performance metrics:") - - for metric_name, metric_data in perf_metrics.items(): - if metric_data: - values = [m["value"] for m in metric_data] - avg_value = sum(values) / len(values) - print(f" - {metric_name}: {avg_value:.2f} (n={len(values)})") - - # Show recent failures - if job_stat["recent_failures"]: - print(f" Recent failures:") - for failure in job_stat["recent_failures"][:3]: - print(f" - Run #{failure['run_number']}: {failure['run_url']}") - - print("=" * 80) - - # Save to file if requested - if output_file: - with open(output_file, "w") as f: - json.dump(stats, f, indent=2, default=str) - print(f"\nDetailed stats saved to: {output_file}") - - def generate_nightly_github_summary(self, stats: Dict): - """Generate GitHub Actions summary for nightly test analysis""" - try: - github_step_summary = os.environ.get("GITHUB_STEP_SUMMARY") - if not github_step_summary: - print( - "Not running in GitHub Actions, skipping nightly summary generation" - ) - return - - print("Generating GitHub Actions summary for Nightly Analysis...") - - summary_lines = [] - summary_lines.append("# Nightly Test Monitor Report") - summary_lines.append("") - summary_lines.append( - f"**Report Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" - ) - summary_lines.append("") - - # Overall statistics - total = stats["total_runs"] - success = stats["successful_runs"] - failed = stats["failed_runs"] - cancelled = stats["cancelled_runs"] - - summary_lines.append("## Overall Statistics") - summary_lines.append("") - summary_lines.append("| Metric | Count | Percentage |") - summary_lines.append("|--------|-------|------------|") - summary_lines.append(f"| Total Runs | {total} | 100% |") - summary_lines.append( - f"| Successful | {success} | {success/max(1,total)*100:.1f}% |" - ) - summary_lines.append( - f"| Failed | {failed} | {failed/max(1,total)*100:.1f}% |" - ) - summary_lines.append( - f"| Cancelled | {cancelled} | {cancelled/max(1,total)*100:.1f}% |" - ) - summary_lines.append("") - - # Daily trends - summary_lines.append("## Daily Trends") - summary_lines.append("") - summary_lines.append( - "| Date | Total Runs | Success | Failed | Success Rate |" - ) - summary_lines.append( - "|------|------------|---------|--------|--------------|" - ) - - daily_stats = sorted(stats["daily_stats"].items(), reverse=True)[:7] - for date, day_stats in daily_stats: - success_rate = (day_stats["success"] / max(1, day_stats["total"])) * 100 - summary_lines.append( - f"| {date} | {day_stats['total']} | {day_stats['success']} | " - f"{day_stats['failure']} | {success_rate:.1f}% |" - ) - summary_lines.append("") - - # Job statistics with performance metrics - if stats["job_stats"]: - summary_lines.append("## Job Statistics") - summary_lines.append("") - - job_stats_sorted = sorted( - stats["job_stats"].items(), - key=lambda x: x[1]["failure"], - reverse=True, - ) - - for job_name, job_stat in job_stats_sorted: - total_job = job_stat["total"] - success_job = job_stat["success"] - failure_job = job_stat["failure"] - success_rate_job = (success_job / max(1, total_job)) * 100 - avg_duration = job_stat["avg_duration_minutes"] - - summary_lines.append(f"### {job_name}") - summary_lines.append("") - summary_lines.append( - f"**Stats:** {total_job} runs | {success_job} success ({success_rate_job:.1f}%) | " - f"{failure_job} failed | Avg duration: {avg_duration:.1f}m" - ) - summary_lines.append("") - - # Performance metrics - if job_stat.get("performance_metrics"): - summary_lines.append("**Performance Metrics:**") - summary_lines.append("") - summary_lines.append("| Metric | Avg Value | Samples |") - summary_lines.append("|--------|-----------|---------|") - - for metric_name, metric_data in job_stat[ - "performance_metrics" - ].items(): - if metric_data: - values = [m["value"] for m in metric_data] - avg_value = sum(values) / len(values) - summary_lines.append( - f"| {metric_name} | {avg_value:.2f} | {len(values)} |" - ) - summary_lines.append("") - - # Recent failures - if job_stat["recent_failures"]: - summary_lines.append("**Recent Failures:**") - for failure in job_stat["recent_failures"][:3]: - summary_lines.append( - f"- [Run #{failure['run_number']}]({failure['run_url']})" - ) - summary_lines.append("") - - with open(github_step_summary, "a", encoding="utf-8") as f: - f.write("\n".join(summary_lines)) - f.write("\n\n---\n\n") - - print("GitHub Actions nightly summary generated successfully") - - except Exception as e: - print(f"Failed to generate nightly GitHub Actions summary: {e}") - - def detect_nightly_regressions(self, stats: Dict) -> List[Dict]: - """Detect regressions in nightly tests""" - regressions = [] - - for job_name, job_stat in stats["job_stats"].items(): - total = job_stat["total"] - failure = job_stat["failure"] - - if total > 0: - failure_rate = (failure / total) * 100 - - # Flag jobs with high failure rates - if failure_rate > 30: - regressions.append( - { - "job_name": job_name, - "type": "high_failure_rate", - "failure_rate": failure_rate, - "total_runs": total, - "failures": failure, - } - ) - - # Flag jobs with recent consecutive failures - recent_failures = len(job_stat["recent_failures"]) - if recent_failures >= 3: - regressions.append( - { - "job_name": job_name, - "type": "consecutive_failures", - "recent_failure_count": recent_failures, - } - ) - - if regressions: - print("\n" + "=" * 80) - print("REGRESSIONS DETECTED:") - print("=" * 80) - for regression in regressions: - print(f"\nJob: {regression['job_name']}") - if regression["type"] == "high_failure_rate": - print( - f" High failure rate: {regression['failure_rate']:.1f}% " - f"({regression['failures']}/{regression['total_runs']})" - ) - elif regression["type"] == "consecutive_failures": - print( - f" {regression['recent_failure_count']} recent consecutive failures" - ) - print("=" * 80) - - return regressions - - -def main(): - parser = argparse.ArgumentParser(description="SGLang CI Analyzer") - parser.add_argument("--token", required=True, help="GitHub Personal Access Token") - parser.add_argument( - "--mode", - choices=["ci", "nightly"], - default="ci", - help="Analysis mode: 'ci' for general CI analysis, 'nightly' for nightly test monitoring (default: ci)", - ) - parser.add_argument( - "--limit", - type=int, - default=100, - help="Number of runs to analyze (for ci mode, default: 100)", - ) - parser.add_argument( - "--days", - type=int, - default=2, - help="Number of days to analyze (for nightly mode, default: 2)", - ) - parser.add_argument( - "--output", - help="Output file for detailed stats (JSON)", - ) - parser.add_argument( - "--branch", - default=None, - help="Filter runs by branch (default: None - all branches). Specify branch name to filter.", - ) - - args = parser.parse_args() - - analyzer = SGLangCIAnalyzer(args.token) - - try: - if args.mode == "nightly": - # Nightly test monitoring mode - runs = analyzer.get_nightly_runs(days=args.days) - - if not runs: - print("No nightly test runs found in the specified time period.") - sys.exit(1) - - stats = analyzer.analyze_nightly_with_metrics(runs) - analyzer.generate_nightly_report(stats, args.output) - analyzer.generate_nightly_github_summary(stats) - regressions = analyzer.detect_nightly_regressions(stats) - - # Report regressions but don't stop the monitor - if regressions: - print("\n⚠️ Regressions detected - see report above") - else: - print("\n✓ No significant regressions detected") - sys.exit(0) - - else: - # Regular CI analysis mode - branch = args.branch if args.branch else None - runs = analyzer.get_recent_runs(args.limit, branch) - - if not runs: - print("No CI run data found") - return - - stats = analyzer.analyze_ci_failures(runs) - analyzer.generate_report(stats) - - output_file = args.output or "ci_analysis.json" - analyzer.save_detailed_report(stats, output_file) - analyzer.generate_github_summary(stats) - - except Exception as e: - print(f"Error during analysis: {e}") - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/scripts/ci_monitor/ci_analyzer_balance.py b/scripts/ci_monitor/ci_analyzer_balance.py deleted file mode 100755 index 9217b3c81..000000000 --- a/scripts/ci_monitor/ci_analyzer_balance.py +++ /dev/null @@ -1,534 +0,0 @@ -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 = [ - "stage-a-test-1", - "unit-test-backend-1-gpu", - "unit-test-backend-2-gpu", - "stage-b-test-4-gpu-b200", - "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() diff --git a/scripts/ci_monitor/ci_analyzer_perf.py b/scripts/ci_monitor/ci_analyzer_perf.py deleted file mode 100755 index fa8822dda..000000000 --- a/scripts/ci_monitor/ci_analyzer_perf.py +++ /dev/null @@ -1,1375 +0,0 @@ -#!/usr/bin/env python3 -""" -SGLang CI Performance Analyzer - Simplified Version -Collect performance data based on actual log format -""" - -import argparse -import base64 -import csv -import os -import re -import sys -import time -from concurrent.futures import ThreadPoolExecutor, as_completed -from datetime import datetime -from typing import Dict, List, Optional - -import matplotlib.dates as mdates -import matplotlib.pyplot as plt -import pandas as pd -import requests -from matplotlib import rcParams - - -class SGLangPerfAnalyzer: - """SGLang CI Performance Analyzer""" - - 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-Perf-Analyzer/1.0", - } - self.session = requests.Session() - self.session.headers.update(self.headers) - - # Performance test job names - self.performance_jobs = [ - "performance-test-1-gpu-part-1", - "performance-test-1-gpu-part-2", - "performance-test-2-gpu", - ] - - # Strictly match tests and metrics shown in the images - self.target_tests_and_metrics = { - "performance-test-1-gpu-part-1": { - "test_bs1_default": ["output_throughput_token_s"], - "test_online_latency_default": ["median_e2e_latency_ms"], - "test_offline_throughput_default": ["output_throughput_token_s"], - "test_offline_throughput_non_stream_small_batch_size": [ - "output_throughput_token_s" - ], - "test_online_latency_eagle": ["median_e2e_latency_ms", "accept_length"], - "test_lora_online_latency": ["median_e2e_latency_ms", "median_ttft_ms"], - "test_lora_online_latency_with_concurrent_adapter_updates": [ - "median_e2e_latency_ms", - "median_ttft_ms", - ], - }, - "performance-test-1-gpu-part-2": { - "test_offline_throughput_without_radix_cache": [ - "output_throughput_token_s" - ], - "test_offline_throughput_with_triton_attention_backend": [ - "output_throughput_token_s" - ], - "test_offline_throughput_default_fp8": ["output_throughput_token_s"], - "test_vlm_offline_throughput": ["output_throughput_token_s"], - "test_vlm_online_latency": ["median_e2e_latency_ms"], - }, - "performance-test-2-gpu": { - "test_moe_tp2_bs1": ["output_throughput_token_s"], - "test_torch_compile_tp2_bs1": ["output_throughput_token_s"], - "test_moe_offline_throughput_default": ["output_throughput_token_s"], - "test_moe_offline_throughput_without_radix_cache": [ - "output_throughput_token_s" - ], - "test_pp_offline_throughput_default_decode": [ - "output_throughput_token_s" - ], - "test_pp_long_context_prefill": ["input_throughput_token_s"], - }, - } - - # Performance metric patterns - only keep metrics needed in images - self.perf_patterns = { - # Key metrics shown in images - "output_throughput_token_s": r"Output token throughput \(tok/s\):\s*([\d.]+)", - "Output_throughput_token_s": r"Output throughput:\s*([\d.]+)\s*token/s", - "median_e2e_latency_ms": r"Median E2E Latency \(ms\):\s*([\d.]+)", - "median_ttft_ms": r"Median TTFT \(ms\):\s*([\d.]+)", - "accept_length": r"Accept length:\s*([\d.]+)", - "input_throughput_token_s": r"Input token throughput \(tok/s\):\s*([\d.]+)", - } - - # Pre-compile regex patterns for better performance - self.compiled_patterns = { - name: re.compile(pattern, re.IGNORECASE) - for name, pattern in self.perf_patterns.items() - } - - # Pre-compile test pattern - self.test_pattern = re.compile( - r"python3 -m unittest (test_bench_\w+\.TestBench\w+\.test_\w+)" - ) - - # Setup matplotlib fonts and styles - self._setup_matplotlib() - - # GitHub data repository settings - self.data_repo = "sglang-bot/sglang-ci-data" - self.data_branch = "main" - - def _setup_matplotlib(self): - """Setup matplotlib fonts and styles""" - # Set fonts - rcParams["font.sans-serif"] = ["Arial", "DejaVu Sans", "Liberation Sans"] - rcParams["axes.unicode_minus"] = False # Fix minus sign display issue - - # Set chart styles - plt.style.use("default") - rcParams["figure.figsize"] = (12, 6) - rcParams["font.size"] = 10 - rcParams["axes.grid"] = True - rcParams["grid.alpha"] = 0.3 - - def get_recent_runs( - self, limit: int = 100, start_date: str = None, end_date: str = None - ) -> List[Dict]: - """Get recent CI run data with multiple collection strategies""" - - # If date range is specified, get all data in that range - if start_date or end_date: - return self._get_date_range_runs(start_date, end_date) - - print(f"Getting PR Test runs (limit: {limit})...") - - # Use sampling strategy if limit >= 500, otherwise use sequential - if limit >= 500: - print(f"Using uniform sampling for {limit} runs to cover ~30 days...") - return self._get_sampled_runs(limit) - else: - return self._get_sequential_runs(limit) - - def _get_sequential_runs(self, limit: int) -> List[Dict]: - """Original sequential method for smaller limits""" - print(f"Using sequential sampling for {limit} runs...") - - pr_test_runs = [] - page = 1 - per_page = 100 - - while len(pr_test_runs) < limit: - url = f"{self.base_url}/repos/{self.repo}/actions/runs" - params = {"per_page": per_page, "page": page} - - try: - response = self.session.get(url, params=params) - response.raise_for_status() - data = response.json() - - if not data.get("workflow_runs"): - break - - # Filter PR Test runs - current_pr_tests = [ - run for run in data["workflow_runs"] if run.get("name") == "PR Test" - ] - - # Add to result list, but not exceed limit - for run in current_pr_tests: - if len(pr_test_runs) < limit: - pr_test_runs.append(run) - else: - break - - print(f"Got {len(pr_test_runs)} PR test runs...") - - # Exit if no more data on this page or reached limit - if len(data["workflow_runs"]) < per_page or len(pr_test_runs) >= limit: - break - - page += 1 - time.sleep(0.1) # Avoid API rate limiting - - except requests.exceptions.RequestException as e: - print(f"Error getting CI data: {e}") - break - - return pr_test_runs - - def _get_sampled_runs(self, limit: int) -> List[Dict]: - """Uniform sampling method for 30-day coverage""" - from datetime import datetime, timedelta - - # Uniform sampling across 30 days - sampled_runs = self._sample_time_period(limit, days_back=30, uniform=True) - - print( - f"Sampled {len(sampled_runs)} runs from 30-day period (requested: {limit})" - ) - return sampled_runs - - def _sample_time_period( - self, - target_samples: int, - days_back: int, - skip_recent_days: int = 0, - uniform: bool = False, - ) -> List[Dict]: - """Sample runs from a specific time period""" - from datetime import datetime, timedelta - - # Calculate time range - end_time = datetime.utcnow() - timedelta(days=skip_recent_days) - start_time = end_time - timedelta(days=days_back - skip_recent_days) - - sampling_type = "uniform" if uniform else "systematic" - print( - f" {sampling_type.title()} sampling {target_samples} runs from {start_time.strftime('%Y-%m-%d')} to {end_time.strftime('%Y-%m-%d')}" - ) - - collected_runs = [] - page = 1 - per_page = 100 - total_in_period = 0 - - while True: - url = f"{self.base_url}/repos/{self.repo}/actions/runs" - params = {"per_page": per_page, "page": page} - - try: - response = self.session.get(url, params=params) - response.raise_for_status() - data = response.json() - - if not data.get("workflow_runs"): - break - - period_runs = [] - for run in data["workflow_runs"]: - if run.get("name") != "PR Test": - continue - - created_at = run.get("created_at", "") - if created_at: - try: - run_time = datetime.fromisoformat( - created_at.replace("Z", "+00:00") - ).replace(tzinfo=None) - if start_time <= run_time <= end_time: - period_runs.append(run) - total_in_period += 1 - except: - continue - - collected_runs.extend(period_runs) - - # Progress indicator every 5 pages - if page % 5 == 0: - print( - f" Page {page}: Found {total_in_period} runs in target period, collected {len(collected_runs)} total" - ) - - # Check if we've gone past our time window - if data["workflow_runs"]: - last_run_time_str = data["workflow_runs"][-1].get("created_at", "") - if last_run_time_str: - try: - last_run_time = datetime.fromisoformat( - last_run_time_str.replace("Z", "+00:00") - ).replace(tzinfo=None) - if last_run_time < start_time: - print(f" Reached time boundary at page {page}") - break - except: - pass - - if len(data["workflow_runs"]) < per_page: - break - - page += 1 - time.sleep(0.1) - - except requests.exceptions.RequestException as e: - print(f" Error getting data for time period: {e}") - break - - print( - f" Found {total_in_period} runs in time period, collected {len(collected_runs)} for sampling" - ) - - # Debug: Show time range of collected data - if collected_runs: - collected_runs_sorted = sorted( - collected_runs, key=lambda x: x.get("created_at", "") - ) - earliest = ( - collected_runs_sorted[0].get("created_at", "")[:10] - if collected_runs_sorted - else "N/A" - ) - latest = ( - collected_runs_sorted[-1].get("created_at", "")[:10] - if collected_runs_sorted - else "N/A" - ) - print(f" Collected data spans from {earliest} to {latest}") - - # Sample from collected runs - if len(collected_runs) <= target_samples: - return collected_runs - - if uniform: - # Uniform sampling: sort by time and select evenly distributed samples - collected_runs.sort(key=lambda x: x.get("created_at", "")) - step = len(collected_runs) / target_samples - sampled_runs = [] - - for i in range(target_samples): - index = int(i * step) - if index < len(collected_runs): - sampled_runs.append(collected_runs[index]) - else: - # Systematic sampling for even distribution - step = len(collected_runs) / target_samples - sampled_runs = [] - - for i in range(target_samples): - index = int(i * step) - if index < len(collected_runs): - sampled_runs.append(collected_runs[index]) - - print( - f" Sampled {len(sampled_runs)} runs from {len(collected_runs)} available" - ) - - # Debug: Show time range of sampled data - if sampled_runs: - sampled_runs_sorted = sorted( - sampled_runs, key=lambda x: x.get("created_at", "") - ) - earliest = ( - sampled_runs_sorted[0].get("created_at", "")[:10] - if sampled_runs_sorted - else "N/A" - ) - latest = ( - sampled_runs_sorted[-1].get("created_at", "")[:10] - if sampled_runs_sorted - else "N/A" - ) - print(f" Sampled data spans from {earliest} to {latest}") - - return sampled_runs - - def _get_date_range_runs( - self, start_date: str = None, end_date: str = None - ) -> List[Dict]: - """Get all CI runs within specified date range""" - from datetime import datetime, timedelta - - # Parse dates - if start_date: - try: - start_time = datetime.strptime(start_date, "%Y-%m-%d") - except ValueError: - raise ValueError( - f"Invalid start_date format. Use YYYY-MM-DD, got: {start_date}" - ) - else: - # Default to 30 days ago if no start date - start_time = datetime.utcnow() - timedelta(days=30) - - if end_date: - try: - end_time = datetime.strptime(end_date, "%Y-%m-%d") + timedelta( - days=1 - ) # Include the end date - except ValueError: - raise ValueError( - f"Invalid end_date format. Use YYYY-MM-DD, got: {end_date}" - ) - else: - # Default to now if no end date - end_time = datetime.utcnow() - - # Validate date range - if start_time >= end_time: - raise ValueError( - f"start_date ({start_date}) must be before end_date ({end_date})" - ) - - print( - f"Getting ALL CI runs from {start_time.strftime('%Y-%m-%d')} to {end_time.strftime('%Y-%m-%d')}" - ) - - collected_runs = [] - page = 1 - per_page = 100 - total_in_period = 0 - - while True: - url = f"{self.base_url}/repos/{self.repo}/actions/runs" - params = {"per_page": per_page, "page": page} - - try: - response = self.session.get(url, params=params) - response.raise_for_status() - data = response.json() - - if not data.get("workflow_runs"): - break - - # Filter runs in date range and PR Test runs - period_runs = [] - for run in data["workflow_runs"]: - if run.get("name") != "PR Test": - continue - - created_at = run.get("created_at", "") - if created_at: - try: - run_time = datetime.fromisoformat( - created_at.replace("Z", "+00:00") - ).replace(tzinfo=None) - if start_time <= run_time <= end_time: - period_runs.append(run) - total_in_period += 1 - except: - continue - - collected_runs.extend(period_runs) - - # Progress indicator every 5 pages - if page % 5 == 0: - print( - f" Page {page}: Found {total_in_period} runs in date range, collected {len(collected_runs)} total" - ) - - # Check if we've gone past our time window - if data["workflow_runs"]: - last_run_time_str = data["workflow_runs"][-1].get("created_at", "") - if last_run_time_str: - try: - last_run_time = datetime.fromisoformat( - last_run_time_str.replace("Z", "+00:00") - ).replace(tzinfo=None) - if last_run_time < start_time: - print(f" Reached time boundary at page {page}") - break - except: - pass - - if len(data["workflow_runs"]) < per_page: - break - - page += 1 - time.sleep(0.1) - - except requests.exceptions.RequestException as e: - print(f" Error getting data for date range: {e}") - break - - print( - f"Found {total_in_period} runs in date range {start_time.strftime('%Y-%m-%d')} to {end_time.strftime('%Y-%m-%d')}" - ) - - # Sort by creation time (newest first) - collected_runs.sort(key=lambda x: x.get("created_at", ""), reverse=True) - - return collected_runs - - def get_job_logs(self, run_id: int, job_name: str) -> Optional[str]: - """Get logs for specific job with early exit optimization""" - try: - # First get job list with pagination to ensure we get all jobs - jobs_url = f"{self.base_url}/repos/{self.repo}/actions/runs/{run_id}/jobs" - response = self.session.get(jobs_url, params={"per_page": 100}) - response.raise_for_status() - jobs_data = response.json() - - # Find matching job with early exit - target_job = None - for job in jobs_data.get("jobs", []): - if job_name in job.get("name", ""): - # Early exit if job failed or was skipped - if job.get("conclusion") not in ["success", "neutral"]: - return None - target_job = job - break - - if not target_job: - return None - - # Get logs - 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: - # Reduce verbose error logging for common failures - if "404" not in str(e): - print(f"Failed to get job {job_name} logs: {e}") - return None - - def get_all_job_logs_parallel(self, run_id: int) -> Dict[str, Optional[str]]: - """Get logs for all performance jobs in parallel""" - - def fetch_job_logs(job_name: str) -> tuple[str, Optional[str]]: - """Fetch logs for a single job""" - logs = self.get_job_logs(run_id, job_name) - return job_name, logs - - results = {} - with ThreadPoolExecutor( - max_workers=8 - ) as executor: # Increased concurrent requests - # Submit all job log requests - future_to_job = { - executor.submit(fetch_job_logs, job_name): job_name - for job_name in self.performance_jobs - } - - # Collect results as they complete - for future in as_completed(future_to_job): - job_name, logs = future.result() - results[job_name] = logs - - return results - - def parse_performance_data( - self, log_content: str, job_name: str - ) -> Dict[str, Dict[str, str]]: - """Parse specified performance data from logs""" - if not log_content: - return {} - - test_data = {} - - # Get target tests for current job - target_tests = self.target_tests_and_metrics.get(job_name, {}) - if not target_tests: - return test_data - - # Find all unittest tests using pre-compiled pattern - test_matches = self.test_pattern.findall(log_content) - - for test_match in test_matches: - test_name = test_match.split(".")[-1] # Extract test name - - # Only process target tests - if test_name not in target_tests: - continue - - # Find performance data after this test - test_section = self._extract_test_section(log_content, test_match) - if test_section: - # Only find metrics needed for this test - target_metrics = target_tests[test_name] - perf_data = {} - - for metric_name in target_metrics: - if metric_name in self.compiled_patterns: - compiled_pattern = self.compiled_patterns[metric_name] - matches = compiled_pattern.findall(test_section) - if matches: - perf_data[metric_name] = matches[-1] # Take the last match - - if perf_data: - test_data[test_name] = perf_data - - return test_data - - def _extract_test_section(self, log_content: str, test_pattern: str) -> str: - """Extract log section for specific test""" - lines = log_content.split("\n") - test_start = -1 - test_end = len(lines) - - # Find test start position - for i, line in enumerate(lines): - if test_pattern in line: - test_start = i - break - - if test_start == -1: - return "" - - # Find test end position (next test start or major separator) - for i in range(test_start + 1, len(lines)): - line = lines[i] - if ( - "python3 -m unittest" in line and "test_" in line - ) or "##[group]" in line: - test_end = i - break - - return "\n".join(lines[test_start:test_end]) - - def collect_performance_data(self, runs: List[Dict]) -> Dict[str, List[Dict]]: - """Collect all performance data""" - print("Starting performance data collection...") - - # Create data list for each test - all_test_data = {} - - total_runs = len(runs) - for i, run in enumerate(runs, 1): - if not isinstance(run, dict): - print(f" Warning: run #{i} is not a dict, skipping.") - continue - - run_info = { - "run_number": run.get("run_number"), - "created_at": run.get("created_at"), - "head_sha": (run.get("head_sha") or "")[:8], - "author": "Unknown", - "pr_number": None, - "url": f"https://github.com/{self.repo}/actions/runs/{run.get('id')}", - } - head_commit = run.get("head_commit", {}) - if isinstance(head_commit, dict): - run_info["author"] = head_commit.get("author", {}).get( - "name", "Unknown" - ) - - # Extract PR number - pull_requests = run.get("pull_requests", []) - if pull_requests: - run_info["pr_number"] = pull_requests[0].get("number") - - # Get all job logs in parallel - all_job_logs = self.get_all_job_logs_parallel(run.get("id")) - - # Process each performance test job - for job_name, logs in all_job_logs.items(): - if not logs: - continue - - # Parse performance data - test_results = self.parse_performance_data(logs, job_name) - - for test_name, perf_data in test_results.items(): - # Create full test name including job info - full_test_name = f"{job_name}_{test_name}" - - if full_test_name not in all_test_data: - all_test_data[full_test_name] = [] - - test_entry = {**run_info, **perf_data} - all_test_data[full_test_name].append(test_entry) - print( - f" Found {test_name} performance data: {list(perf_data.keys())}" - ) - - time.sleep(0.2) - return all_test_data - - def generate_performance_tables( - self, test_data: Dict[str, List[Dict]], output_dir: str = "performance_tables" - ): - """Generate performance data tables""" - print(f"Generating performance tables to directory: {output_dir}") - - # Create output directory structure - os.makedirs(output_dir, exist_ok=True) - - # Create subdirectory for each job - job_dirs = {} - for job_name in self.performance_jobs: - job_dir = os.path.join(output_dir, f"{job_name}_summary") - os.makedirs(job_dir, exist_ok=True) - job_dirs[job_name] = job_dir - - # Generate table for each test - for full_test_name, data_list in test_data.items(): - if not data_list: - continue - - # Determine which job this test belongs to - job_name = None - test_name = full_test_name - for job in self.performance_jobs: - if full_test_name.startswith(job): - job_name = job - test_name = full_test_name[len(job) + 1 :] # Remove job prefix - break - - if not job_name: - continue - - job_dir = job_dirs[job_name] - table_file = os.path.join(job_dir, f"{test_name}.csv") - - # Generate CSV table - self._write_csv_table(table_file, test_name, data_list) - - # Generate corresponding chart - print(f" Generating chart for {test_name}...") - self._generate_chart(table_file, test_name, data_list, job_dir) - - print("Performance tables and charts generation completed!") - - def _write_csv_table(self, file_path: str, test_name: str, data_list: List[Dict]): - """Write CSV table""" - if not data_list: - return - - # Get all possible columns - all_columns = set() - for entry in data_list: - all_columns.update(entry.keys()) - - # Define column order - base_columns = ["created_at", "run_number", "pr_number", "author", "head_sha"] - perf_columns = [col for col in all_columns if col not in base_columns + ["url"]] - columns = base_columns + sorted(perf_columns) + ["url"] - - with open(file_path, "w", encoding="utf-8", newline="") as f: - writer = csv.writer(f) - - # Write header - writer.writerow(columns) - - # Write data rows - for entry in sorted( - data_list, key=lambda x: x.get("created_at", ""), reverse=True - ): - row = [] - for col in columns: - value = entry.get(col, "") - if col == "created_at" and value: - # Format time to consistent format - try: - # Handle ISO 8601 format: "2025-09-26T11:16:40Z" - if "T" in value and "Z" in value: - dt = datetime.fromisoformat( - value.replace("Z", "+00:00") - ) - value = dt.strftime("%Y-%m-%d %H:%M") - # If already in desired format, keep it - elif len(value) == 16 and " " in value: - # Validate format - datetime.strptime(value, "%Y-%m-%d %H:%M") - else: - # Try to parse and reformat - dt = datetime.fromisoformat(value) - value = dt.strftime("%Y-%m-%d %H:%M") - except: - # If all parsing fails, keep original value - pass - elif col == "pr_number" and value: - value = f"#{value}" - row.append(str(value)) - writer.writerow(row) - - print(f" Generated table: {file_path} ({len(data_list)} records)") - - def _generate_chart( - self, csv_file_path: str, test_name: str, data_list: List[Dict], output_dir: str - ): - """Generate corresponding time series charts for tables""" - print( - f" Starting chart generation for {test_name} with {len(data_list)} data points" - ) - - if not data_list or len(data_list) < 2: - print( - f" Skipping chart for {test_name}: insufficient data ({len(data_list) if data_list else 0} records)" - ) - return - - try: - # Prepare data - timestamps = [] - metrics_data = {} - - # Get performance metric columns (exclude basic info columns) - base_columns = { - "created_at", - "run_number", - "pr_number", - "author", - "head_sha", - "url", - } - perf_metrics = [] - - for entry in data_list: - for key in entry.keys(): - if key not in base_columns and key not in perf_metrics: - perf_metrics.append(key) - - if not perf_metrics: - print( - f" Skipping chart for {test_name}: no performance metrics found" - ) - return - - print(f" Found performance metrics: {perf_metrics}") - - # Parse data - for entry in data_list: - # Parse time - try: - time_str = entry.get("created_at", "") - if time_str: - # Handle different time formats - timestamp = None - - # Try ISO 8601 format first (from GitHub API): "2025-09-26T11:16:40Z" - if "T" in time_str and "Z" in time_str: - try: - # Parse and convert to naive datetime (remove timezone info) - dt_with_tz = datetime.fromisoformat( - time_str.replace("Z", "+00:00") - ) - timestamp = dt_with_tz.replace(tzinfo=None) - except: - # Fallback for older Python versions - timestamp = datetime.strptime( - time_str, "%Y-%m-%dT%H:%M:%SZ" - ) - - # Try CSV format: "2025-09-26 08:43" - elif " " in time_str and len(time_str) == 16: - timestamp = datetime.strptime(time_str, "%Y-%m-%d %H:%M") - - # Try other common formats - else: - formats_to_try = [ - "%Y-%m-%d %H:%M:%S", - "%Y-%m-%dT%H:%M:%S", - "%Y-%m-%d", - ] - for fmt in formats_to_try: - try: - timestamp = datetime.strptime(time_str, fmt) - break - except: - continue - - if timestamp: - timestamps.append(timestamp) - - # Collect metric data - for metric in perf_metrics: - if metric not in metrics_data: - metrics_data[metric] = [] - - value = entry.get(metric, "") - try: - numeric_value = float(value) - metrics_data[metric].append(numeric_value) - except: - metrics_data[metric].append(None) - else: - print( - f" Failed to parse timestamp format: '{time_str}'" - ) - - except Exception as e: - print(f" Error processing entry: {e}") - continue - - if not timestamps: - print( - f" Skipping chart for {test_name}: no valid timestamps found" - ) - return - - print(f" Parsed {len(timestamps)} timestamps") - - # Sort by time - sorted_data = sorted( - zip(timestamps, *[metrics_data[m] for m in perf_metrics]) - ) - timestamps = [item[0] for item in sorted_data] - for i, metric in enumerate(perf_metrics): - metrics_data[metric] = [item[i + 1] for item in sorted_data] - - # Create chart for each metric - for metric in perf_metrics: - values = metrics_data[metric] - valid_data = [ - (t, v) for t, v in zip(timestamps, values) if v is not None - ] - - if len(valid_data) < 2: - print( - f" Skipping chart for {test_name}_{metric}: insufficient valid data ({len(valid_data)} points)" - ) - continue - - valid_timestamps, valid_values = zip(*valid_data) - - # Create chart - plt.figure(figsize=(12, 6)) - plt.plot( - valid_timestamps, - valid_values, - marker="o", - linewidth=2, - markersize=4, - ) - - # Set title and labels - title = f"{test_name} - {self._format_metric_name(metric)}" - plt.title(title, fontsize=14, fontweight="bold") - plt.xlabel("Time", fontsize=12) - plt.ylabel(self._get_metric_unit(metric), fontsize=12) - - # Format x-axis - plt.gca().xaxis.set_major_formatter(mdates.DateFormatter("%m-%d %H:%M")) - plt.gca().xaxis.set_major_locator( - mdates.HourLocator(interval=max(1, len(valid_timestamps) // 10)) - ) - plt.xticks(rotation=45) - - # Add grid - plt.grid(True, alpha=0.3) - - # Adjust layout - plt.tight_layout() - - # Save chart - chart_filename = f"{test_name}_{metric}.png" - chart_path = os.path.join(output_dir, chart_filename) - plt.savefig(chart_path, dpi=300, bbox_inches="tight") - plt.close() - - print(f" Generated chart: {chart_path}") - - except Exception as e: - print(f" Failed to generate chart for {test_name}: {e}") - import traceback - - traceback.print_exc() - - def _format_metric_name(self, metric: str) -> str: - """Format metric name for display""" - name_mapping = { - "output_throughput_token_s": "Output Throughput", - "median_e2e_latency_ms": "Median E2E Latency", - "median_ttft_ms": "Median TTFT", - "accept_length": "Accept Length", - "input_throughput_token_s": "Input Throughput", - } - return name_mapping.get(metric, metric) - - def _get_metric_unit(self, metric: str) -> str: - """Get metric unit""" - if "throughput" in metric and "token_s" in metric: - return "token/s" - elif "latency" in metric and "ms" in metric: - return "ms" - elif "accept_length" in metric: - return "length" - else: - return "value" - - def generate_summary_report(self, test_data: Dict[str, List[Dict]]): - """Generate summary report""" - print("\n" + "=" * 60) - print("SGLang CI Performance Data Collection Report") - print("=" * 60) - - total_tests = len([test for test, data in test_data.items() if data]) - total_records = sum(len(data) for data in test_data.values()) - - print(f"\nOverall Statistics:") - print(f" Number of tests collected: {total_tests}") - print(f" Total records: {total_records}") - - print(f"\nStatistics by job:") - for job_name in self.performance_jobs: - job_tests = [test for test in test_data.keys() if test.startswith(job_name)] - job_records = sum(len(test_data[test]) for test in job_tests) - print(f" {job_name}: {len(job_tests)} tests, {job_records} records") - - for test in job_tests: - data = test_data[test] - test_short_name = test[len(job_name) + 1 :] - print(f" - {test_short_name}: {len(data)} records") - - print("\n" + "=" * 60) - - def upload_file_to_github( - self, file_path: str, github_path: str, commit_message: str - ) -> bool: - """Upload a file to GitHub repository with retry logic""" - max_retries = 30 - retry_count = 0 - - while retry_count < max_retries: - try: - # Read file content - with open(file_path, "rb") as f: - content = f.read() - - # Encode content to base64 - content_encoded = base64.b64encode(content).decode("utf-8") - - # Check if file exists to get SHA - check_url = ( - f"{self.base_url}/repos/{self.data_repo}/contents/{github_path}" - ) - check_response = self.session.get(check_url) - - sha = None - if check_response.status_code == 200: - sha = check_response.json().get("sha") - - # Prepare upload data - upload_data = { - "message": commit_message, - "content": content_encoded, - "branch": self.data_branch, - } - - if sha: - upload_data["sha"] = sha - - # Upload file - response = self.session.put(check_url, json=upload_data) - - if response.status_code in [200, 201]: - print(f" ✅ Uploaded: {github_path}") - return True - elif response.status_code == 403: - retry_count += 1 - wait_time = min(2**retry_count, 30) - print( - f" ⚠️ Upload forbidden (403) for {github_path}, retrying in {wait_time}s... (attempt {retry_count}/{max_retries})" - ) - if retry_count >= max_retries: - print( - f" ❌ Failed to upload {github_path} after {max_retries} attempts (403 Forbidden)" - ) - return False - time.sleep(wait_time) - else: - response.raise_for_status() - - except requests.exceptions.RequestException as e: - retry_count += 1 - wait_time = min(2**retry_count, 30) - print( - f" ⚠️ Upload error for {github_path} (attempt {retry_count}/{max_retries}): {e}" - ) - if retry_count >= max_retries: - print( - f" ❌ Failed to upload {github_path} after {max_retries} attempts: {e}" - ) - return False - print(f" Retrying in {wait_time}s...") - time.sleep(wait_time) - except Exception as e: - print(f" ❌ Failed to upload {github_path}: {e}") - return False - - return False - - def upload_performance_data_to_github(self, output_dir: str): - """Upload performance_tables to GitHub with original structure""" - print("📤 Uploading performance data to GitHub...") - - # Check if target repository exists with retry logic - repo_url = f"{self.base_url}/repos/{self.data_repo}" - max_retries = 30 - retry_count = 0 - - print(f"🔍 Checking repository access to {self.data_repo}...") - - while retry_count < max_retries: - try: - repo_response = self.session.get(repo_url) - - if repo_response.status_code == 200: - print(f"✅ Repository {self.data_repo} is accessible") - break - elif repo_response.status_code == 404: - print( - f"❌ Repository {self.data_repo} does not exist or is not accessible" - ) - print(" Please ensure:") - print(" 1. The repository exists") - print(" 2. Your GitHub token has access to this repository") - print(" 3. Your token has 'contents:write' permission") - return - elif repo_response.status_code == 403: - retry_count += 1 - wait_time = min(2**retry_count, 60) # Exponential backoff, max 60s - print( - f"⚠️ Repository access forbidden (403), retrying in {wait_time}s... (attempt {retry_count}/{max_retries})" - ) - if retry_count >= max_retries: - print( - f"❌ Failed to access repository after {max_retries} attempts" - ) - print(" This might be due to:") - print(" 1. GitHub API rate limiting") - print(" 2. Token permissions issue") - print(" 3. Repository access restrictions") - return - time.sleep(wait_time) - else: - retry_count += 1 - wait_time = min(2**retry_count, 60) - print( - f"⚠️ Repository access failed with status {repo_response.status_code}, retrying in {wait_time}s... (attempt {retry_count}/{max_retries})" - ) - if retry_count >= max_retries: - print( - f"❌ Failed to access repository {self.data_repo} after {max_retries} attempts" - ) - return - time.sleep(wait_time) - - except Exception as e: - retry_count += 1 - wait_time = min(2**retry_count, 60) - print( - f"⚠️ Error checking repository (attempt {retry_count}/{max_retries}): {e}" - ) - if retry_count >= max_retries: - print( - f"❌ Failed to check repository after {max_retries} attempts: {e}" - ) - return - print(f" Retrying in {wait_time}s...") - time.sleep(wait_time) - - # Generate timestamp for this upload - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - - uploaded_count = 0 - - # Upload all files maintaining original structure - for root, dirs, files in os.walk(output_dir): - for file in files: - local_path = os.path.join(root, file) - - # Keep original directory structure - rel_path = os.path.relpath(local_path, output_dir) - github_path = f"performance_data/{timestamp}/{rel_path}".replace( - "\\", "/" - ) - - # Upload file - commit_msg = f"Add performance data: {rel_path} ({timestamp})" - if self.upload_file_to_github(local_path, github_path, commit_msg): - uploaded_count += 1 - - print(f"📤 Uploaded {uploaded_count} files to GitHub") - - # Print access info - base_url = f"https://github.com/{self.data_repo}/tree/{self.data_branch}/performance_data/{timestamp}" - print(f"🔗 View uploaded data at: {base_url}") - - # Generate GitHub Actions summary - self._generate_github_summary(output_dir, timestamp) - - def _generate_github_summary(self, output_dir: str, timestamp: str): - """Generate GitHub Actions summary with performance data""" - try: - # Check if running in GitHub Actions - 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...") - - # Collect all CSV and PNG files - csv_files = [] - png_files = [] - - for root, dirs, files in os.walk(output_dir): - for file in files: - file_path = os.path.join(root, file) - rel_path = os.path.relpath(file_path, output_dir) - - if file.endswith(".csv"): - csv_files.append((file_path, rel_path)) - elif file.endswith(".png"): - png_files.append((file_path, rel_path)) - - # Sort files by job and test name - csv_files.sort(key=lambda x: x[1]) - png_files.sort(key=lambda x: x[1]) - - # Generate markdown summary - summary_lines = [] - summary_lines.append("# 📊 SGLang Performance Analysis Report") - summary_lines.append("") - summary_lines.append(f"**Analysis Timestamp:** {timestamp}") - summary_lines.append(f"**Total CSV Files:** {len(csv_files)}") - summary_lines.append(f"**Total Chart Files:** {len(png_files)}") - summary_lines.append("") - - # GitHub data repository link - base_url = f"https://github.com/{self.data_repo}/tree/{self.data_branch}/performance_data/{timestamp}" - summary_lines.append(f"🔗 **[View All Data on GitHub]({base_url})**") - summary_lines.append("") - - # Group by job - job_groups = {} - for csv_path, rel_path in csv_files: - # Extract job name from path: job_summary/test_name.csv - parts = rel_path.split("/") - if len(parts) >= 2: - job_name = parts[0].replace("_summary", "") - test_name = parts[1].replace(".csv", "") - - if job_name not in job_groups: - job_groups[job_name] = [] - job_groups[job_name].append((csv_path, test_name, rel_path)) - - # Generate summary for each job - for job_name in sorted(job_groups.keys()): - summary_lines.append(f"## 🚀 {job_name}") - summary_lines.append("") - - tests = job_groups[job_name] - tests.sort(key=lambda x: x[1]) # Sort by test name - - for csv_path, test_name, rel_path in tests: - summary_lines.append(f"### 📈 {test_name}") - - # Add CSV data preview - try: - with open(csv_path, "r", encoding="utf-8") as f: - lines = f.readlines() - if len(lines) > 1: # Has header and data - summary_lines.append("") - summary_lines.append("**Recent Performance Data:**") - summary_lines.append("") - - # Show header - header = lines[0].strip() - summary_lines.append( - f"| {' | '.join(header.split(','))} |" - ) - summary_lines.append( - f"| {' | '.join(['---'] * len(header.split(',')))} |" - ) - - # Show most recent 5 records (CSV is already sorted newest first) - data_lines = lines[1:] - for line in data_lines[ - :5 - ]: # Take first 5 lines (most recent) - if line.strip(): - summary_lines.append( - f"| {' | '.join(line.strip().split(','))} |" - ) - - summary_lines.append("") - except Exception as e: - summary_lines.append(f"*Error reading CSV data: {e}*") - summary_lines.append("") - - # Add chart image if exists - test_prefix = rel_path.replace(".csv", "") - matching_charts = [ - (png_path, png_rel) - for png_path, png_rel in png_files - if png_rel.startswith(test_prefix) - ] - - for png_path, chart_rel_path in matching_charts: - chart_url = f"https://github.com/{self.data_repo}/raw/{self.data_branch}/performance_data/{timestamp}/{chart_rel_path}" - # Extract metric name from filename: test_name_metric_name.png - filename = os.path.basename(chart_rel_path) - metric_name = filename.replace(f"{test_name}_", "").replace( - ".png", "" - ) - summary_lines.append( - f"**{self._format_metric_name(metric_name)} Trend:**" - ) - summary_lines.append("") - summary_lines.append( - f"![{test_name}_{metric_name}]({chart_url})" - ) - summary_lines.append("") - - summary_lines.append("---") - summary_lines.append("") - - # Write summary to GitHub Actions (append mode to preserve CI Analysis report) - with open(github_step_summary, "a", 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}") - import traceback - - traceback.print_exc() - - -def main(): - parser = argparse.ArgumentParser(description="SGLang CI Performance Analyzer") - parser.add_argument("--token", required=True, help="GitHub Personal Access Token") - parser.add_argument( - "--limit", - type=int, - default=100, - help="Number of runs to analyze (default: 100)", - ) - parser.add_argument( - "--output-dir", - default="performance_tables", - help="Output directory (default: performance_tables)", - ) - parser.add_argument( - "--upload-to-github", - action="store_true", - help="Upload results to sglang-bot/sglang-ci-data repository", - ) - parser.add_argument( - "--start-date", - type=str, - help="Start date for date range query (YYYY-MM-DD format). When specified with --end-date, gets ALL runs in range.", - ) - parser.add_argument( - "--end-date", - type=str, - help="End date for date range query (YYYY-MM-DD format). When specified with --start-date, gets ALL runs in range.", - ) - - args = parser.parse_args() - - # Create analyzer - analyzer = SGLangPerfAnalyzer(args.token) - - try: - # Get CI run data - runs = analyzer.get_recent_runs(args.limit, args.start_date, args.end_date) - - if not runs: - print("No CI run data found") - return - - # Collect performance data - test_data = analyzer.collect_performance_data(runs) - - # Generate performance tables - analyzer.generate_performance_tables(test_data, args.output_dir) - - # Upload to GitHub if requested - if args.upload_to_github: - analyzer.upload_performance_data_to_github(args.output_dir) - - # Generate summary report - analyzer.generate_summary_report(test_data) - - except Exception as e: - print(f"Error during analysis: {e}") - import traceback - - traceback.print_exc() - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/scripts/ci_monitor/post_ci_failures_to_slack.py b/scripts/ci_monitor/post_ci_failures_to_slack.py index 60eb0b292..2a56e5b5b 100755 --- a/scripts/ci_monitor/post_ci_failures_to_slack.py +++ b/scripts/ci_monitor/post_ci_failures_to_slack.py @@ -156,7 +156,7 @@ def post_ci_failures_to_slack(report_file: str) -> bool: if not hardware_jobs: summary = "✅ No critical failures detected in scheduled runs" if workflow_url: - summary += f"\n<{workflow_url}|View CI Monitor Run>" + summary += f"\n<{workflow_url}|View CI Failure Monitor run>" color = "good" else: # Ping relevant people when there are failures @@ -177,7 +177,9 @@ def post_ci_failures_to_slack(report_file: str) -> bool: summary_lines.append(f" • {test_type}: {job_list}") if workflow_url: - summary_lines.append(f"\n<{workflow_url}|View Full CI Monitor Report>") + summary_lines.append( + f"\n<{workflow_url}|View full CI Failure Monitor report>" + ) summary = "\n".join(summary_lines) color = "danger" @@ -188,7 +190,7 @@ def post_ci_failures_to_slack(report_file: str) -> bool: attachments=[ { "color": color, - "footer": "SGLang CI Monitor", + "footer": "SGLang CI Failure Monitor", "footer_icon": "https://github.githubassets.com/images/modules/logos_page/GitHub-Mark.png", "ts": int(datetime.now().timestamp()), }