From 8c96fcda703e9737feb9229355193f75674b2b8a Mon Sep 17 00:00:00 2001 From: Douglas Yang Date: Sun, 14 Dec 2025 13:47:15 -0800 Subject: [PATCH] feature: ci failure monitor slack bot (#15110) Co-authored-by: Claude Sonnet 4.5 --- .github/workflows/ci-failure-monitor.yml | 22 +- scripts/ci_monitor/ci_failures_analysis.py | 48 ++-- .../ci_monitor/post_ci_failures_to_slack.py | 230 ++++++++++++++++++ 3 files changed, 278 insertions(+), 22 deletions(-) create mode 100755 scripts/ci_monitor/post_ci_failures_to_slack.py diff --git a/.github/workflows/ci-failure-monitor.yml b/.github/workflows/ci-failure-monitor.yml index 2a016355e..5dc9b4048 100644 --- a/.github/workflows/ci-failure-monitor.yml +++ b/.github/workflows/ci-failure-monitor.yml @@ -29,7 +29,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install requests + pip install requests slack_sdk - name: Run Failure Analysis env: @@ -41,7 +41,6 @@ jobs: python ci_failures_analysis.py \ --token $GITHUB_TOKEN \ --limit 100 \ - --threshold 4 \ --output ci_failure_analysis_$(date +%Y%m%d_%H%M%S).json - name: Upload Analysis Results @@ -51,3 +50,22 @@ jobs: path: | scripts/ci_monitor/ci_failure_analysis_*.json retention-days: 7 + + - name: Send Slack Notification + if: always() + env: + SGLANG_DIFFUSION_SLACK_TOKEN: ${{ secrets.SGLANG_DIFFUSION_SLACK_TOKEN }} + run: | + cd scripts/ci_monitor + LATEST_REPORT=$(ls -t ci_failure_analysis_*.json | head -1) + + if [ ! -f "$LATEST_REPORT" ]; then + echo "No report found, so skipping Slack notification" + exit 0 + fi + + if [ -n "$SGLANG_DIFFUSION_SLACK_TOKEN" ]; then + python3 post_ci_failures_to_slack.py --report-file "$LATEST_REPORT" + else + echo "SGLANG_DIFFUSION_SLACK_TOKEN not configured, skipping notification" + fi diff --git a/scripts/ci_monitor/ci_failures_analysis.py b/scripts/ci_monitor/ci_failures_analysis.py index c3c580408..5e52b4b83 100644 --- a/scripts/ci_monitor/ci_failures_analysis.py +++ b/scripts/ci_monitor/ci_failures_analysis.py @@ -12,7 +12,7 @@ Features: - Generates detailed reports with actionable recommendations Usage: - python ci_failures_analysis.py --token --limit 500 --threshold 3 + python ci_failures_analysis.py --token --limit 100 """ import argparse @@ -30,9 +30,8 @@ import requests class SGLangFailuresAnalyzer: """Analyzes consecutive failures in GitHub Actions workflows.""" - def __init__(self, token: str, alert_threshold: int = 3): + def __init__(self, token: str): self.token = token - self.alert_threshold = alert_threshold self.base_url = "https://api.github.com" self.repo = "sgl-project/sglang" self.headers = { @@ -103,14 +102,32 @@ class SGLangFailuresAnalyzer: return all_runs def get_jobs_for_run(self, run_id: int) -> List[Dict]: - """Get all jobs for a specific workflow run.""" + """Get all jobs for a specific workflow run, handling pagination.""" try: + all_jobs = [] url = f"{self.base_url}/repos/{self.repo}/actions/runs/{run_id}/jobs" - response = self.session.get(url, timeout=30) - response.raise_for_status() - data = response.json() - jobs = data.get("jobs", []) - return jobs + params = {"per_page": 100} # Max per page + + while url: + response = self.session.get(url, params=params, timeout=30) + response.raise_for_status() + data = response.json() + jobs = data.get("jobs", []) + all_jobs.extend(jobs) + + # Check for next page in Link header + link_header = response.headers.get("Link", "") + next_url = None + if link_header: + links = link_header.split(", ") + for link in links: + if 'rel="next"' in link: + next_url = link.split(";")[0].strip("<>") + break + url = next_url + params = {} # Clear params for subsequent requests (URL has them) + + return all_jobs except requests.exceptions.RequestException as e: print(f"Error fetching jobs for run {run_id}: {e}") return [] @@ -1105,7 +1122,6 @@ class SGLangFailuresAnalyzer: 1 for j in sorted_jobs if j[1]["current_streak"] > 0 ), "total_runners": len(runner_stats) if runner_stats else 0, - "alert_threshold": self.alert_threshold, "analysis_timestamp": datetime.now().isoformat(), "avg_queue_time_seconds": overall_avg_queue, "p90_queue_time_seconds": overall_p90_queue, @@ -1169,9 +1185,7 @@ class SGLangFailuresAnalyzer: summary_lines.append( f"**Analysis Timestamp:** {report_data['summary']['analysis_timestamp']}" ) - summary_lines.append( - f"**Alert Threshold:** {report_data['summary']['alert_threshold']} consecutive failures" - ) + summary_lines.append("_Note: Recent runs are shown left to right_") summary_lines.append("") # Summary stats - COLLAPSIBLE @@ -1562,12 +1576,6 @@ def main(): default=100, help="Number of workflow runs to analyze per workflow for general analysis (default: 100)", ) - parser.add_argument( - "--threshold", - type=int, - default=3, - help="Alert threshold for consecutive failures (default: 3)", - ) parser.add_argument( "--output", default=None, @@ -1576,7 +1584,7 @@ def main(): args = parser.parse_args() - analyzer = SGLangFailuresAnalyzer(args.token, alert_threshold=args.threshold) + analyzer = SGLangFailuresAnalyzer(args.token) try: # Fetch runs for each category separately diff --git a/scripts/ci_monitor/post_ci_failures_to_slack.py b/scripts/ci_monitor/post_ci_failures_to_slack.py new file mode 100755 index 000000000..814541dc7 --- /dev/null +++ b/scripts/ci_monitor/post_ci_failures_to_slack.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +""" +Post CI failure analysis results to Slack. + +This is a standalone script that doesn't depend on sglang package installation. +""" + +import argparse +import json +import logging +import os +import sys +from datetime import datetime + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +def post_ci_failures_to_slack(report_file: str) -> bool: + """ + Post CI failure report to Slack with threaded details. + + Creates a parent message with summary (workflow: job1, job2, ...) + and a threaded reply with detailed failure information. + + Args: + report_file: Path to JSON file containing failure analysis from ci_failures_analysis.py + + Returns: + bool: True if successful, False otherwise + """ + try: + from slack_sdk import WebClient + + token = os.environ.get("SGLANG_DIFFUSION_SLACK_TOKEN") + if not token: + logger.info("Slack post failed: no token") + return False + + # CI failures channel + channel_id = "C09HCG2HM1T" + + # Get GitHub run ID for linking to the workflow run + run_id = os.environ.get("GITHUB_RUN_ID", "") + + # Load report data + with open(report_file, "r") as f: + report_data = json.load(f) + + client = WebClient(token=token) + + # Parse the real JSON structure + # The JSON has workflow sections like "pr_test_nvidia_scheduled_data", "nightly_scheduled_data" + # Each section contains jobs with their stats including "current_streak" + + critical_failures = [] + + # Map workflow data keys to display names + workflow_name_map = { + # PR Tests - Scheduled (5 workflows) + "pr_test_nvidia_scheduled_data": "PR Test (Nvidia, scheduled)", + "pr_test_amd_scheduled_data": "PR Test (AMD, scheduled)", + "pr_test_xeon_scheduled_data": "PR Test (Xeon, scheduled)", + "pr_test_xpu_scheduled_data": "PR Test (XPU, scheduled)", + "pr_test_npu_scheduled_data": "PR Test (NPU, scheduled)", + # Nightly Tests - Scheduled (4 workflows) + "nightly_nvidia_scheduled_data": "Nightly Test (Nvidia, scheduled)", + "nightly_amd_scheduled_data": "Nightly Test (AMD, scheduled)", + "nightly_intel_scheduled_data": "Nightly Test (Intel, scheduled)", + "nightly_npu_scheduled_data": "Nightly Test (NPU, scheduled)", + } + + # Iterate through each workflow section + for workflow_key, workflow_data in report_data.items(): + # Skip non-workflow keys (summary, limits, etc.) + if not isinstance(workflow_data, dict) or not any( + isinstance(v, dict) and "current_streak" in v + for v in workflow_data.values() + ): + continue + + # Get workflow display name + workflow_name = workflow_name_map.get(workflow_key, workflow_key) + + # Only process scheduled workflows + if "scheduled" not in workflow_key.lower(): + continue + + # Check each job in this workflow + for job_name, job_data in workflow_data.items(): + if not isinstance(job_data, dict): + continue + + current_streak = job_data.get("current_streak", 0) + + # Filter for jobs with streak >= 2 + if current_streak >= 2: + first_failure = job_data.get("first_failure_in_streak", {}) + last_failure = job_data.get("last_failure_in_streak", {}) + + critical_failures.append( + { + "workflow_name": workflow_name, + "job_name": job_name, + "consecutive_failures": current_streak, + "first_failed_at": ( + first_failure.get("created_at", "unknown") + if first_failure + else "unknown" + ), + "first_failed_url": ( + first_failure.get("job_url", "") + if first_failure + else "" + ), + "last_failed_at": ( + last_failure.get("created_at", "unknown") + if last_failure + else "unknown" + ), + "last_failed_url": ( + last_failure.get("job_url", "") if last_failure else "" + ), + } + ) + + # Group by workflow + workflow_jobs = {} + for job in critical_failures: + workflow = job.get("workflow_name", "Unknown") + job_name = job.get("job_name", "unknown") + if workflow not in workflow_jobs: + workflow_jobs[workflow] = [] + workflow_jobs[workflow].append(job_name) + + # Create summary message + workflow_url = "" + if run_id: + workflow_url = ( + f"https://github.com/sgl-project/sglang/actions/runs/{run_id}" + ) + + if not workflow_jobs: + summary = "✅ No critical failures detected in scheduled runs" + if workflow_url: + summary += f"\n<{workflow_url}|View CI Monitor Run>" + color = "good" + else: + # Ping relevant people when there are failures + mentions = "<@U09RR5TNC94> <@U09ABMCKQPM>" + summary_lines = [f"{mentions} 🚨 *CI Critical Failures (Scheduled Runs)*"] + for workflow, jobs in sorted(workflow_jobs.items()): + job_list = ", ".join(jobs) + summary_lines.append(f"• *{workflow}*: {job_list}") + if workflow_url: + summary_lines.append(f"\n<{workflow_url}|View Full CI Monitor Report>") + summary = "\n".join(summary_lines) + color = "danger" + + # Post parent message + response = client.chat_postMessage( + channel=channel_id, + text=summary, + attachments=[ + { + "color": color, + "footer": "SGLang CI Monitor", + "footer_icon": "https://github.githubassets.com/images/modules/logos_page/GitHub-Mark.png", + "ts": int(datetime.now().timestamp()), + } + ], + ) + + thread_ts = response["ts"] + + # If there are failures, post detailed breakdown in thread + if workflow_jobs: + details_lines = ["*Detailed Failure Breakdown*\n"] + + for job in critical_failures: + workflow = job.get("workflow_name", "Unknown") + job_name = job.get("job_name", "unknown") + consecutive = job.get("consecutive_failures", 0) + first_url = job.get("first_failed_url", "") + first_at = job.get("first_failed_at", "unknown") + last_url = job.get("last_failed_url", "") + last_at = job.get("last_failed_at", "unknown") + + details_lines.append( + f"• *{workflow}* → `{job_name}`\n" + f" Consecutive failures: {consecutive}\n" + f" First failed: <{first_url}|{first_at}>\n" + f" Last failed: <{last_url}|{last_at}>\n" + ) + + details_text = "\n".join(details_lines) + + client.chat_postMessage( + channel=channel_id, + thread_ts=thread_ts, + text=details_text, + ) + + logger.info("CI failure report posted to Slack successfully") + return True + + except Exception as e: + logger.error(f"Failed to post CI failures to Slack: {e}") + return False + + +def main(): + parser = argparse.ArgumentParser( + description="Post CI failure analysis results to Slack" + ) + parser.add_argument( + "--report-file", + type=str, + required=True, + help="Path to CI failure analysis JSON report", + ) + + args = parser.parse_args() + + success = post_ci_failures_to_slack(args.report_file) + sys.exit(0 if success else 1) + + +if __name__ == "__main__": + main()