From 8900f996aa64509851781e2d35a699453992a507 Mon Sep 17 00:00:00 2001 From: Douglas Yang Date: Tue, 18 Nov 2025 22:55:38 -0800 Subject: [PATCH] CI Failure Monitor Improvements (#13558) --- scripts/ci_monitor/ci_failures_analysis.py | 1228 +++++++++++++------- 1 file changed, 820 insertions(+), 408 deletions(-) diff --git a/scripts/ci_monitor/ci_failures_analysis.py b/scripts/ci_monitor/ci_failures_analysis.py index c6844bcb7..ad63a6867 100644 --- a/scripts/ci_monitor/ci_failures_analysis.py +++ b/scripts/ci_monitor/ci_failures_analysis.py @@ -55,6 +55,8 @@ class SGLangFailuresAnalyzer: "check-changes", "pr-test-finish", # Nvidia workflow teardown "pr-test-amd-finish", # AMD workflow teardown + "call-gate", + "pr-gate", ] def get_recent_runs(self, limit: int = 500) -> List[Dict]: @@ -142,8 +144,8 @@ class SGLangFailuresAnalyzer: lambda: defaultdict(int) ) - # Track queue times per runner - runner_queue_times: Dict[str, List[float]] = defaultdict(list) + # Track queue times per runner instance (can aggregate for runner labels if needed) + runner_instance_queue_times: Dict[str, List[float]] = defaultdict(list) # Track individual runner instances (runner_name + runner_id) runner_instance_stats: Dict[str, Dict] = defaultdict( @@ -154,13 +156,21 @@ class SGLangFailuresAnalyzer: runner_current_streak: Dict[str, int] = defaultdict(int) runner_max_streak: Dict[str, int] = defaultdict(int) runner_first_failure_in_streak: Dict[str, Optional[Dict]] = {} + runner_last_failure_in_streak: Dict[str, Optional[Dict]] = {} runner_recovery_info: Dict[str, Optional[Dict]] = {} + runner_error_signatures: Dict[str, Dict[str, int]] = defaultdict( + lambda: defaultdict(int) + ) # Track consecutive failures per runner instance runner_instance_current_streak: Dict[str, int] = defaultdict(int) runner_instance_max_streak: Dict[str, int] = defaultdict(int) runner_instance_first_failure: Dict[str, Optional[Dict]] = {} + runner_instance_last_failure: Dict[str, Optional[Dict]] = {} runner_instance_recovery: Dict[str, Optional[Dict]] = {} + runner_instance_error_signatures: Dict[str, Dict[str, int]] = defaultdict( + lambda: defaultdict(int) + ) total_runs_processed = len(sorted_runs) for i, run in enumerate(sorted_runs, 1): @@ -192,6 +202,9 @@ class SGLangFailuresAnalyzer: runner_had_success: Dict[str, bool] = defaultdict(bool) runner_instance_had_failure: Dict[str, bool] = defaultdict(bool) runner_instance_had_success: Dict[str, bool] = defaultdict(bool) + # Track first failed job for each runner in this run (for linking) + runner_first_failed_job: Dict[str, Dict] = {} + runner_instance_first_failed_job: Dict[str, Dict] = {} for job in jobs: job_name = job.get("name", "") @@ -227,27 +240,6 @@ class SGLangFailuresAnalyzer: runner_total_jobs[runner_key] += 1 runner_job_totals[runner_key][job_name] += 1 - # Calculate queue time (time from created to started) - created_at = job.get("created_at") - started_at = job.get("started_at") - if created_at and started_at: - try: - from datetime import datetime - - created_time = datetime.fromisoformat( - created_at.replace("Z", "+00:00") - ) - started_time = datetime.fromisoformat( - started_at.replace("Z", "+00:00") - ) - queue_time_seconds = ( - started_time - created_time - ).total_seconds() - if queue_time_seconds >= 0: # Sanity check - runner_queue_times[runner_key].append(queue_time_seconds) - except (ValueError, AttributeError): - pass # Skip if timestamp parsing fails - # Track by specific runner instance if runner_id: runner_instance_key = f"{runner_labels_str}_{runner_id}" @@ -257,6 +249,29 @@ class SGLangFailuresAnalyzer: "runner_name" ] = runner_name + # Calculate queue time (time from created to started) per instance + created_at = job.get("created_at") + started_at = job.get("started_at") + if created_at and started_at: + try: + from datetime import datetime + + created_time = datetime.fromisoformat( + created_at.replace("Z", "+00:00") + ) + started_time = datetime.fromisoformat( + started_at.replace("Z", "+00:00") + ) + queue_time_seconds = ( + started_time - created_time + ).total_seconds() + if queue_time_seconds >= 0: # Sanity check + runner_instance_queue_times[runner_instance_key].append( + queue_time_seconds + ) + except (ValueError, AttributeError): + pass # Skip if timestamp parsing fails + conclusion = job.get("conclusion") if conclusion == "failure": @@ -265,6 +280,19 @@ class SGLangFailuresAnalyzer: runner_job_failures[runner_key][job_name] += 1 runner_had_failure[runner_key] = True + # Track first failed job for this runner in this run (for linking) + if runner_key not in runner_first_failed_job: + runner_first_failed_job[runner_key] = { + "job_id": job.get("id"), + "job_url": job.get("html_url", run_info["url"]), + "job_name": job_name, + } + + # Extract error signature for runner + error_signature = self._extract_error_signature(job) + if error_signature: + runner_error_signatures[runner_key][error_signature] += 1 + if runner_id: runner_instance_stats[runner_instance_key]["failed_jobs"] += 1 runner_instance_stats[runner_instance_key]["jobs_failed"][ @@ -272,6 +300,20 @@ class SGLangFailuresAnalyzer: ] += 1 runner_instance_had_failure[runner_instance_key] = True + # Track first failed job for this runner instance in this run + if runner_instance_key not in runner_instance_first_failed_job: + runner_instance_first_failed_job[runner_instance_key] = { + "job_id": job.get("id"), + "job_url": job.get("html_url", run_info["url"]), + "job_name": job_name, + } + + # Extract error signature for runner instance + if error_signature: + runner_instance_error_signatures[runner_instance_key][ + error_signature + ] += 1 + elif conclusion == "success": runner_had_success[runner_key] = True if runner_id: @@ -284,13 +326,20 @@ class SGLangFailuresAnalyzer: ): if runner_had_failure[runner_key]: runner_current_streak[runner_key] += 1 + failure_info = { + **run_info, + "runner_key": runner_key, + } + + # Include job URL if we have it + if runner_key in runner_first_failed_job: + failure_info.update(runner_first_failed_job[runner_key]) # Track if this is the first failure in a new streak if runner_current_streak[runner_key] == 1: - runner_first_failure_in_streak[runner_key] = { - **run_info, - "runner_key": runner_key, - } + runner_first_failure_in_streak[runner_key] = failure_info + # Always update last failure to the most recent one + runner_last_failure_in_streak[runner_key] = failure_info # Update max streak if ( @@ -312,6 +361,7 @@ class SGLangFailuresAnalyzer: runner_current_streak[runner_key] = 0 runner_first_failure_in_streak[runner_key] = None + runner_last_failure_in_streak[runner_key] = None # Update instance streaks for runner_instance_key in set( @@ -322,10 +372,30 @@ class SGLangFailuresAnalyzer: runner_instance_current_streak[runner_instance_key] += 1 if runner_instance_current_streak[runner_instance_key] == 1: - runner_instance_first_failure[runner_instance_key] = { + failure_info = { **run_info, "runner_instance": runner_instance_key, } + # Include job URL if we have it + if runner_instance_key in runner_instance_first_failed_job: + failure_info.update( + runner_instance_first_failed_job[runner_instance_key] + ) + runner_instance_first_failure[runner_instance_key] = ( + failure_info + ) + + # Always update last failure to the most recent one + failure_info = { + **run_info, + "runner_instance": runner_instance_key, + } + # Include job URL if we have it + if runner_instance_key in runner_instance_first_failed_job: + failure_info.update( + runner_instance_first_failed_job[runner_instance_key] + ) + runner_instance_last_failure[runner_instance_key] = failure_info if ( runner_instance_current_streak[runner_instance_key] @@ -347,6 +417,7 @@ class SGLangFailuresAnalyzer: runner_instance_current_streak[runner_instance_key] = 0 runner_instance_first_failure[runner_instance_key] = None + runner_instance_last_failure[runner_instance_key] = None time.sleep(0.05) @@ -357,12 +428,27 @@ class SGLangFailuresAnalyzer: failed = runner_failed_jobs[runner_key] failure_rate = (failed / total * 100) if total > 0 else 0 - # Calculate queue time statistics - queue_times = runner_queue_times[runner_key] - avg_queue_time = sum(queue_times) / len(queue_times) if queue_times else 0 + # Calculate queue time statistics by aggregating from runner instances + # Find all instances that match this runner label + aggregated_queue_times = [] + for instance_key, queue_times in runner_instance_queue_times.items(): + # Extract the labels part from "labels_id" + instance_labels = ( + instance_key.rsplit("_", 1)[0] + if "_" in instance_key + else instance_key + ) + if instance_labels == runner_key: + aggregated_queue_times.extend(queue_times) + + avg_queue_time = ( + sum(aggregated_queue_times) / len(aggregated_queue_times) + if aggregated_queue_times + else 0 + ) p90_queue_time = 0 - if queue_times: - sorted_queue_times = sorted(queue_times) + if aggregated_queue_times: + sorted_queue_times = sorted(aggregated_queue_times) p90_index = int(len(sorted_queue_times) * 0.9) p90_queue_time = ( sorted_queue_times[p90_index] @@ -379,12 +465,25 @@ class SGLangFailuresAnalyzer: "jobs_total": dict(runner_job_totals[runner_key]), "avg_queue_time_seconds": avg_queue_time, "p90_queue_time_seconds": p90_queue_time, - "queue_time_samples": len(queue_times), + "queue_time_samples": len(aggregated_queue_times), } - # Convert runner instance stats to regular dicts + # Convert runner instance stats to regular dicts with queue time stats runner_instance_data = {} for instance_key, stats in runner_instance_stats.items(): + # Calculate queue time statistics for this instance + queue_times = runner_instance_queue_times[instance_key] + avg_queue_time = sum(queue_times) / len(queue_times) if queue_times else 0 + p90_queue_time = 0 + if queue_times: + sorted_queue_times = sorted(queue_times) + p90_index = int(len(sorted_queue_times) * 0.9) + p90_queue_time = ( + sorted_queue_times[p90_index] + if p90_index < len(sorted_queue_times) + else sorted_queue_times[-1] + ) + runner_instance_data[instance_key] = { "total_jobs": stats["total_jobs"], "failed_jobs": stats["failed_jobs"], @@ -395,11 +494,20 @@ class SGLangFailuresAnalyzer: ), "jobs_failed": dict(stats["jobs_failed"]), "runner_name": stats.get("runner_name", "unknown"), + "avg_queue_time_seconds": avg_queue_time, + "p90_queue_time_seconds": p90_queue_time, + "queue_time_samples": len(queue_times), } # Build runner streak data runner_streak_data = {} for runner_key in runner_total_jobs.keys(): + # Get top 3 error signatures for this runner + error_sigs = runner_error_signatures.get(runner_key, {}) + top_errors = sorted(error_sigs.items(), key=lambda x: x[1], reverse=True)[ + :3 + ] + runner_streak_data[runner_key] = { "current_streak": runner_current_streak[runner_key], "max_streak": runner_max_streak[runner_key], @@ -414,12 +522,20 @@ class SGLangFailuresAnalyzer: "first_failure_in_streak": runner_first_failure_in_streak.get( runner_key ), + "last_failure_in_streak": runner_last_failure_in_streak.get(runner_key), "recovery_info": runner_recovery_info.get(runner_key), + "top_error_signatures": top_errors, } # Build runner instance streak data runner_instance_streak_data = {} for instance_key in runner_instance_stats.keys(): + # Get top 3 error signatures for this runner instance + error_sigs = runner_instance_error_signatures.get(instance_key, {}) + top_errors = sorted(error_sigs.items(), key=lambda x: x[1], reverse=True)[ + :3 + ] + runner_instance_streak_data[instance_key] = { "current_streak": runner_instance_current_streak[instance_key], "max_streak": runner_instance_max_streak[instance_key], @@ -439,7 +555,11 @@ class SGLangFailuresAnalyzer: "first_failure_in_streak": runner_instance_first_failure.get( instance_key ), + "last_failure_in_streak": runner_instance_last_failure.get( + instance_key + ), "recovery_info": runner_instance_recovery.get(instance_key), + "top_error_signatures": top_errors, } return ( @@ -449,6 +569,154 @@ class SGLangFailuresAnalyzer: runner_instance_streak_data, ) + def _extract_error_signature(self, job: Dict) -> str: + """ + Extract error signature from a failed job. + + Returns a simplified error type string. + """ + # Check if job has steps with failures + steps = job.get("steps", []) + if not steps: + return "Unknown Error" + + # Look for failed steps + failed_steps = [s for s in steps if s.get("conclusion") == "failure"] + if not failed_steps: + return "Unknown Error" + + # Try to fetch and parse logs for the first failed step + first_failed_step = failed_steps[0] + step_number = first_failed_step.get("number") + + # Attempt to get detailed error from logs + if step_number is not None: + try: + job_id = job.get("id") + # Fetch logs for this specific step + log_url = ( + f"{self.base_url}/repos/{self.repo}/actions/jobs/{job_id}/logs" + ) + response = self.session.get(log_url, timeout=10) + + if response.status_code == 200: + log_text = response.text + + # Check for specific error patterns in logs (case-insensitive) + log_lower = log_text.lower() + + # CUDA/GPU Memory errors (most common for GPU clusters) + if ( + "cuda out of memory" in log_lower + or "cudaerror: out of memory" in log_lower + ): + return "CUDA OOM" + elif "out of memory" in log_lower and ( + "gpu" in log_lower or "device" in log_lower + ): + return "GPU OOM" + elif "out of memory" in log_lower and "cuda" not in log_lower: + return "Out of Memory" + + # CUDA/GPU device errors + if ( + "cuda error: device-side assert" in log_lower + or "device-side assert" in log_lower + ): + return "CUDA Device Assert" + elif ( + "cuda error: an illegal memory access" in log_lower + or "illegal memory access" in log_lower + ): + return "CUDA Illegal Memory Access" + elif "cuda error" in log_lower or "cudaerror" in log_lower: + return "CUDA Error" + elif "gpu" in log_lower and ( + "hang" in log_lower or "hung" in log_lower + ): + return "GPU Hang" + elif ( + "no cuda-capable device" in log_lower + or "cuda device count" in log_lower + and "0" in log_lower + ): + return "No GPU Available" + + # ROCm/AMD GPU errors + if ( + "hipoutofmemoryerror" in log_lower + or "hip out of memory" in log_lower + ): + return "ROCm OOM" + elif "hiperror" in log_lower or "rocm error" in log_lower: + return "ROCm/HIP Error" + + # NCCL/collective communication errors (multi-GPU) + if "nccl error" in log_lower or "ncclerror" in log_lower: + return "NCCL Error" + elif "timeout after" in log_lower and "nccl" in log_lower: + return "NCCL Timeout" + + # Process/system errors + if "killed" in log_lower and ( + "oom" in log_lower or "out of memory" in log_lower + ): + return "Process Killed (OOM)" + elif "killed" in log_lower or "sigkill" in log_lower: + return "Process Killed" + elif "segmentation fault" in log_lower or "sigsegv" in log_lower: + return "Segmentation Fault" + + # Timeout errors + if "timeout" in log_lower or "timed out" in log_lower: + return "Timeout" + + # Connection/network errors + if ( + "connection refused" in log_lower + or "connection reset" in log_lower + ): + return "Connection Error" + elif "ssh" in log_lower and ( + "failed" in log_lower or "error" in log_lower + ): + return "SSH Error" + + # Import/module errors + if "modulenotfounderror" in log_lower or "importerror" in log_lower: + return "Import Error" + + # Assertion errors + if "assertionerror" in log_lower: + return "Assertion Error" + + # Pytest-specific errors + if ( + "pytest" in log_lower + and "error" in log_lower + and "collection" in log_lower + ): + return "Pytest Collection Error" + + except Exception: + # If log fetching fails, fall back to step name analysis + pass + + # Fallback to step name analysis if we couldn't get logs or didn't find specific errors + step_name = first_failed_step.get("name", "Unknown Step") + + # Simplify common patterns based on step name + if "timeout" in step_name.lower(): + return "Timeout" + elif "setup" in step_name.lower() or "install" in step_name.lower(): + return "Setup/Installation Error" + elif "test" in step_name.lower(): + return f"Test Failure: {step_name[:50]}" + elif "build" in step_name.lower(): + return "Build Error" + else: + return f"Step Failed: {step_name[:50]}" + def analyze_consecutive_failures( self, runs: List[Dict] ) -> Tuple[Dict[str, Dict], Dict[str, int]]: @@ -469,7 +737,11 @@ class SGLangFailuresAnalyzer: job_total_failures: Dict[str, int] = defaultdict(int) job_total_runs: Dict[str, int] = defaultdict(int) job_first_failure_in_streak: Dict[str, Optional[Dict]] = {} + job_last_failure_in_streak: Dict[str, Optional[Dict]] = {} job_recovery_info: Dict[str, Optional[Dict]] = {} + job_error_signatures: Dict[str, Dict[str, int]] = defaultdict( + lambda: defaultdict(int) + ) total_runs_processed = len(sorted_runs) for i, run in enumerate(sorted_runs, 1): @@ -518,9 +790,25 @@ class SGLangFailuresAnalyzer: job_first_failure_in_streak[job_name] = { **run_info, "job_name": job_name, + "job_id": job.get("id"), + "job_url": job.get("html_url", run_info["url"]), "conclusion": conclusion, } + # Always update last failure to the most recent one + job_last_failure_in_streak[job_name] = { + **run_info, + "job_name": job_name, + "job_id": job.get("id"), + "job_url": job.get("html_url", run_info["url"]), + "conclusion": conclusion, + } + + # Extract error signature from job + error_signature = self._extract_error_signature(job) + if error_signature: + job_error_signatures[job_name][error_signature] += 1 + # Update max streak if job_current_streak[job_name] > job_max_streak[job_name]: job_max_streak[job_name] = job_current_streak[job_name] @@ -537,12 +825,19 @@ class SGLangFailuresAnalyzer: job_current_streak[job_name] = 0 job_first_failure_in_streak[job_name] = None + job_last_failure_in_streak[job_name] = None time.sleep(0.05) # Build final results job_streak_data = {} for job_name in job_current_streak.keys(): + # Get top 3 error signatures + error_sigs = job_error_signatures.get(job_name, {}) + top_errors = sorted(error_sigs.items(), key=lambda x: x[1], reverse=True)[ + :3 + ] + job_streak_data[job_name] = { "current_streak": job_current_streak[job_name], "max_streak": job_max_streak[job_name], @@ -554,7 +849,9 @@ class SGLangFailuresAnalyzer: else 0 ), "first_failure_in_streak": job_first_failure_in_streak.get(job_name), + "last_failure_in_streak": job_last_failure_in_streak.get(job_name), "recovery_info": job_recovery_info.get(job_name), + "top_error_signatures": top_errors, } return job_streak_data, job_current_streak @@ -588,6 +885,8 @@ class SGLangFailuresAnalyzer: "max_streak": data["max_streak"], "failure_rate": data["failure_rate"], "first_failure": data["first_failure_in_streak"], + "last_failure": data["last_failure_in_streak"], + "top_error_signatures": data.get("top_error_signatures", []), "alert_type": "consecutive_failures", "severity": "high" if current_streak >= 5 else "medium", } @@ -610,6 +909,10 @@ class SGLangFailuresAnalyzer: "total_jobs": streak_data["total_jobs"], "jobs_failed": streak_data.get("jobs_failed", {}), "first_failure": streak_data["first_failure_in_streak"], + "last_failure": streak_data["last_failure_in_streak"], + "top_error_signatures": streak_data.get( + "top_error_signatures", [] + ), "alert_type": "runner_consecutive_failures", "severity": ( "high" @@ -623,6 +926,10 @@ class SGLangFailuresAnalyzer: if runner_instance_streak_data: for instance_key, streak_data in runner_instance_streak_data.items(): if streak_data["current_streak"] >= self.alert_threshold: + # Get queue time info from runner_instance_data + instance_data = runner_instance_data.get(instance_key, {}) + avg_queue = instance_data.get("avg_queue_time_seconds", 0) + runner_alerts.append( { "runner_instance": instance_key, @@ -634,6 +941,11 @@ class SGLangFailuresAnalyzer: "total_jobs": streak_data["total_jobs"], "jobs_failed": streak_data.get("jobs_failed", {}), "first_failure": streak_data["first_failure_in_streak"], + "last_failure": streak_data["last_failure_in_streak"], + "top_error_signatures": streak_data.get( + "top_error_signatures", [] + ), + "avg_queue_time_seconds": avg_queue, "alert_type": "runner_instance_consecutive_failures", "severity": ( "high" @@ -710,23 +1022,23 @@ class SGLangFailuresAnalyzer: reverse=True, ) + # Summary Statistics + print("\n## Summary Statistics") print( - f"\nTotal (unique) jobs analyzed across PR Test workflows: {len(sorted_jobs)}" + f"Total (unique) jobs analyzed across PR Test workflows: {len(sorted_jobs)}" ) print( - f"Jobs with active failure streaks: {sum(1 for j in sorted_jobs if j[1]['current_streak'] > 0)}" + f"Jobs with Active Failure Streaks: {sum(1 for j in sorted_jobs if j[1]['current_streak'] > 0)}" ) - print( - f"Job alerts triggered (>={self.alert_threshold} consecutive failures): {len(job_alerts)}" - ) - + print(f"Job Alerts Triggered: {len(job_alerts)}") if runner_stats: - print(f"Total runners analyzed: {len(runner_stats)}") + print(f"Total Runners Analyzed: {len(runner_stats)}") print( - f"Runner alerts triggered: {len(runner_alerts) if runner_alerts else 0}" + f"Runner Alerts Triggered: {len(runner_alerts) if runner_alerts else 0}" ) - # Calculate overall queue time statistics + # Queue Time Summary + if runner_stats: all_avg_queue_times = [] all_p90_queue_times = [] for stats in runner_stats.values(): @@ -737,79 +1049,213 @@ class SGLangFailuresAnalyzer: if all_avg_queue_times: overall_avg = sum(all_avg_queue_times) / len(all_avg_queue_times) overall_p90 = sum(all_p90_queue_times) / len(all_p90_queue_times) - print(f"\n--- Queue Time Summary ---") + print("\n## Queue Time Summary") print( - f"Average queue time across all runners: {overall_avg / 60:.1f} minutes ({overall_avg:.0f}s)" + f"Average Queue Time (across all runners): {overall_avg / 60:.1f} minutes ({overall_avg:.0f}s)" ) print( - f"P90 queue time across all runners: {overall_p90 / 60:.1f} minutes ({overall_p90:.0f}s)" + f"P90 Queue Time (across all runners): {overall_p90 / 60:.1f} minutes ({overall_p90:.0f}s)" ) - # Section 1: Currently Broken Jobs (Consecutive Failures) - URGENT - print("\n" + "=" * 100) - print("SECTION 1: Currently Broken Jobs (Active Consecutive Failures)") - print("=" * 100) + # ALERTS: Critical Consecutive Job Failures (streak >= 2) + if job_alerts: + # Filter alerts with streak >= 2 + filtered_job_alerts = [a for a in job_alerts if a["current_streak"] >= 2] + if filtered_job_alerts: + print("\n" + "=" * 150) + print("## ALERTS: Critical Consecutive Job Failures") + print("=" * 150) + print( + f"\n{'Job Name':<40} {'Streak':<8} {'Max':<6} {'First Failure':<16} {'Last Failure':<16} {'Top Errors':<60}" + ) + print("-" * 150) + + for alert in sorted( + filtered_job_alerts, key=lambda x: x["current_streak"], reverse=True + ): + job_name = alert["job_name"] + display_name = ( + job_name if len(job_name) <= 38 else job_name[:35] + "..." + ) + + first_failure = alert.get("first_failure") + first_failure_str = ( + f"Run #{first_failure['run_number']}" + if first_failure + else "N/A" + ) + + last_failure = alert.get("last_failure") + last_failure_str = ( + f"Run #{last_failure['run_number']}" if last_failure else "N/A" + ) + + # Format top errors - don't truncate + top_errors = alert.get("top_error_signatures", []) + if top_errors: + error_display = ", ".join( + [f"{err[0]} ({err[1]})" for err in top_errors] + ) + else: + error_display = "N/A" + + print( + f"{display_name:<40} {alert['current_streak']:<8} {alert['max_streak']:<6} {first_failure_str:<16} {last_failure_str:<16} {error_display:<60}" + ) + else: + print("\n" + "=" * 100) + print("## ALERTS: Critical Consecutive Job Failures") + print("=" * 100) + print( + "\nNothing to display (no jobs with consecutive failure streak >= 2)" + ) + + # ALERTS: Runners with Issues (streak >= 2) + if runner_alerts: + # Only show consecutive failure alerts with streak >= 2, and only machine instances + instance_alerts = [ + a + for a in runner_alerts + if a["alert_type"] == "runner_instance_consecutive_failures" + and a.get("current_streak", 0) >= 2 + ] + + if instance_alerts: + print("\n" + "=" * 170) + print("## ALERTS: Runners with Issues") + print("=" * 170) + print("\n### Runner Consecutive Failures") + print( + f"\n{'Runner':<30} {'Str':<5} {'Max':<5} {'Fail%':<7} {'AvgQ':<7} {'First':<13} {'Last':<13} {'Top Errors':<45} {'Jobs Failed':<40}" + ) + print("-" * 170) + + for alert in sorted( + instance_alerts, + key=lambda x: x.get("current_streak", 0), + reverse=True, + ): + # Use the actual machine name instead of labels or instance key + runner_name = alert.get("runner_name", "unknown") + display_name = ( + runner_name + if len(runner_name) <= 28 + else runner_name[:25] + "..." + ) + + # Get all failed jobs - don't truncate + jobs_failed = alert.get("jobs_failed", {}) + top_jobs = sorted( + jobs_failed.items(), key=lambda x: x[1], reverse=True + ) + jobs_display = ( + ", ".join([f"{job} ({count})" for job, count in top_jobs]) + if top_jobs + else "N/A" + ) + + # Format queue time + avg_queue = alert.get("avg_queue_time_seconds", 0) + avg_queue_str = f"{avg_queue / 60:.1f}m" if avg_queue > 0 else "N/A" + + first_failure = alert.get("first_failure") + first_failure_str = ( + f"Run #{first_failure['run_number']}" + if first_failure + else "N/A" + ) + + last_failure = alert.get("last_failure") + last_failure_str = ( + f"Run #{last_failure['run_number']}" if last_failure else "N/A" + ) + + # Format top errors - don't truncate + top_errors = alert.get("top_error_signatures", []) + if top_errors: + error_display = ", ".join( + [f"{err[0]} ({err[1]})" for err in top_errors] + ) + else: + error_display = "N/A" + + print( + f"{display_name:<30} {alert['current_streak']:<5} {alert['max_streak']:<5} {alert['failure_rate']:>5.1f}% {avg_queue_str:<7} {first_failure_str:<13} {last_failure_str:<13} {error_display:<45} {jobs_display:<40}" + ) + else: + print("\n" + "=" * 100) + print("## ALERTS: Runners with Issues") + print("=" * 100) + print( + "\nNothing to display (no runners with consecutive failure streak > 2)" + ) + + # Section 1: Currently Broken Jobs (streak >= 2) broken_jobs = [ - (name, data) for name, data in sorted_jobs if data["current_streak"] > 0 + (name, data) for name, data in sorted_jobs if data["current_streak"] >= 2 ] if broken_jobs: + print("\n" + "=" * 140) + print("## Section 1: Top 15 Consecutively Failing Jobs") + print("=" * 140) print( - f"\n{'Rank':<4} {'Job Name':<50} {'Current Streak':<16} {'Max Streak':<12}" + f"\n{'Job Name':<40} {'Streak':<8} {'Max':<6} {'First':<13} {'Last':<13} {'Top Errors':<50}" ) - print("-" * 100) - for i, (job_name, data) in enumerate(broken_jobs[:20], 1): - print( - f"{i:<4} {job_name:<50} {data['current_streak']:<16} {data['max_streak']:<12}" + print("-" * 140) + for job_name, data in broken_jobs[:20]: + display_name = ( + job_name if len(job_name) <= 38 else job_name[:35] + "..." ) - else: - print("\n✓ No jobs are currently in a failure streak!") - # Print job alerts - if job_alerts: - print("\n" + "!" * 40) - print("ALERTS: Jobs with Consecutive Failures Exceeding Threshold") - print("!" * 40) - - for alert in sorted( - job_alerts, key=lambda x: x["current_streak"], reverse=True - ): - print(f"\n {alert['job_name']}") - print( - f" Current Streak: {alert['current_streak']} consecutive failures" + # Get first and last failure info + first_failure = data.get("first_failure_in_streak") + first_failure_str = ( + f"Run #{first_failure['run_number']}" if first_failure else "N/A" ) - print(f" Max Streak: {alert['max_streak']}") - print(f" Severity: {alert['severity'].upper()}") - if alert["first_failure"]: - first = alert["first_failure"] - print( - f" First Failure in Streak: Run #{first['run_number']} ({first['created_at']})" + last_failure = data.get("last_failure_in_streak") + last_failure_str = ( + f"Run #{last_failure['run_number']}" if last_failure else "N/A" + ) + + # Format top errors - don't truncate + top_errors = data.get("top_error_signatures", []) + if top_errors: + error_display = ", ".join( + [f"{err[0]} ({err[1]})" for err in top_errors] ) - print(f" Link: {first['url']}") + else: + error_display = "N/A" - # Section 2: Runner Health Analysis - if runner_stats and runner_streak_data: - print("\n" + "=" * 100) - print("SECTION 2: Runner Health Analysis") - print("=" * 100) + print( + f"{display_name:<40} {data['current_streak']:<8} {data['max_streak']:<6} {first_failure_str:<13} {last_failure_str:<13} {error_display:<50}" + ) - # Combine stats with streak data and sort by consecutive failures first + # Section 2: Runner Health Analysis - Use machine names from runner instances (streak >= 2) + if runner_instance_data and runner_instance_streak_data: + # Combine instance stats with streak data and sort by consecutive failures first combined_data = [] - for runner_labels, stats in runner_stats.items(): - streak_data = runner_streak_data.get(runner_labels, {}) + for instance_key, stats in runner_instance_data.items(): + streak_data = runner_instance_streak_data.get(instance_key, {}) combined_data.append( { - "runner_labels": runner_labels, + "runner_name": stats.get("runner_name", "unknown"), + "instance_key": instance_key, "current_streak": streak_data.get("current_streak", 0), "max_streak": streak_data.get("max_streak", 0), "failure_rate": stats["failure_rate"], "total_jobs": stats["total_jobs"], - "unique_jobs": stats["unique_jobs_with_failures"], - "avg_queue": stats["avg_queue_time_seconds"], - "p90_queue": stats["p90_queue_time_seconds"], - "queue_samples": stats["queue_time_samples"], + "unique_jobs": len(stats.get("jobs_failed", {})), + "avg_queue": stats.get("avg_queue_time_seconds", 0), + "p90_queue": stats.get("p90_queue_time_seconds", 0), + "queue_samples": stats.get("queue_time_samples", 0), + "first_failure": streak_data.get("first_failure_in_streak"), + "last_failure": streak_data.get("last_failure_in_streak"), + "top_error_signatures": streak_data.get( + "top_error_signatures", [] + ), } ) @@ -820,151 +1266,64 @@ class SGLangFailuresAnalyzer: reverse=True, ) - print(f"\nTop 15 Runners by Consecutive Failures:") - print( - " (High failure + Low unique jobs = Same job failing repeatedly → Likely job/test issue)" - ) - print( - " (High failure + High unique jobs = Many different jobs failing → Likely runner/infrastructure issue)" - ) - print("-" * 160) - print( - f"{'Rank':<4} {'Runner Labels':<35} {'Streak':<8} {'Max':<6} {'Fail Rate':<10} {'Total':<7} {'Unique Jobs':<13} {'Avg Queue':<11} {'P90 Queue':<11}" - ) - print("-" * 160) - - for i, runner_data in enumerate(sorted_runners[:15], 1): - # Truncate labels if too long for display - display_labels = ( - runner_data["runner_labels"] - if len(runner_data["runner_labels"]) <= 33 - else runner_data["runner_labels"][:30] + "..." - ) - - # Format streak - current_streak = runner_data["current_streak"] - streak_str = f"{current_streak}" if current_streak > 0 else "-" - - # Format max streak - max_streak = runner_data["max_streak"] - max_str = f"{max_streak}" if max_streak > 0 else "-" - - # Format queue times - avg_queue_str = ( - f"{runner_data['avg_queue'] / 60:.1f}m" - if runner_data["queue_samples"] > 0 - else "N/A" - ) - p90_queue_str = ( - f"{runner_data['p90_queue'] / 60:.1f}m" - if runner_data["queue_samples"] > 0 - else "N/A" - ) - - print( - f"{i:<4} {display_labels:<35} {streak_str:<8} {max_str:<6} {runner_data['failure_rate']:>8.1f}% " - f"{runner_data['total_jobs']:<7} {runner_data['unique_jobs']:<13} " - f"{avg_queue_str:<11} {p90_queue_str:<11}" - ) - - # Print runner alerts - if runner_alerts: - print("\n" + "!" * 40) - print("ALERTS: Runners with Issues") - print("!" * 40) - - # Only show consecutive failure alerts - consecutive_alerts = [ - a - for a in runner_alerts - if a["alert_type"] - in [ - "runner_consecutive_failures", - "runner_instance_consecutive_failures", - ] + # Only show runners with streak >= 2 + runners_with_issues = [ + r for r in sorted_runners if r["current_streak"] >= 2 ] - if consecutive_alerts: - print("\n--- CONSECUTIVE FAILURE ALERTS ---") + if runners_with_issues: + print("\n" + "=" * 160) + print("## Section 2: Top 15 Workers by Consecutive Failures") + print("=" * 160) print( - "(Runners that have failed in multiple consecutive workflow runs)" + f"\n{'Machine Name':<30} {'Str':<5} {'Max':<5} {'Fail%':<7} {'AvgQ':<7} {'First':<13} {'Last':<13} {'Top Errors':<45} {'Total Jobs':<11} {'Unique Jobs':<12}" ) - print() + print("-" * 160) - for alert in sorted( - consecutive_alerts, - key=lambda x: (x.get("current_streak", 0), x.get("failure_rate", 0)), - reverse=True, - ): - if alert["alert_type"] == "runner_consecutive_failures": - print(f"\n Runner Labels: {alert['runner_labels']}") - print( - f" Current Streak: {alert['current_streak']} consecutive runs with failures" - ) - print(f" Max Streak: {alert['max_streak']}") - print(f" Failure Rate: {alert['failure_rate']:.1f}%") - print( - f" Total Failures: {alert['total_failures']} / {alert['total_jobs']}" + for runner_data in runners_with_issues[:15]: + # Truncate machine name if too long for display + display_name = ( + runner_data["runner_name"] + if len(runner_data["runner_name"]) <= 28 + else runner_data["runner_name"][:25] + "..." ) - # Show jobs that failed on this runner type - jobs_failed = alert.get("jobs_failed", {}) - if jobs_failed: - print(f" Jobs That Failed:") - for job_name, count in sorted( - jobs_failed.items(), key=lambda x: x[1], reverse=True - ): - print(f" - {job_name}: {count} failure(s)") + # Format streaks + streak_str = str(runner_data["current_streak"]) + max_str = str(runner_data["max_streak"]) - print(f" Severity: {alert['severity'].upper()}") - if alert.get("first_failure"): - first = alert["first_failure"] - print( - f" First Failure in Streak: Run #{first['run_number']} ({first['created_at']})" + # Format queue time + avg_queue_str = ( + f"{runner_data['avg_queue'] / 60:.1f}m" + if runner_data["queue_samples"] > 0 + else "N/A" + ) + + # Get first and last failure info + first_failure = runner_data.get("first_failure") + first_failure_str = ( + f"Run #{first_failure['run_number']}" + if first_failure + else "N/A" + ) + + last_failure = runner_data.get("last_failure") + last_failure_str = ( + f"Run #{last_failure['run_number']}" if last_failure else "N/A" + ) + + # Format top errors - don't truncate + top_errors = runner_data.get("top_error_signatures", []) + if top_errors: + error_display = ", ".join( + [f"{err[0]} ({err[1]})" for err in top_errors] ) - print(f" Link: {first['url']}") - elif alert["alert_type"] == "runner_instance_consecutive_failures": - # Extract runner labels from instance key (format: "labels_id") - instance_key = alert["runner_instance"] - runner_labels = ( - instance_key.rsplit("_", 1)[0] - if "_" in instance_key - else instance_key - ) - runner_id = ( - instance_key.rsplit("_", 1)[1] - if "_" in instance_key - else "unknown" - ) + else: + error_display = "N/A" - print(f"\n Runner Type: {runner_labels}") - print(f" Specific Instance ID: {runner_id}") - print(f" Machine Name: {alert['runner_name']}") print( - f" Current Streak: {alert['current_streak']} consecutive runs with failures" + f"{display_name:<30} {streak_str:<5} {max_str:<5} {runner_data['failure_rate']:>5.1f}% {avg_queue_str:<7} {first_failure_str:<13} {last_failure_str:<13} {error_display:<45} {runner_data['total_jobs']:<11} {runner_data['unique_jobs']:<12}" ) - print(f" Max Streak: {alert['max_streak']}") - print(f" Failure Rate: {alert['failure_rate']:.1f}%") - print( - f" Total Failures: {alert['total_failures']} / {alert['total_jobs']}" - ) - - # Show jobs that failed on this runner instance - jobs_failed = alert.get("jobs_failed", {}) - if jobs_failed: - print(f" Jobs That Failed:") - for job_name, count in sorted( - jobs_failed.items(), key=lambda x: x[1], reverse=True - ): - print(f" - {job_name}: {count} failure(s)") - - print(f" Severity: {alert['severity'].upper()}") - if alert.get("first_failure"): - first = alert["first_failure"] - print( - f" First Failure in Streak: Run #{first['run_number']} ({first['created_at']})" - ) - print(f" Link: {first['url']}") # Build report data (always needed for GitHub summary) # Calculate overall queue time for summary @@ -1089,202 +1448,238 @@ class SGLangFailuresAnalyzer: ) summary_lines.append("") - # Job Alerts section + # Job Alerts section (streak >= 2) if report_data.get("job_alerts"): - summary_lines.append("## ALERTS: Critical Consecutive Job Failures") - summary_lines.append("") - summary_lines.append( - "| Job Name | Current Streak | Max Streak | First Failure | Link |" - ) - summary_lines.append( - "|----------|----------------|------------|---------------|------|" - ) - - for alert in sorted( - report_data["job_alerts"], - key=lambda x: x["current_streak"], - reverse=True, - ): - job_name = alert["job_name"] - if len(job_name) > 40: - job_name = job_name[:37] + "..." - - first_failure = alert.get("first_failure") - first_failure_str = ( - f"Run #{first_failure['run_number']}" - if first_failure - else "N/A" - ) - first_failure_link = first_failure["url"] if first_failure else "" - - summary_lines.append( - f"| `{job_name}` | {alert['current_streak']} | {alert['max_streak']} | " - f"{first_failure_str} | [View]({first_failure_link}) |" - ) - - summary_lines.append("") - - # Runner Alerts section - if report_data.get("runner_alerts"): - summary_lines.append("## ALERTS: Runners with Issues") - summary_lines.append("") - - # Only show consecutive failure alerts - consecutive_alerts = [ - a - for a in report_data["runner_alerts"] - if a["alert_type"] - in [ - "runner_consecutive_failures", - "runner_instance_consecutive_failures", - ] + # Filter alerts with streak >= 2 + filtered_job_alerts = [ + a for a in report_data["job_alerts"] if a["current_streak"] >= 2 ] - if consecutive_alerts: - summary_lines.append("### Runner Consecutive Failures") + if filtered_job_alerts: + summary_lines.append("## ALERTS: Critical Consecutive Job Failures") summary_lines.append("") summary_lines.append( - "| Runner | Current Streak | Max Streak | Failure Rate | Jobs Failed | First Failure | Link |" + "| Job Name | Streak | Max | First Failure | Last Failure | Top Errors |" ) summary_lines.append( - "|--------|----------------|------------|--------------|-------------|---------------|------|" + "|----------|--------|-----|---------------|--------------|------------|" ) for alert in sorted( - consecutive_alerts, + filtered_job_alerts, + key=lambda x: x["current_streak"], + reverse=True, + ): + job_name = alert["job_name"] + if len(job_name) > 35: + job_name = job_name[:32] + "..." + + first_failure = alert.get("first_failure") + if first_failure: + first_failure_str = f"[Run #{first_failure['run_number']}]({first_failure.get('job_url', first_failure['url'])})" + else: + first_failure_str = "N/A" + + last_failure = alert.get("last_failure") + if last_failure: + last_failure_str = f"[Run #{last_failure['run_number']}]({last_failure.get('job_url', last_failure['url'])})" + else: + last_failure_str = "N/A" + + # Format top errors as bullet list + top_errors = alert.get("top_error_signatures", []) + if top_errors: + error_str = "
".join( + [f"• {err[0]} ({err[1]})" for err in top_errors] + ) + else: + error_str = "N/A" + + summary_lines.append( + f"| `{job_name}` | {alert['current_streak']} | {alert['max_streak']} | " + f"{first_failure_str} | {last_failure_str} | {error_str} |" + ) + + summary_lines.append("") + else: + summary_lines.append("## ALERTS: Critical Consecutive Job Failures") + summary_lines.append("") + summary_lines.append( + "Nothing to display (no jobs with consecutive failure streak >= 2)" + ) + summary_lines.append("") + + # Runner Alerts section (streak >= 2) + if report_data.get("runner_alerts"): + # Only show consecutive failure alerts with streak >= 2, and only machine instances + instance_alerts = [ + a + for a in report_data["runner_alerts"] + if a["alert_type"] == "runner_instance_consecutive_failures" + and a.get("current_streak", 0) >= 2 + ] + + if instance_alerts: + summary_lines.append("## ALERTS: Workers with Issues") + summary_lines.append("") + summary_lines.append( + "| Runner | Streak | Max | Fail Rate | Avg Queue | First Failure | Last Failure | Top Errors | Jobs Failed |" + ) + summary_lines.append( + "|--------|--------|-----|-----------|-----------|---------------|--------------|------------|-------------|" + ) + + for alert in sorted( + instance_alerts, key=lambda x: x.get("current_streak", 0), reverse=True, ): - if alert["alert_type"] == "runner_consecutive_failures": - runner_labels = alert["runner_labels"] - if len(runner_labels) > 35: - runner_labels = runner_labels[:32] + "..." + # Use the actual machine name instead of labels or instance key + runner_name = alert.get("runner_name", "unknown") + if len(runner_name) > 28: + runner_name = runner_name[:25] + "..." - # Get top 3 failed jobs - jobs_failed = alert.get("jobs_failed", {}) - top_jobs = sorted( - jobs_failed.items(), key=lambda x: x[1], reverse=True - )[:3] - jobs_str = ( - ", ".join( - [f"{job} ({count})" for job, count in top_jobs] - ) - if top_jobs - else "N/A" + # Get all failed jobs as bullet list + jobs_failed = alert.get("jobs_failed", {}) + top_jobs = sorted( + jobs_failed.items(), key=lambda x: x[1], reverse=True + ) + jobs_str = ( + "
".join( + [f"• {job} ({count})" for job, count in top_jobs] ) + if top_jobs + else "N/A" + ) - first_failure = alert.get("first_failure") - first_failure_str = ( - f"Run #{first_failure['run_number']}" - if first_failure - else "N/A" - ) - first_failure_link = ( - first_failure["url"] if first_failure else "" - ) + # Format queue time + avg_queue = alert.get("avg_queue_time_seconds", 0) + avg_queue_str = ( + f"{avg_queue / 60:.1f}m" if avg_queue > 0 else "N/A" + ) - summary_lines.append( - f"| `{runner_labels}` | {alert['current_streak']} | {alert['max_streak']} | " - f"{alert['failure_rate']:.1f}% | {jobs_str} | {first_failure_str} | [View]({first_failure_link}) |" - ) - elif ( - alert["alert_type"] - == "runner_instance_consecutive_failures" - ): - instance = alert["runner_instance"] - if len(instance) > 35: - instance = instance[:32] + "..." + first_failure = alert.get("first_failure") + if first_failure: + first_failure_str = f"[Run #{first_failure['run_number']}]({first_failure.get('job_url', first_failure['url'])})" + else: + first_failure_str = "N/A" - # Get top 3 failed jobs - jobs_failed = alert.get("jobs_failed", {}) - top_jobs = sorted( - jobs_failed.items(), key=lambda x: x[1], reverse=True - )[:3] - jobs_str = ( - ", ".join( - [f"{job} ({count})" for job, count in top_jobs] - ) - if top_jobs - else "N/A" - ) + last_failure = alert.get("last_failure") + if last_failure: + last_failure_str = f"[Run #{last_failure['run_number']}]({last_failure.get('job_url', last_failure['url'])})" + else: + last_failure_str = "N/A" - first_failure = alert.get("first_failure") - first_failure_str = ( - f"Run #{first_failure['run_number']}" - if first_failure - else "N/A" - ) - first_failure_link = ( - first_failure["url"] if first_failure else "" + # Format top errors as bullet list + top_errors = alert.get("top_error_signatures", []) + if top_errors: + error_str = "
".join( + [f"• {err[0]} ({err[1]})" for err in top_errors] ) + else: + error_str = "N/A" - summary_lines.append( - f"| `{instance}` (instance) | {alert['current_streak']} | {alert['max_streak']} | " - f"{alert['failure_rate']:.1f}% | {jobs_str} | {first_failure_str} | [View]({first_failure_link}) |" - ) + summary_lines.append( + f"| `{runner_name}` | {alert['current_streak']} | {alert['max_streak']} | " + f"{alert['failure_rate']:.1f}% | {avg_queue_str} | {first_failure_str} | {last_failure_str} | " + f"{error_str} | {jobs_str} |" + ) summary_lines.append("") summary_lines.append("") + else: + summary_lines.append("## ALERTS: Runners with Issues") + summary_lines.append("") + summary_lines.append( + "Nothing to display (no runners with consecutive failure streak > 2)" + ) + summary_lines.append("") + summary_lines.append("") - # Section 1: Currently Broken Jobs - summary_lines.append( - "## Section 1: Currently Broken Jobs (Active Failures)" - ) - summary_lines.append("") - + # Section 1: Currently Broken Jobs - Only show if there are broken jobs sorted_jobs = sorted( report_data["job_streak_data"].items(), key=lambda x: (x[1]["current_streak"], x[1]["failure_rate"]), reverse=True, ) + # Only show jobs with streak >= 2 broken_jobs = [ - (name, data) for name, data in sorted_jobs if data["current_streak"] > 0 + (name, data) + for name, data in sorted_jobs + if data["current_streak"] >= 2 ] if broken_jobs: + summary_lines.append("## Section 1: Top 15 Consecutively Failing Jobs") + summary_lines.append("") summary_lines.append( - "| Rank | Job Name | Current Streak | Max Streak |" + "| Job Name | Streak | Max | First Failure | Last Failure | Top Errors |" ) summary_lines.append( - "|------|----------|----------------|------------|" + "|----------|--------|-----|---------------|--------------|------------|" ) - for i, (job_name, data) in enumerate(broken_jobs[:20], 1): + for job_name, data in broken_jobs[:20]: display_name = ( - job_name if len(job_name) <= 40 else job_name[:37] + "..." + job_name if len(job_name) <= 35 else job_name[:32] + "..." ) + + # Get first and last failure info + first_failure = data.get("first_failure_in_streak") + if first_failure: + first_failure_str = f"[Run #{first_failure['run_number']}]({first_failure.get('job_url', first_failure['url'])})" + else: + first_failure_str = "N/A" + + last_failure = data.get("last_failure_in_streak") + if last_failure: + last_failure_str = f"[Run #{last_failure['run_number']}]({last_failure.get('job_url', last_failure['url'])})" + else: + last_failure_str = "N/A" + + # Format top errors as bullet list + top_errors = data.get("top_error_signatures", []) + if top_errors: + error_str = "
".join( + [f"• {err[0]} ({err[1]})" for err in top_errors] + ) + else: + error_str = "N/A" + summary_lines.append( - f"| {i} | `{display_name}` | {data['current_streak']} | {data['max_streak']} |" + f"| `{display_name}` | {data['current_streak']} | {data['max_streak']} | " + f"{first_failure_str} | {last_failure_str} | {error_str} |" ) - else: - summary_lines.append("No jobs are currently in a failure streak!") - summary_lines.append("") - - # Section 2: Runner Health Analysis - if report_data.get("runner_stats") and report_data.get( - "runner_streak_data" - ): - summary_lines.append("## Section 2: Runner Health Analysis") summary_lines.append("") - # Combine stats with streak data and sort by consecutive failures first + # Section 2: Runner Health Analysis - Use machine names from runner instances + if report_data.get("runner_instance_data") and report_data.get( + "runner_instance_streak_data" + ): + # Combine instance stats with streak data and sort by consecutive failures first combined_data = [] - for runner_labels, stats in report_data["runner_stats"].items(): - streak_data = report_data["runner_streak_data"].get( - runner_labels, {} + for instance_key, stats in report_data["runner_instance_data"].items(): + streak_data = report_data["runner_instance_streak_data"].get( + instance_key, {} ) combined_data.append( { - "runner_labels": runner_labels, + "runner_name": stats.get("runner_name", "unknown"), + "instance_key": instance_key, "current_streak": streak_data.get("current_streak", 0), "max_streak": streak_data.get("max_streak", 0), "failure_rate": stats["failure_rate"], "total_jobs": stats["total_jobs"], - "unique_jobs": stats["unique_jobs_with_failures"], - "avg_queue": stats["avg_queue_time_seconds"], - "p90_queue": stats["p90_queue_time_seconds"], + "unique_jobs": len(stats.get("jobs_failed", {})), + "avg_queue": stats.get("avg_queue_time_seconds", 0), + "p90_queue": stats.get("p90_queue_time_seconds", 0), "queue_samples": stats.get("queue_time_samples", 0), + "first_failure": streak_data.get("first_failure_in_streak"), + "last_failure": streak_data.get("last_failure_in_streak"), + "top_error_signatures": streak_data.get( + "top_error_signatures", [] + ), } ) @@ -1299,53 +1694,70 @@ class SGLangFailuresAnalyzer: reverse=True, ) - summary_lines.append("### Top 15 Runners by Consecutive Failures") - summary_lines.append("") - summary_lines.append( - "| Rank | Runner Labels | Streak | Max | Fail Rate | Total | Unique Jobs | Avg Queue | P90 Queue |" - ) - summary_lines.append( - "|------|---------------|--------|-----|-----------|-------|-------------|-----------|-----------|" - ) - - for i, runner_data in enumerate(sorted_runners[:15], 1): - display_labels = ( - runner_data["runner_labels"] - if len(runner_data["runner_labels"]) <= 30 - else runner_data["runner_labels"][:27] + "..." - ) - - # Format streaks - streak_str = ( - str(runner_data["current_streak"]) - if runner_data["current_streak"] > 0 - else "-" - ) - max_str = ( - str(runner_data["max_streak"]) - if runner_data["max_streak"] > 0 - else "-" - ) - - # Format queue times - avg_queue_str = ( - f"{runner_data['avg_queue'] / 60:.1f}m" - if runner_data["queue_samples"] > 0 - else "N/A" - ) - p90_queue_str = ( - f"{runner_data['p90_queue'] / 60:.1f}m" - if runner_data["queue_samples"] > 0 - else "N/A" - ) + # Only show runners with streak >= 2 + runners_with_issues = [ + r for r in sorted_runners if r["current_streak"] >= 2 + ] + if runners_with_issues: summary_lines.append( - f"| {i} | `{display_labels}` | {streak_str} | {max_str} | {runner_data['failure_rate']:.1f}% | " - f"{runner_data['total_jobs']} | {runner_data['unique_jobs']} | " - f"{avg_queue_str} | {p90_queue_str} |" + "## Section 2: Top 15 Consecutively Failing Workers" + ) + summary_lines.append("") + summary_lines.append( + "| Machine Name | Streak | Max | Fail Rate | Avg Queue | First Failure | Last Failure | Top Errors | Total Jobs | Unique Jobs |" + ) + summary_lines.append( + "|--------------|--------|-----|-----------|-----------|---------------|--------------|------------|------------|-------------|" ) - summary_lines.append("") + for runner_data in runners_with_issues[:15]: + display_name = ( + runner_data["runner_name"] + if len(runner_data["runner_name"]) <= 28 + else runner_data["runner_name"][:25] + "..." + ) + + # Format streaks + streak_str = str(runner_data["current_streak"]) + max_str = str(runner_data["max_streak"]) + + # Format queue time + avg_queue_str = ( + f"{runner_data['avg_queue'] / 60:.1f}m" + if runner_data["queue_samples"] > 0 + else "N/A" + ) + + # Get first and last failure info + first_failure = runner_data.get("first_failure") + if first_failure: + first_failure_str = f"[Run #{first_failure['run_number']}]({first_failure.get('job_url', first_failure['url'])})" + else: + first_failure_str = "N/A" + + last_failure = runner_data.get("last_failure") + if last_failure: + last_failure_str = f"[Run #{last_failure['run_number']}]({last_failure.get('job_url', last_failure['url'])})" + else: + last_failure_str = "N/A" + + # Format top errors as bullet list + top_errors = runner_data.get("top_error_signatures", []) + if top_errors: + error_str = "
".join( + [f"• {err[0]} ({err[1]})" for err in top_errors] + ) + else: + error_str = "N/A" + + summary_lines.append( + f"| `{display_name}` | {streak_str} | {max_str} | {runner_data['failure_rate']:.1f}% | " + f"{avg_queue_str} | {first_failure_str} | {last_failure_str} | {error_str} | " + f"{runner_data['total_jobs']} | {runner_data['unique_jobs']} |" + ) + + summary_lines.append("") # Write summary with open(github_step_summary, "a", encoding="utf-8") as f: