chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

[Refactore] [CI] Remove redundant CI test runs step 2 (#17584)

Browse Source
This commit is contained in:
Makcum888e
2026-01-24 23:39:48 -08:00
committed by GitHub
parent 9bd92ba0f6
commit d1042e0d62
36 changed files with 199 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/ci/utils/ci_coverage_report.py
Executable
+477
View File
@@ -0,0 +1,477 @@
#!/usr/bin/env python3
"""
CI Coverage Report Generator
Collects all CI test registrations from test/registered/ and generates
a coverage report organized by folder, backend, and suite.
Usage:
python scripts/ci/utils/ci_coverage_report.py [--output-format markdown|json]
"""
import argparse
import glob
import json
import os
import sys
from collections import defaultdict
from pathlib import Path
# Add the ci_register module path directly to avoid heavy sglang imports
sys.path.insert(
0,
str(
Path(__file__).parent.parent.parent.parent / "python" / "sglang" / "test" / "ci"
),
)
from ci_register import CIRegistry, HWBackend, ut_parse_one_file
def collect_all_tests(registered_dir: str) -> list[CIRegistry]:
"""Collect all CI registrations from registered directory."""
files = glob.glob(f"{registered_dir}/**/*.py", recursive=True)
all_tests = []
for file in sorted(files):
try:
registries = ut_parse_one_file(file)
all_tests.extend(registries)
except Exception as e:
print(f"Warning: Failed to parse {file}: {e}", file=sys.stderr)
return all_tests
def get_folder_name(filename: str) -> str:
"""Extract folder name from test filename."""
# e.g., "registered/models/test_foo.py" -> "models"
parts = Path(filename).parts
if "registered" in parts:
idx = parts.index("registered")
if idx + 1 < len(parts) - 1: # Has subfolder
return parts[idx + 1]
return "root"
def get_test_basename(filename: str) -> str:
"""Extract just the test file name from the path."""
return Path(filename).name
def organize_test_data(tests: list[CIRegistry]) -> dict:
"""Organize tests into various groupings."""
by_backend = defaultdict(list)
by_folder = defaultdict(list)
disabled_tests = []
for t in tests:
by_backend[t.backend.name].append(t)
by_folder[get_folder_name(t.filename)].append(t)
if t.disabled:
disabled_tests.append(t)
# Count unique test files (a file may be registered for multiple backends)
unique_files = set(t.filename for t in tests)
unique_enabled_files = set(t.filename for t in tests if not t.disabled)
unique_disabled_files = set(t.filename for t in tests if t.disabled)
return {
"total": len(tests),
"total_unique_files": len(unique_files),
"enabled": len(tests) - len(disabled_tests),
"enabled_unique_files": len(unique_enabled_files),
"disabled_count": len(disabled_tests),
"disabled_unique_files": len(unique_disabled_files),
"by_backend": by_backend,
"by_folder": by_folder,
"disabled_tests": disabled_tests,
}
def generate_summary_section(data: dict) -> str:
"""Generate the summary/overview section."""
lines = []
lines.append("# CI Coverage Overview\n")
lines.append(
f"**Unique Test Files:** {data['total_unique_files']} ({data['enabled_unique_files']} enabled, {data['disabled_unique_files']} disabled)\n"
)
lines.append(
f"**Total Registrations:** {data['total']} ({data['enabled']} enabled, {data['disabled_count']} disabled)\n"
)
lines.append(
"*Note: A test file may be registered for multiple backends (e.g., CUDA + AMD), so total registrations > unique files.*\n"
)
by_backend = data["by_backend"]
by_folder = data["by_folder"]
disabled_tests = data["disabled_tests"]
# Backend summary (collapsible)
lines.append("<details>")
lines.append("<summary><h2>Backend Summary</h2></summary>\n")
lines.append("| Backend | Total | Enabled | Disabled | Per-Commit | Nightly |")
lines.append("|---------|-------|---------|----------|------------|---------|")
for backend in ["CUDA", "AMD", "NPU", "CPU"]:
backend_tests = by_backend.get(backend, [])
if not backend_tests:
continue
b_total = len(backend_tests)
b_disabled = sum(1 for t in backend_tests if t.disabled)
b_enabled = b_total - b_disabled
b_per_commit = sum(1 for t in backend_tests if not t.nightly and not t.disabled)
b_nightly = sum(1 for t in backend_tests if t.nightly and not t.disabled)
lines.append(
f"| {backend} | {b_total} | {b_enabled} | {b_disabled} | {b_per_commit} | {b_nightly} |"
)
lines.append("\n</details>\n")
# Folder summary (collapsible)
lines.append("<details>")
lines.append("<summary><h2>Folder Summary</h2></summary>\n")
lines.append("| Folder | CUDA | AMD | NPU | CPU | Total |")
lines.append("|--------|------|-----|-----|-----|-------|")
for folder in sorted(by_folder.keys()):
folder_tests = by_folder[folder]
cuda = sum(1 for t in folder_tests if t.backend == HWBackend.CUDA)
amd = sum(1 for t in folder_tests if t.backend == HWBackend.AMD)
npu = sum(1 for t in folder_tests if t.backend == HWBackend.NPU)
cpu = sum(1 for t in folder_tests if t.backend == HWBackend.CPU)
lines.append(
f"| {folder} | {cuda} | {amd} | {npu} | {cpu} | {len(folder_tests)} |"
)
lines.append("\n</details>\n")
# Disabled tests section (collapsible)
if disabled_tests:
lines.append("<details>")
lines.append("<summary><h2>Disabled Tests</h2></summary>\n")
lines.append("| File | Backend | Suite | Reason |")
lines.append("|------|---------|-------|--------|")
for t in sorted(disabled_tests, key=lambda x: (x.backend.name, x.filename)):
test_name = get_test_basename(t.filename)
reason = t.disabled[:50] + "..." if len(t.disabled) > 50 else t.disabled
lines.append(f"| `{test_name}` | {t.backend.name} | {t.suite} | {reason} |")
lines.append("\n</details>\n")
return "\n".join(lines)
def generate_by_folder_section(data: dict) -> str:
"""Generate the 'All Tests by Folder' section."""
lines = []
by_folder = data["by_folder"]
lines.append("# All Tests by Folder\n")
for folder in sorted(by_folder.keys()):
folder_tests = by_folder[folder]
lines.append("<details>")
lines.append(
f"<summary><h2>{folder}/ ({len(folder_tests)} tests)</h2></summary>\n"
)
# Group by backend within folder
folder_by_backend = defaultdict(list)
for t in folder_tests:
folder_by_backend[t.backend.name].append(t)
for backend in ["CUDA", "AMD", "NPU", "CPU"]:
backend_tests = folder_by_backend.get(backend, [])
if not backend_tests:
continue
lines.append(f"### {backend} ({len(backend_tests)} tests)\n")
lines.append("| Test File | Suite | Est. Time | Status |")
lines.append("|-----------|-------|-----------|--------|")
for t in sorted(backend_tests, key=lambda x: x.filename):
test_name = get_test_basename(t.filename)
status = (
"Disabled"
if t.disabled
else ("Nightly" if t.nightly else "Per-Commit")
)
lines.append(
f"| `{test_name}` | {t.suite} | {t.est_time:.0f}s | {status} |"
)
lines.append("")
lines.append("</details>\n")
return "\n".join(lines)
def generate_by_suite_section(data: dict) -> str:
"""Generate the 'All Tests by Test Suite' section."""
lines = []
by_backend = data["by_backend"]
lines.append("# All Tests by Test Suite\n")
for backend in ["CUDA", "AMD", "NPU", "CPU"]:
backend_tests = by_backend.get(backend, [])
if not backend_tests:
continue
b_total = len(backend_tests)
b_disabled = sum(1 for t in backend_tests if t.disabled)
b_enabled = b_total - b_disabled
lines.append("<details>")
lines.append(
f"<summary><h2>{backend} Backend ({b_enabled} enabled, {b_disabled} disabled)</h2></summary>\n"
)
# Group by suite within backend
backend_suites = defaultdict(list)
for t in backend_tests:
backend_suites[t.suite].append(t)
for suite in sorted(backend_suites.keys()):
suite_tests = backend_suites[suite]
s_enabled = sum(1 for t in suite_tests if not t.disabled)
s_disabled = sum(1 for t in suite_tests if t.disabled)
s_est_time = sum(t.est_time for t in suite_tests if not t.disabled)
is_nightly = any(t.nightly for t in suite_tests if not t.disabled)
suite_type = "Nightly" if is_nightly else "Per-Commit"
lines.append("<details>")
lines.append(
f"<summary><h3>{suite} ({s_enabled} enabled, {s_disabled} disabled) - {suite_type}</h3></summary>\n"
)
lines.append(f"*Estimated total time: {s_est_time:.0f}s*\n")
lines.append("| Test File | Folder | Est. Time | Status |")
lines.append("|-----------|--------|-----------|--------|")
for t in sorted(suite_tests, key=lambda x: x.filename):
test_name = get_test_basename(t.filename)
folder = get_folder_name(t.filename)
if t.disabled:
status = (
f"Disabled: {t.disabled[:30]}..."
if len(t.disabled) > 30
else f"Disabled: {t.disabled}"
)
else:
status = "Nightly" if t.nightly else "Per-Commit"
lines.append(
f"| `{test_name}` | {folder} | {t.est_time:.0f}s | {status} |"
)
lines.append("\n</details>\n")
lines.append("</details>\n")
return "\n".join(lines)
def generate_markdown_report(tests: list[CIRegistry], section: str = "all") -> str:
"""Generate markdown report for GitHub step summary."""
data = organize_test_data(tests)
if section == "summary":
return generate_summary_section(data)
elif section == "by-folder":
return generate_by_folder_section(data)
elif section == "by-suite":
return generate_by_suite_section(data)
else: # "all"
parts = [
generate_summary_section(data),
"---",
generate_by_folder_section(data),
"---",
generate_by_suite_section(data),
]
return "\n".join(parts)
def generate_json_report(tests: list[CIRegistry]) -> str:
"""Generate JSON report with detailed test listings."""
by_backend = defaultdict(list)
by_folder = defaultdict(list)
for t in tests:
by_backend[t.backend.name].append(t)
by_folder[get_folder_name(t.filename)].append(t)
disabled_tests = [t for t in tests if t.disabled]
# Build structured data
data = {
"summary": {
"total": len(tests),
"enabled": len(tests) - len(disabled_tests),
"disabled": len(disabled_tests),
},
"tests_by_folder": {},
"tests_by_suite": {},
"backend_summary": {},
"folder_summary": {},
"disabled_tests": [],
}
# Section 1: Tests by Folder
for folder in sorted(by_folder.keys()):
folder_tests = by_folder[folder]
folder_by_backend = defaultdict(list)
for t in folder_tests:
folder_by_backend[t.backend.name].append(t)
data["tests_by_folder"][folder] = {
"total": len(folder_tests),
"backends": {},
}
for backend in ["CUDA", "AMD", "NPU", "CPU"]:
backend_tests = folder_by_backend.get(backend, [])
if backend_tests:
data["tests_by_folder"][folder]["backends"][backend] = [
{
"filename": get_test_basename(t.filename),
"suite": t.suite,
"est_time": t.est_time,
"status": (
"disabled"
if t.disabled
else ("nightly" if t.nightly else "per-commit")
),
}
for t in sorted(backend_tests, key=lambda x: x.filename)
]
# Section 2: Tests by Suite (Backend -> Suite)
for backend in ["CUDA", "AMD", "NPU", "CPU"]:
backend_tests = by_backend.get(backend, [])
if not backend_tests:
continue
backend_suites = defaultdict(list)
for t in backend_tests:
backend_suites[t.suite].append(t)
data["tests_by_suite"][backend] = {
"total": len(backend_tests),
"enabled": sum(1 for t in backend_tests if not t.disabled),
"disabled": sum(1 for t in backend_tests if t.disabled),
"suites": {},
}
for suite in sorted(backend_suites.keys()):
suite_tests = backend_suites[suite]
is_nightly = any(t.nightly for t in suite_tests if not t.disabled)
data["tests_by_suite"][backend]["suites"][suite] = {
"total": len(suite_tests),
"enabled": sum(1 for t in suite_tests if not t.disabled),
"disabled": sum(1 for t in suite_tests if t.disabled),
"est_time": sum(t.est_time for t in suite_tests if not t.disabled),
"type": "nightly" if is_nightly else "per-commit",
"tests": [
{
"filename": get_test_basename(t.filename),
"folder": get_folder_name(t.filename),
"est_time": t.est_time,
"status": (
"disabled"
if t.disabled
else ("nightly" if t.nightly else "per-commit")
),
"disabled_reason": t.disabled if t.disabled else None,
}
for t in sorted(suite_tests, key=lambda x: x.filename)
],
}
# Backend summary
for backend in ["CUDA", "AMD", "NPU", "CPU"]:
backend_tests = by_backend.get(backend, [])
if backend_tests:
data["backend_summary"][backend] = {
"total": len(backend_tests),
"enabled": sum(1 for t in backend_tests if not t.disabled),
"disabled": sum(1 for t in backend_tests if t.disabled),
"per_commit": sum(
1 for t in backend_tests if not t.nightly and not t.disabled
),
"nightly": sum(
1 for t in backend_tests if t.nightly and not t.disabled
),
}
# Folder summary
for folder in sorted(by_folder.keys()):
folder_tests = by_folder[folder]
data["folder_summary"][folder] = {
"CUDA": sum(1 for t in folder_tests if t.backend == HWBackend.CUDA),
"AMD": sum(1 for t in folder_tests if t.backend == HWBackend.AMD),
"NPU": sum(1 for t in folder_tests if t.backend == HWBackend.NPU),
"CPU": sum(1 for t in folder_tests if t.backend == HWBackend.CPU),
"total": len(folder_tests),
}
# Disabled tests
for t in sorted(disabled_tests, key=lambda x: (x.backend.name, x.filename)):
data["disabled_tests"].append(
{
"filename": get_test_basename(t.filename),
"backend": t.backend.name,
"suite": t.suite,
"reason": t.disabled,
}
)
return json.dumps(data, indent=2)
def main():
parser = argparse.ArgumentParser(description="Generate CI coverage report")
parser.add_argument(
"--output-format",
choices=["markdown", "json"],
default="markdown",
help="Output format (default: markdown)",
)
parser.add_argument(
"--section",
choices=["all", "summary", "by-folder", "by-suite"],
default="all",
help="Which section to output (default: all). Only applies to markdown format.",
)
parser.add_argument(
"--registered-dir",
default="test/registered",
help="Path to registered test directory",
)
args = parser.parse_args()
# Change to repo root if needed
script_dir = Path(__file__).parent.parent
repo_root = script_dir.parent.parent
os.chdir(repo_root)
tests = collect_all_tests(args.registered_dir)
if args.output_format == "markdown":
report = generate_markdown_report(tests, section=args.section)
else:
report = generate_json_report(tests)
print(report)
# Write to GITHUB_STEP_SUMMARY if available
summary_file = os.environ.get("GITHUB_STEP_SUMMARY")
if summary_file and args.output_format == "markdown":
with open(summary_file, "a") as f:
f.write(report)
if __name__ == "__main__":
main()
scripts/ci/utils/cleanup_hf_cache.py
Executable
+146
View File
@@ -0,0 +1,146 @@
#!/usr/bin/env python3
"""
Clean up stale HuggingFace cache artifacts from previous failed downloads.
This script removes incomplete marker files, temporary files, and lock files
from the HuggingFace cache directory. These artifacts can accumulate from
interrupted or failed downloads and may interfere with future downloads.
"""
import os
import sys
from pathlib import Path
from typing import List
try:
from huggingface_hub import constants
HF_HUB_AVAILABLE = True
except ImportError:
print("Warning: huggingface_hub not available")
HF_HUB_AVAILABLE = False
def get_hf_cache_dir() -> str:
"""Get the HuggingFace cache directory."""
if HF_HUB_AVAILABLE:
return constants.HF_HUB_CACHE
# Fallback to environment variable or default
hf_home = os.environ.get("HF_HOME", os.path.expanduser("~/.cache/huggingface"))
return os.path.join(hf_home, "hub")
def find_stale_artifacts(cache_dir: str) -> List[Path]:
"""
Find stale artifact files in the HuggingFace cache.
Args:
cache_dir: HuggingFace cache directory
Returns:
List of paths to stale artifact files
"""
cache_path = Path(cache_dir)
if not cache_path.exists():
return []
# Patterns for stale files to clean up
patterns = [
"**/*.incomplete", # Incomplete download markers
"**/*.tmp", # Temporary files
"**/*.lock", # Lock files from interrupted downloads
]
stale_files = []
for pattern in patterns:
stale_files.extend(cache_path.glob(pattern))
return stale_files
def cleanup_artifacts(artifacts: List[Path]) -> tuple[int, int]:
"""
Remove stale artifact files.
Args:
artifacts: List of file paths to remove
Returns:
Tuple of (successful_removals, failed_removals)
"""
successful = 0
failed = 0
for file_path in artifacts:
try:
file_path.unlink()
print(f" Removed: {file_path}")
successful += 1
except Exception as e:
print(f" Warning: Could not remove {file_path}: {e}")
failed += 1
return successful, failed
def main() -> int:
"""
Main cleanup logic.
Returns:
Always returns 0 (cleanup is best-effort and should not fail CI)
"""
print("=" * 70)
print("HuggingFace Cache Cleanup")
print("=" * 70)
# Get cache directory
cache_dir = get_hf_cache_dir()
print(f"Cache directory: {cache_dir}")
if not os.path.exists(cache_dir):
print("Cache directory does not exist - nothing to clean")
return 0
print("-" * 70)
# Find stale artifacts
print("Scanning for stale artifacts...")
stale_artifacts = find_stale_artifacts(cache_dir)
if not stale_artifacts:
print("✓ No stale cache artifacts found")
return 0
# Clean up artifacts
print(f"Found {len(stale_artifacts)} stale artifact(s) to remove:")
successful, failed = cleanup_artifacts(stale_artifacts)
print("-" * 70)
# Summary
if failed > 0:
print(f"⚠ Cleaned up {successful} file(s), {failed} removal(s) failed")
else:
print(f"✓ Successfully cleaned up {successful} stale file(s)")
# Always return 0 - cleanup failures should not fail CI
return 0
if __name__ == "__main__":
try:
exit_code = main()
sys.exit(exit_code)
except KeyboardInterrupt:
print("\nInterrupted by user")
sys.exit(0)
except Exception as e:
print(f"ERROR: Unexpected error during cleanup: {e}")
import traceback
traceback.print_exc()
# Still return 0 - cleanup failures should not fail CI
sys.exit(0)
scripts/ci/utils/prevalidate_cached_models.py
Executable
+407
View File
@@ -0,0 +1,407 @@
#!/usr/bin/env python3
"""
Pre-validate all cached HuggingFace models to provide detailed feedback.
This script runs once during CI initialization (in prepare_runner.sh) to:
1. Scan snapshots in ~/.cache/huggingface/hub/ (with time/quantity limits)
2. Validate completeness (config/tokenizer/weights)
3. Output detailed failure reasons for debugging
NOTE: This script no longer writes shared validation markers. Each test run
independently validates its cache using per-run markers to avoid cross-runner
cache state pollution.
"""
import glob
import json
import os
import sys
import time
from pathlib import Path
# Add python directory to path to import sglang modules
REPO_ROOT = Path(__file__).parent.parent.parent.parent
sys.path.insert(0, str(REPO_ROOT / "python"))
from sglang.srt.model_loader.ci_weight_validation import ( # noqa: E402
_validate_diffusion_model,
validate_cache_with_detailed_reason,
)
# Limits to avoid spending too much time on validation
MAX_VALIDATION_TIME_SECONDS = 300 # Max 5 minutes total
def find_all_hf_snapshots():
"""
Find all HuggingFace snapshots in cache.
Returns:
List of (model_name, snapshot_dir) tuples, sorted by mtime (newest first)
"""
hf_home = os.environ.get("HF_HOME", os.path.expanduser("~/.cache/huggingface"))
hub_dir = os.path.join(hf_home, "hub")
if not os.path.isdir(hub_dir):
print(f"HF hub directory not found: {hub_dir}")
return []
snapshots = []
# Pattern: models--org--model/snapshots/hash
for model_dir in glob.glob(os.path.join(hub_dir, "models--*")):
# Extract model name from directory (models--org--model -> org/model)
dir_name = os.path.basename(model_dir)
if not dir_name.startswith("models--"):
continue
# models--meta-llama--Llama-2-7b-hf -> meta-llama/Llama-2-7b-hf
# Handle multi-part names: models--a--b--c -> a/b-c (join parts 1+ with /)
parts = dir_name.split("--")
if len(parts) < 3 or parts[0] != "models":
# Invalid format, skip
continue
# Standard format: models--org--repo -> org/repo
# Extended format: models--org--repo--extra -> org/repo-extra (join with -)
model_name = parts[1] + "/" + "-".join(parts[2:])
snapshots_dir = os.path.join(model_dir, "snapshots")
if not os.path.isdir(snapshots_dir):
continue
# Find all snapshot hashes
for snapshot_hash_dir in os.listdir(snapshots_dir):
snapshot_path = os.path.join(snapshots_dir, snapshot_hash_dir)
if os.path.isdir(snapshot_path):
try:
mtime = os.path.getmtime(snapshot_path)
snapshots.append((model_name, snapshot_path, mtime))
except OSError:
continue
# Sort by mtime (newest first) - prioritize recently used models
snapshots.sort(key=lambda x: x[2], reverse=True)
# Return without mtime
return [(name, path) for name, path, _ in snapshots]
def is_transformers_text_model(snapshot_dir):
"""
Check if a snapshot is a transformers text model.
Only excludes (returns False) for models with STRONG evidence of being
diffusers/generation pipelines. Uses conservative heuristics to avoid
false negatives on multimodal LLMs with tokenizers.
Args:
snapshot_dir: Path to snapshot directory
Returns:
True if this looks like a transformers text model, False otherwise (N/A)
"""
# Check for diffusers pipeline markers (strong evidence)
diffusers_markers = [
"model_index.json", # Diffusers pipeline config
"scheduler", # Scheduler directory (diffusers)
]
if any(
os.path.exists(os.path.join(snapshot_dir, marker))
for marker in diffusers_markers
):
return False
config_path = os.path.join(snapshot_dir, "config.json")
if not os.path.exists(config_path):
# No config.json - likely not a transformers model
return False
try:
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
# Check for explicit diffusers/generation model types (conservative keywords)
model_type = config.get("_class_name") or config.get("model_type")
if model_type:
model_type_lower = str(model_type).lower()
# Only exclude clear diffusion/generation models
if any(
keyword in model_type_lower
for keyword in [
"diffusion",
"unet",
"vae",
"controlnet",
"stable-diffusion",
"latent-diffusion",
]
):
return False
# Check architectures for explicit generation/diffusion classes
architectures = config.get("architectures", [])
if architectures:
arch_str = " ".join(architectures).lower()
# Conservative: only exclude obvious diffusion/generation architectures
# Use word boundaries to avoid false positives (e.g., "dit" in "conditional")
for keyword in [
"diffusion",
"unet2d",
"unet3d",
"vaedecoder", # More specific than "vae"
"vaeencoder",
"controlnet",
"autoencoder",
"ditmodel", # Diffusion Transformer - use more specific pattern
"pixart", # PixArt diffusion model
]:
if keyword in arch_str:
return False
# Check for standalone vision encoder/image processor (no text component)
# Only if model name explicitly indicates non-text usage
model_name = config.get("_name_or_path", "").lower()
if any(
keyword in model_name
for keyword in [
"image-edit-", # Pure image editing (e.g., Qwen-Image-Edit)
"-image-editing",
"dit-", # DiT generation models
"pixart-", # PixArt generation models
]
):
# Additional check: does it have tokenizer? If yes, might be multimodal LLM
has_tokenizer = any(
os.path.exists(os.path.join(snapshot_dir, fname))
for fname in ["tokenizer.json", "tokenizer.model", "tiktoken.model"]
)
if not has_tokenizer:
# Image-edit model without tokenizer -> likely pure vision pipeline
return False
# Default: assume it's a transformers text/multimodal model
# Even if it lacks tokenizer, let validation report the actual error
# (better false positive than false negative for text models)
return True
except (json.JSONDecodeError, OSError, KeyError):
# Can't parse config - assume it's transformers and let validation report failure
return True
def scan_weight_files(snapshot_dir):
"""
Scan for weight files in a snapshot.
Returns:
List of weight file paths, or empty list if scan fails
"""
weight_files = []
# First, look for index files
index_patterns = ["*.safetensors.index.json", "pytorch_model.bin.index.json"]
index_files = []
for pattern in index_patterns:
index_files.extend(glob.glob(os.path.join(snapshot_dir, pattern)))
# If we have safetensors index, collect shards from it
for index_file in index_files:
if index_file.endswith(".safetensors.index.json"):
try:
with open(index_file, "r", encoding="utf-8") as f:
index_data = json.load(f)
weight_map = index_data.get("weight_map", {})
for weight_file in set(weight_map.values()):
weight_path = os.path.join(snapshot_dir, weight_file)
if os.path.exists(weight_path):
weight_files.append(weight_path)
except Exception as e:
print(
f" Warning: Failed to parse index {os.path.basename(index_file)}: {e}"
)
# If no index found or no shards from index, do recursive glob
if not weight_files:
matched = glob.glob(
os.path.join(snapshot_dir, "**/*.safetensors"), recursive=True
)
MAX_WEIGHT_FILES = 1000
if len(matched) > MAX_WEIGHT_FILES:
print(
f" Warning: Too many safetensors files ({len(matched)} > {MAX_WEIGHT_FILES})"
)
return []
for f in matched:
if os.path.exists(f): # Filter out broken symlinks
weight_files.append(f)
return weight_files
def validate_snapshot(model_name, snapshot_dir, weight_files, validated_cache):
"""
Validate a snapshot and return detailed status.
Uses in-process cache to avoid duplicate validation within the same run.
Args:
model_name: Model identifier
snapshot_dir: Path to snapshot directory
weight_files: List of weight files to validate
validated_cache: Dict to track already-validated snapshots in this run
Returns:
Tuple of (result, reason):
- (True, None) if validation passed
- (False, reason_str) if validation failed
- (None, None) if skipped (already validated in this run)
"""
# Fast path: check in-process cache first
if snapshot_dir in validated_cache:
return None, None # Already validated in this run, skip
try:
# Perform validation with detailed reason
is_complete, reason = validate_cache_with_detailed_reason(
snapshot_dir=snapshot_dir,
weight_files=weight_files,
model_name_or_path=model_name,
)
# Cache result to avoid re-validation in this run
validated_cache[snapshot_dir] = (is_complete, reason)
return is_complete, reason
except Exception as e:
error_msg = f"Validation raised exception: {e}"
return False, error_msg
def main():
start_time = time.time()
print("=" * 70)
print("CI_OFFLINE: Pre-validating cached HuggingFace models")
print("=" * 70)
print(f"Max time: {MAX_VALIDATION_TIME_SECONDS}s")
print()
print("Scanning HuggingFace cache for models...")
snapshots = find_all_hf_snapshots()
if not snapshots:
print("No cached models found, skipping validation")
print("=" * 70)
return
print(f"Found {len(snapshots)} snapshot(s) in cache")
print()
validated_count = 0
failed_count = 0
skipped_count = 0
processed_count = 0
# In-process cache to avoid re-validating same snapshot in this run
validated_cache = {}
for model_name, snapshot_dir in snapshots:
# Check time limit
elapsed = time.time() - start_time
if elapsed > MAX_VALIDATION_TIME_SECONDS:
print()
print(
f"Time limit reached ({elapsed:.1f}s > {MAX_VALIDATION_TIME_SECONDS}s)"
)
print(
f"Stopping validation, {len(snapshots) - processed_count} snapshots remaining"
)
break
snapshot_hash = os.path.basename(snapshot_dir)
print(
f"[{processed_count + 1}/{len(snapshots)}] {model_name} ({snapshot_hash[:8]}...)"
)
processed_count += 1
# Determine model type by checking for model_index.json (diffusers pipeline marker)
model_index_path = os.path.join(snapshot_dir, "model_index.json")
is_diffusion_model = os.path.exists(model_index_path)
if is_diffusion_model:
# This is a diffusers pipeline - use diffusion validation
try:
is_valid, reason = _validate_diffusion_model(snapshot_dir)
if is_valid:
print(" PASS (diffusion) - Cache complete & valid")
validated_count += 1
else:
print(f" FAIL (diffusion) - {reason}")
failed_count += 1
except Exception as e:
print(f" FAIL (diffusion) - Validation raised exception: {e}")
failed_count += 1
continue
# Transformers model - use standard validation
# First check if this looks like a transformers text model
if not is_transformers_text_model(snapshot_dir):
# Not a recognized model type, skip
print(
" SKIP (unknown type) - Not a diffusers pipeline or transformers model"
)
skipped_count += 1
continue
# Scan weight files
weight_files = scan_weight_files(snapshot_dir)
if not weight_files:
print(" SKIP (no weights) - empty or incomplete download")
skipped_count += 1
continue
# Validate
try:
result, reason = validate_snapshot(
model_name, snapshot_dir, weight_files, validated_cache
)
if result is True:
print(" PASS - Cache complete & valid")
validated_count += 1
elif result is False:
# Print detailed failure reason
if reason:
print(f" FAIL (incomplete) - {reason}")
else:
print(" FAIL (incomplete) - cache validation failed")
failed_count += 1
else: # None (skipped)
print(" SKIP (already validated in this run)")
skipped_count += 1
except Exception as e:
print(f" FAIL (error) - Validation raised exception: {e}")
failed_count += 1
elapsed_total = time.time() - start_time
print()
print("=" * 70)
print(f"Validation summary (completed in {elapsed_total:.1f}s):")
print(f" PASS (complete & valid): {validated_count}")
print(f" FAIL (incomplete/corrupted): {failed_count}")
print(f" SKIP (no weights/duplicate): {skipped_count}")
print(f" Total processed: {processed_count}/{len(snapshots)}")
print("=" * 70)
if __name__ == "__main__":
main()
scripts/ci/utils/publish_traces.py
+502
View File
@@ -0,0 +1,502 @@
"""
Publish performance traces to GitHub repository
"""
import argparse
import base64
import json
import os
import sys
import time
import warnings
from urllib.error import HTTPError
from urllib.request import Request, urlopen
def is_rate_limit_error(e):
"""Check if an exception is a GitHub rate limit error (not permission error)"""
if not isinstance(e, HTTPError):
return False
if e.code == 429:
return True
if e.code == 403:
# 403 can be rate limit OR permission error - check the message
error_body = getattr(e, "error_body", "")
if isinstance(error_body, str):
# Rate limit errors contain specific phrases
rate_limit_phrases = [
"rate limit",
"abuse detection",
"secondary rate limit",
]
return any(phrase in error_body.lower() for phrase in rate_limit_phrases)
return False
def is_permission_error(e):
"""Check if an exception is a GitHub permission error"""
if not isinstance(e, HTTPError) or e.code != 403:
return False
error_body = getattr(e, "error_body", "")
if isinstance(error_body, str):
permission_phrases = [
"resource not accessible",
"must have push access",
"permission",
"denied",
]
return any(phrase in error_body.lower() for phrase in permission_phrases)
return False
def make_github_request(url, token, method="GET", data=None):
"""Make authenticated request to GitHub API"""
headers = {
"Accept": "application/vnd.github+json",
"Authorization": f"Bearer {token}",
# "User-Agent": "sglang-ci",
"X-GitHub-Api-Version": "2022-11-28",
}
if data:
headers["Content-Type"] = "application/json"
data = json.dumps(data).encode("utf-8")
req = Request(url, data=data, headers=headers, method=method)
try:
with urlopen(req) as response:
return response.read().decode("utf-8")
except HTTPError as e:
print(f"GitHub API request failed: {e}")
try:
error_body = e.read().decode("utf-8")
print(f"Error response body: {error_body}")
e.error_body = error_body # Attach for later inspection
except Exception:
e.error_body = ""
raise
except Exception as e:
print(f"GitHub API request failed with a non-HTTP error: {e}")
raise
def verify_token_permissions(repo_owner, repo_name, token):
"""Verify that the token has necessary permissions for the repository"""
print("Verifying token permissions...")
checks = [
(
f"https://api.github.com/repos/{repo_owner}/{repo_name}", # Check if we can access the repository
"Repository access verified",
),
(
f"https://api.github.com/repos/{repo_owner}/{repo_name}/contents", # Check if we can read the repository contents
"Repository contents access verified",
),
]
for url, success_message in checks:
try:
response = make_github_request(url, token)
if success_message == "Repository access verified":
repo_data = json.loads(response)
print(f"{success_message}: {repo_data['full_name']}")
else:
print(success_message)
except Exception as e:
if is_rate_limit_error(e):
warnings.warn(
"GitHub API rate limit exceeded during token verification."
)
return "rate_limited"
print(f"Failed to verify permissions for {url}: {e}")
return False
return True
def get_branch_sha(repo_owner, repo_name, branch, token):
"""Get SHA of the branch head"""
url = (
f"https://api.github.com/repos/{repo_owner}/{repo_name}/git/refs/heads/{branch}"
)
response = make_github_request(url, token)
data = json.loads(response)
return data["object"]["sha"]
def get_tree_sha(repo_owner, repo_name, commit_sha, token):
"""Get tree SHA from commit"""
url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/git/commits/{commit_sha}"
response = make_github_request(url, token)
data = json.loads(response)
return data["tree"]["sha"]
def create_blob(repo_owner, repo_name, content, token, max_retries=3):
"""Create a blob with file content"""
url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/git/blobs"
# Encode content as base64 for GitHub API
content_b64 = base64.b64encode(content).decode("utf-8")
data = {"content": content_b64, "encoding": "base64"}
for attempt in range(max_retries):
try:
response = make_github_request(url, token, method="POST", data=data)
return json.loads(response)["sha"]
except Exception as e:
# Don't retry on rate limit errors - fail fast
if is_rate_limit_error(e):
raise
if attempt < max_retries - 1:
wait_time = 2**attempt # Exponential backoff: 1s, 2s, 4s
print(
f"Blob creation failed (attempt {attempt + 1}/{max_retries}), retrying in {wait_time}s..."
)
time.sleep(wait_time)
else:
raise
def create_blobs(repo_owner, repo_name, files, token):
"""Create blobs for all files and return tree items with blob SHAs"""
tree_items = []
for i, (file_path, content) in enumerate(files):
# Create blob first to get SHA
blob_sha = create_blob(repo_owner, repo_name, content, token)
tree_items.append(
{
"path": file_path,
"mode": "100644",
"type": "blob",
"sha": blob_sha,
}
)
# Progress indicator for large uploads
if (i + 1) % 10 == 0 or (i + 1) == len(files):
print(f"Created {i + 1}/{len(files)} blobs...")
return tree_items
def create_tree(repo_owner, repo_name, base_tree_sha, tree_items, token, max_retries=3):
"""Create a new tree from pre-created blob SHAs"""
url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/git/trees"
data = {"base_tree": base_tree_sha, "tree": tree_items}
for attempt in range(max_retries):
try:
response = make_github_request(url, token, method="POST", data=data)
return json.loads(response)["sha"]
except Exception as e:
# Don't retry on rate limit errors - fail fast
if is_rate_limit_error(e):
raise
if attempt < max_retries - 1:
wait_time = 2**attempt
print(
f"Tree creation failed (attempt {attempt + 1}/{max_retries}), retrying in {wait_time}s..."
)
time.sleep(wait_time)
else:
raise
def create_commit(
repo_owner, repo_name, tree_sha, parent_sha, message, token, max_retries=3
):
"""Create a new commit"""
url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/git/commits"
data = {"tree": tree_sha, "parents": [parent_sha], "message": message}
for attempt in range(max_retries):
try:
response = make_github_request(url, token, method="POST", data=data)
commit_sha = json.loads(response)["sha"]
# Verify the commit was actually created
verify_url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/git/commits/{commit_sha}"
verify_response = make_github_request(verify_url, token)
verify_data = json.loads(verify_response)
if verify_data["sha"] != commit_sha:
raise Exception(
f"Commit verification failed: expected {commit_sha}, got {verify_data['sha']}"
)
return commit_sha
except Exception as e:
# Don't retry on rate limit errors - fail fast
if is_rate_limit_error(e):
raise
if attempt < max_retries - 1:
wait_time = 2**attempt
print(
f"Commit creation failed (attempt {attempt + 1}/{max_retries}), retrying in {wait_time}s..."
)
time.sleep(wait_time)
else:
raise
def update_branch_ref(repo_owner, repo_name, branch, commit_sha, token, max_retries=3):
"""Update branch reference to point to new commit"""
url = (
f"https://api.github.com/repos/{repo_owner}/{repo_name}/git/refs/heads/{branch}"
)
data = {"sha": commit_sha}
for attempt in range(max_retries):
try:
make_github_request(url, token, method="PATCH", data=data)
return
except HTTPError as e:
# Don't retry on rate limit errors - fail fast
if is_rate_limit_error(e):
raise
# Check if this is an "Object does not exist" error
is_object_not_exist = False
if hasattr(e, "error_body"):
try:
error_data = json.loads(e.error_body)
if "Object does not exist" in error_data.get("message", ""):
is_object_not_exist = True
except Exception:
pass
if is_object_not_exist and attempt < max_retries - 1:
# This might be a transient consistency issue - wait and retry
wait_time = 2**attempt
print(
f"Branch update failed with 'Object does not exist' (attempt {attempt + 1}/{max_retries}), waiting {wait_time}s for consistency..."
)
time.sleep(wait_time)
else:
raise
except Exception as e:
# Don't retry on rate limit errors - fail fast
if is_rate_limit_error(e):
raise
if attempt < max_retries - 1:
wait_time = 2**attempt
print(
f"Branch update failed (attempt {attempt + 1}/{max_retries}), retrying in {wait_time}s..."
)
time.sleep(wait_time)
else:
raise
def copy_trace_files(source_dir, target_base_path):
"""Copy trace files and return list of files to upload.
Only uploads traces from TP rank 0 to avoid duplicated data across tensor parallel ranks.
"""
files_to_upload = []
if not os.path.exists(source_dir):
print(f"Warning: Traces directory {source_dir} does not exist")
return files_to_upload
# Walk through source directory and find .json.gz files
for root, dirs, files in os.walk(source_dir):
for file in files:
if file.endswith(".json.gz"):
# Only upload TP rank 0 traces to avoid duplicates across tensor parallel ranks
if "TP-" in file and "TP-0" not in file:
continue
source_file = os.path.join(root, file)
# Calculate relative path from source_dir
rel_path = os.path.relpath(source_file, source_dir)
target_path = f"{target_base_path}/{rel_path}"
# Read file content
with open(source_file, "rb") as f:
content = f.read()
files_to_upload.append((target_path, content))
return files_to_upload
def publish_traces(traces_dir, run_id, run_number):
"""Publish traces to GitHub repository in a single commit"""
# Get environment variables
token = os.getenv("GITHUB_TOKEN")
if not token:
print("Error: GITHUB_TOKEN environment variable not set")
sys.exit(1)
# Repository configuration
repo_owner = "sglang-bot"
repo_name = "sglang-ci-data"
branch = "main"
target_base_path = f"traces/{run_id}"
# Copy trace files
files_to_upload = copy_trace_files(traces_dir, target_base_path)
if not files_to_upload:
print("No trace files found to upload")
return
print(f"Found {len(files_to_upload)} files to upload")
# Verify token permissions before proceeding
permission_check = verify_token_permissions(repo_owner, repo_name, token)
if permission_check == "rate_limited":
warnings.warn(
"Skipping trace upload due to GitHub API rate limit. "
"This is expected during high CI activity and does not indicate a test failure."
)
return
elif not permission_check:
print(
"Token permission verification failed. Please check the token permissions."
)
sys.exit(1)
max_retries = 5
retry_delay = 5 # seconds
# Create blobs once before retry loop to avoid re-uploading on failures
try:
tree_items = create_blobs(repo_owner, repo_name, files_to_upload, token)
except Exception as e:
# Check for rate limit errors during blob creation
if is_rate_limit_error(e):
warnings.warn(
"GitHub API rate limit exceeded during blob creation. Skipping trace upload."
)
return
# Check for permission errors - these should fail loudly
if is_permission_error(e):
print(
f"ERROR: Token does not have write permission to {repo_owner}/{repo_name}. "
"Please update the GH_PAT_FOR_NIGHTLY_CI_DATA secret with a token that has "
"'contents: write' permission for the repository."
)
sys.exit(1)
print(f"Failed to create blobs: {e}")
raise
for attempt in range(max_retries):
try:
# Get current branch head
branch_sha = get_branch_sha(repo_owner, repo_name, branch, token)
print(f"Current branch head: {branch_sha}")
# Get current tree
tree_sha = get_tree_sha(repo_owner, repo_name, branch_sha, token)
print(f"Current tree SHA: {tree_sha}")
# Create new tree with pre-created blobs
new_tree_sha = create_tree(
repo_owner, repo_name, tree_sha, tree_items, token
)
print(f"Created new tree: {new_tree_sha}")
# Create commit
commit_message = f"Nightly traces for run {run_id} at {run_number} ({len(files_to_upload)} files)"
commit_sha = create_commit(
repo_owner,
repo_name,
new_tree_sha,
branch_sha,
commit_message,
token,
)
print(f"Created commit: {commit_sha}")
# Update branch reference
update_branch_ref(repo_owner, repo_name, branch, commit_sha, token)
print("Updated branch reference")
print("Successfully published all traces in a single commit")
return
except Exception as e:
# Check for retryable errors
is_retryable = False
error_type = "unknown"
if hasattr(e, "error_body"):
if "Update is not a fast forward" in e.error_body:
is_retryable = True
error_type = "fast-forward conflict"
elif "Object does not exist" in e.error_body:
is_retryable = True
error_type = "object consistency"
# Also retry on HTTP errors that might be transient
if isinstance(e, HTTPError) and e.code in [422, 500, 502, 503, 504]:
is_retryable = True
error_type = f"HTTP {e.code}"
# Check for rate limit errors (non-fatal - just warn and skip)
if is_rate_limit_error(e):
warnings.warn("GitHub API rate limit exceeded. Skipping trace upload.")
return
# Check for permission errors - these should fail loudly
if is_permission_error(e):
print(
f"ERROR: Token does not have write permission to {repo_owner}/{repo_name}. "
"Please update the GH_PAT_FOR_NIGHTLY_CI_DATA secret with a token that has "
"'contents: write' permission for the repository."
)
sys.exit(1)
if is_retryable and attempt < max_retries - 1:
print(
f"Attempt {attempt + 1}/{max_retries} failed ({error_type}). Retrying in {retry_delay} seconds..."
)
time.sleep(retry_delay)
else:
print(f"Failed to publish traces after {attempt + 1} attempts: {e}")
raise
def main():
parser = argparse.ArgumentParser(
description="Publish performance traces to GitHub repository"
)
parser.add_argument(
"--traces-dir",
type=str,
required=True,
help="Traces directory to publish",
)
args = parser.parse_args()
# Get environment variables
run_id = os.getenv("GITHUB_RUN_ID", "test")
run_number = os.getenv("GITHUB_RUN_NUMBER", "12345")
if not run_id or not run_number:
print(
"Error: GITHUB_RUN_ID and GITHUB_RUN_NUMBER environment variables must be set"
)
sys.exit(1)
# Use traces directory
traces_dir = args.traces_dir
print(f"Processing traces from directory: {traces_dir}")
# Publish traces
publish_traces(traces_dir, run_id, run_number)
if __name__ == "__main__":
main()
scripts/ci/utils/runner_utilization_report.py
Executable
+309
View File
@@ -0,0 +1,309 @@
#!/usr/bin/env python3
"""
Runner Utilization Report
Analyzes GitHub Actions job data to calculate runner utilization metrics.
Reports idle time, active time, and utilization percentage per runner label.
"""
import argparse
import json
import os
import subprocess
from collections import defaultdict
from datetime import datetime, timedelta, timezone
# Labels to skip when grouping runners (GitHub default labels)
DEFAULT_LABELS_TO_IGNORE = {"self-hosted", "Linux", "X64", "ARM64"}
GITHUB_HOSTED_LABELS = {"ubuntu-latest", "ubuntu-22.04", "ubuntu-24.04"}
def run_gh_command(args: list[str]) -> dict:
"""Run gh CLI command and return JSON result."""
result = subprocess.run(
["gh", "api"] + args,
capture_output=True,
text=True,
)
if result.returncode != 0:
raise Exception(f"gh api failed: {result.stderr}")
return json.loads(result.stdout)
def get_workflow_runs(repo: str, hours: int = 24) -> list[dict]:
"""Get workflow runs from the last N hours."""
since = datetime.now(timezone.utc) - timedelta(hours=hours)
runs = []
page = 1
while True:
data = run_gh_command(
[
f"repos/{repo}/actions/runs?per_page=100&page={page}",
]
)
page_runs = data.get("workflow_runs", [])
# Filter by time
for run in page_runs:
created_at = parse_time(run.get("created_at"))
if created_at and created_at >= since:
runs.append(run)
elif created_at and created_at < since:
# Runs are ordered by created_at desc, so we can stop
return runs
if len(page_runs) < 100:
break
page += 1
if page > 20: # Safety limit
break
return runs
def get_jobs_for_run(repo: str, run_id: int) -> list[dict]:
"""Get all jobs for a workflow run."""
jobs = []
page = 1
while True:
data = run_gh_command(
[
f"repos/{repo}/actions/runs/{run_id}/jobs?per_page=100&page={page}",
]
)
jobs.extend(data.get("jobs", []))
if len(data.get("jobs", [])) < 100:
break
page += 1
if page > 5: # Safety limit
break
return jobs
def get_runners(repo: str, online_only: bool = True) -> list[dict]:
"""Get all self-hosted runners with pagination. Returns empty if no permission."""
try:
all_runners = []
page = 1
while True:
data = run_gh_command(
[f"repos/{repo}/actions/runners?per_page=100&page={page}"]
)
runners = data.get("runners", [])
all_runners.extend(runners)
if len(runners) < 100:
break
page += 1
if page > 10: # Safety limit
break
if online_only:
all_runners = [r for r in all_runners if r.get("status") == "online"]
return all_runners
except Exception as e:
print(f"Warning: Cannot access runners API (need admin): {e}")
return []
def parse_time(time_str: str) -> datetime:
"""Parse ISO timestamp to datetime."""
if not time_str:
return None
return datetime.fromisoformat(time_str.replace("Z", "+00:00"))
# Known runner counts per label (fallback when API unavailable)
KNOWN_RUNNER_COUNTS = {
"1-gpu-5090": 16,
"h200": 8,
"h20": 4,
"b200": 4,
"amd": 8,
"github-hosted": 20, # GitHub hosted runners (variable)
"other": 10,
}
def calculate_utilization(repo: str, hours: int = 24, runner_filter: str = None):
"""Calculate runner utilization metrics."""
print(f"Fetching workflow runs from last {hours} hours...")
runs = get_workflow_runs(repo, hours)
print(f"Found {len(runs)} workflow runs")
# Try to get online runners from API
print("Fetching online runners...")
runners = get_runners(repo, online_only=True)
# Build label -> set of online runner names from API
api_label_runners = defaultdict(set)
if runners:
for runner in runners:
for label in runner.get("labels", []):
label_name = label.get("name", "")
if label_name not in DEFAULT_LABELS_TO_IGNORE:
api_label_runners[label_name].add(runner["name"])
print(f"Got {len(runners)} online runners from API")
else:
print("No runner API access, will use observed runners from job data")
# Track runners seen in jobs (for labels not in API or when API unavailable)
job_label_runners = defaultdict(set)
label_jobs = defaultdict(list) # label -> list of job_info
total_runs = len(runs)
for i, run in enumerate(runs):
if (i + 1) % 50 == 0:
print(f"Processing run {i+1}/{total_runs}...")
try:
jobs = get_jobs_for_run(repo, run["id"])
except Exception:
continue
for job in jobs:
runner_name = job.get("runner_name")
if not runner_name:
continue
created_at = parse_time(job.get("created_at"))
started_at = parse_time(job.get("started_at"))
completed_at = parse_time(job.get("completed_at"))
if not started_at or not completed_at:
continue
duration = (completed_at - started_at).total_seconds()
queue_time = (started_at - created_at).total_seconds() if created_at else 0
job_info = {
"start": started_at,
"end": completed_at,
"duration": duration,
"queue_time": queue_time,
"job_name": job["name"],
"runner_name": runner_name,
}
# Use job labels directly (available in job data)
job_labels = job.get("labels", [])
for label in job_labels:
# Skip generic labels
if label in DEFAULT_LABELS_TO_IGNORE | GITHUB_HOSTED_LABELS:
continue
job_label_runners[label].add(runner_name)
label_jobs[label].append(job_info)
# Merge API runners and job-observed runners
# Prefer API count (online runners) when available
all_labels = set(api_label_runners.keys()) | set(job_label_runners.keys())
# Filter labels if specified
if runner_filter:
all_labels = {lbl for lbl in all_labels if runner_filter in lbl}
print(f"Tracking {len(all_labels)} runner labels: {sorted(all_labels)}")
# Calculate metrics per label
window_seconds = hours * 3600
results = []
for label in sorted(all_labels):
# Use API runner count if available, otherwise use job-observed count
if label in api_label_runners and api_label_runners[label]:
num_runners = len(api_label_runners[label])
elif label in job_label_runners:
num_runners = len(job_label_runners[label])
else:
num_runners = KNOWN_RUNNER_COUNTS.get(label, 1)
total_capacity_seconds = window_seconds * num_runners
jobs = label_jobs.get(label, [])
total_active_seconds = sum(j["duration"] for j in jobs)
utilization = (
(total_active_seconds / total_capacity_seconds * 100)
if total_capacity_seconds > 0
else 0
)
idle_seconds = total_capacity_seconds - total_active_seconds
# Calculate queue time metrics
queue_times = [j["queue_time"] for j in jobs if j["queue_time"] > 0]
avg_queue_time = sum(queue_times) / len(queue_times) if queue_times else 0
max_queue_time = max(queue_times) if queue_times else 0
results.append(
{
"label": label,
"num_runners": num_runners,
"num_jobs": len(jobs),
"total_active_hours": total_active_seconds / 3600,
"total_idle_hours": idle_seconds / 3600,
"total_capacity_hours": total_capacity_seconds / 3600,
"utilization_pct": utilization,
"avg_queue_min": avg_queue_time / 60,
"max_queue_min": max_queue_time / 60,
}
)
return results
def format_report(results: list[dict], hours: int) -> str:
"""Format results as markdown report."""
lines = [
"# Runner Utilization Report",
"",
f"**Time window:** Last {hours} hours",
f"**Generated:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}",
"",
"## Summary by Runner Label",
"",
"| Label | Runners | Jobs | Active (hrs) | Utilization | Avg Queue | Max Queue |",
"|-------|---------|------|--------------|-------------|-----------|-----------|",
]
for r in results:
utilization_bar = "█" * int(r["utilization_pct"] / 10) + "░" * (
10 - int(r["utilization_pct"] / 10)
)
lines.append(
f"| {r['label']} | {r['num_runners']} | {r['num_jobs']} | "
f"{r['total_active_hours']:.1f} | "
f"{r['utilization_pct']:.1f}% {utilization_bar} | "
f"{r['avg_queue_min']:.1f}m | {r['max_queue_min']:.1f}m |"
)
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="Generate runner utilization report")
parser.add_argument("--repo", default="sgl-project/sglang", help="GitHub repo")
parser.add_argument("--hours", type=int, default=24, help="Time window in hours")
parser.add_argument(
"--filter", type=str, help="Filter runner labels (e.g., '5090', 'h200')"
)
parser.add_argument("--output", type=str, help="Output file (default: stdout)")
args = parser.parse_args()
results = calculate_utilization(args.repo, args.hours, args.filter)
report = format_report(results, args.hours)
if args.output:
with open(args.output, "w") as f:
f.write(report)
print(f"Report written to {args.output}")
else:
print(report)
# Also write to GITHUB_STEP_SUMMARY if available
summary_file = os.environ.get("GITHUB_STEP_SUMMARY")
if summary_file:
with open(summary_file, "a") as f:
f.write(report)
if __name__ == "__main__":
main()
scripts/ci/utils/slash_command_handler.py
+462
View File
@@ -0,0 +1,462 @@
import json
import os
import sys
import time
from datetime import datetime, timezone
import requests
from github import Auth, Github
# Configuration
PERMISSIONS_FILE_PATH = ".github/CI_PERMISSIONS.json"
def find_workflow_run_url(
gh_repo,
workflow_id,
ref,
target_stage,
token,
dispatch_time,
pr_head_sha=None,
max_wait=30,
):
"""
Poll for the workflow run URL after dispatch.
Uses the dynamic run-name feature to identify runs:
- Fork PRs: display_title = "[stage-name] sha"
- Non-fork PRs: display_title = "[stage-name]"
Args:
gh_repo: PyGithub repository object
workflow_id: ID of the workflow that was dispatched
ref: Branch/ref the workflow was dispatched on
target_stage: The stage name we're looking for
token: GitHub API token
dispatch_time: Unix timestamp when dispatch was triggered
pr_head_sha: PR head SHA (for fork PRs, used to match display_title)
max_wait: Maximum seconds to wait for the run to appear
Returns:
The workflow run URL if found, None otherwise.
"""
# Build expected display_title pattern based on workflow's run-name
# Format: "[stage-name] sha" for fork PRs, "[stage-name]" for non-fork
if pr_head_sha:
expected_title = f"[{target_stage}] {pr_head_sha}"
else:
expected_title = f"[{target_stage}]"
print(f"Looking for workflow run with display_title: {expected_title}")
for attempt in range(max_wait // 5):
time.sleep(5)
# Get recent workflow_dispatch runs for this workflow
runs_url = f"https://api.github.com/repos/{gh_repo.full_name}/actions/workflows/{workflow_id}/runs"
runs_resp = requests.get(
runs_url,
params={"event": "workflow_dispatch", "branch": ref, "per_page": 10},
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
},
)
if runs_resp.status_code != 200:
print(f"Failed to fetch workflow runs: {runs_resp.status_code}")
continue
for run in runs_resp.json().get("workflow_runs", []):
# Skip runs created before our dispatch (with 10s tolerance)
run_created = datetime.fromisoformat(
run["created_at"].replace("Z", "+00:00")
).timestamp()
if run_created < dispatch_time - 10:
continue
# Match by display_title (set by workflow's run-name directive)
# This is immediately available, unlike job names which require waiting
display_title = run.get("display_title", "")
if display_title == expected_title:
print(
f"Found matching workflow run: {run['id']} with title '{display_title}'"
)
return run["html_url"]
print(f"Could not find workflow run after {max_wait} seconds")
return None
def get_env_var(name):
val = os.getenv(name)
if not val:
print(f"Error: Environment variable {name} not set.")
sys.exit(1)
return val
def load_permissions(user_login):
"""
Reads the permissions JSON from the local file system and returns
the permissions dict for the specific user.
"""
try:
print(f"Loading permissions from {PERMISSIONS_FILE_PATH}...")
if not os.path.exists(PERMISSIONS_FILE_PATH):
print(f"Error: Permissions file not found at {PERMISSIONS_FILE_PATH}")
return None
with open(PERMISSIONS_FILE_PATH, "r") as f:
data = json.load(f)
user_perms = data.get(user_login)
if not user_perms:
print(f"User '{user_login}' not found in permissions file.")
return None
return user_perms
except Exception as e:
print(f"Failed to load or parse permissions file: {e}")
sys.exit(1)
def handle_tag_run_ci(gh_repo, pr, comment, user_perms, react_on_success=True):
"""
Handles the /tag-run-ci-label command.
Returns True if action was taken, False otherwise.
"""
if not user_perms.get("can_tag_run_ci_label", False):
print("Permission denied: can_tag_run_ci_label is false.")
return False
print("Permission granted. Adding 'run-ci' label.")
pr.add_to_labels("run-ci")
if react_on_success:
comment.create_reaction("+1")
print("Label added and comment reacted.")
else:
print("Label added (reaction suppressed).")
return True
def handle_rerun_failed_ci(gh_repo, pr, comment, user_perms, react_on_success=True):
"""
Handles the /rerun-failed-ci command.
Reruns workflows with 'failure' or 'skipped' conclusions.
Returns True if action was taken, False otherwise.
"""
if not user_perms.get("can_rerun_failed_ci", False):
print("Permission denied: can_rerun_failed_ci is false.")
return False
print("Permission granted. Triggering rerun of failed or skipped workflows.")
# Get the SHA of the latest commit in the PR
head_sha = pr.head.sha
print(f"Checking workflows for commit: {head_sha}")
# List all workflow runs for this commit
runs = gh_repo.get_workflow_runs(head_sha=head_sha)
rerun_count = 0
for run in runs:
if run.status != "completed":
continue
if run.conclusion == "failure":
# DEBUG
print(f"Rerunning failed workflow: {run.name} (ID: {run.id})")
try:
# Use rerun_failed_jobs for efficiency on failures
run.rerun_failed_jobs()
rerun_count += 1
except Exception as e:
print(f"Failed to rerun workflow {run.id}: {e}")
elif run.conclusion == "skipped":
print(f"Rerunning skipped workflow: {run.name} (ID: {run.id})")
try:
# Skipped workflows don't have 'failed jobs', so we use full rerun()
run.rerun()
rerun_count += 1
except Exception as e:
print(f"Failed to rerun workflow {run.id}: {e}")
if rerun_count > 0:
print(f"Triggered rerun for {rerun_count} workflows.")
if react_on_success:
comment.create_reaction("+1")
return True
else:
print("No failed or skipped workflows found to rerun.")
return False
def handle_rerun_stage(
gh_repo, pr, comment, user_perms, stage_name, token, react_on_success=True
):
"""
Handles the /rerun-stage <stage-name> command.
Triggers a workflow_dispatch to run only the specified stage, skipping dependencies.
Returns True if action was taken, False otherwise.
"""
if not user_perms.get("can_rerun_stage", False):
print("Permission denied: can_rerun_stage is false.")
return False
if not stage_name:
print("Error: No stage name provided")
comment.create_reaction("confused")
pr.create_issue_comment(
f"❌ Please specify a stage name: `/rerun-stage <stage-name>`\n\n"
f"Examples: `/rerun-stage unit-test-backend-4-gpu`, `/rerun-stage accuracy-test-1-gpu`"
)
return False
print(f"Permission granted. Triggering workflow_dispatch for stage '{stage_name}'.")
# Valid NVIDIA stage names that support target_stage
nvidia_stages = [
"stage-a-test-1",
"stage-a-cpu-only",
"stage-b-test-small-1-gpu",
"stage-b-test-large-1-gpu",
"stage-b-test-large-2-gpu",
"stage-c-test-large-4-gpu",
"stage-c-test-large-4-gpu-b200",
"multimodal-gen-test-1-gpu",
"multimodal-gen-test-2-gpu",
"quantization-test",
"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-8-gpu-b200",
"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-4-gpu-b200",
"unit-test-backend-4-gpu-gb200",
]
# Valid AMD stage names that support target_stage
amd_stages = [
"sgl-kernel-unit-test-amd",
"stage-a-test-1-amd",
"stage-b-test-small-1-gpu-amd",
"stage-b-test-small-1-gpu-amd-mi35x",
"stage-b-test-large-2-gpu-amd",
"stage-b-test-small-1-gpu-performance-amd",
"stage-b-test-large-1-gpu-performance-amd",
"stage-b-test-large-2-gpu-performance-amd",
"stage-c-test-large-8-gpu-amd-mi35x",
"unit-test-backend-1-gpu-amd",
"unit-test-backend-2-gpu-amd",
"unit-test-backend-8-gpu-amd",
"accuracy-test-1-gpu-amd",
"accuracy-test-2-gpu-amd",
]
valid_stages = nvidia_stages + amd_stages
is_amd_stage = stage_name in amd_stages
if stage_name not in valid_stages:
comment.create_reaction("confused")
pr.create_issue_comment(
f"❌ Stage `{stage_name}` doesn't support isolated runs yet.\n\n"
f"**NVIDIA stages:**\n"
+ "\n".join(f"- `{s}`" for s in nvidia_stages)
+ "\n\n**AMD stages:**\n"
+ "\n".join(f"- `{s}`" for s in amd_stages)
+ "\n\nOther stages will be added soon. For now, use `/rerun-failed-ci` for those stages."
)
return False
try:
# Get the appropriate workflow based on stage type
workflow_name = "PR Test (AMD)" if is_amd_stage else "PR Test"
workflows = gh_repo.get_workflows()
target_workflow = None
for wf in workflows:
if wf.name == workflow_name:
target_workflow = wf
break
if not target_workflow:
print(f"Error: {workflow_name} workflow not found")
return False
# Check if PR is from a fork by comparing repo owners
# Handle case where fork repo may have been deleted (pr.head.repo is None)
is_fork = (
pr.head.repo is None or pr.head.repo.owner.login != gh_repo.owner.login
)
print(f"PR is from fork: {is_fork}")
# pr_head_sha is used for fork PRs (passed to workflow and used for URL lookup)
pr_head_sha = None
if is_fork:
# For fork PRs: dispatch on main and pass SHA as input
# This is needed because fork branch names don't exist in the main repo
ref = "main"
pr_head_sha = pr.head.sha
print(
f"Triggering {workflow_name} workflow on ref: {ref}, PR head SHA: {pr_head_sha}"
)
if is_amd_stage:
inputs = {"target_stage": stage_name, "pr_head_sha": pr_head_sha}
else:
inputs = {
"version": "release",
"target_stage": stage_name,
"pr_head_sha": pr_head_sha,
}
else:
# For non-fork PRs: dispatch on the PR branch directly
# This allows testing workflow changes before merge
ref = pr.head.ref
print(f"Triggering {workflow_name} workflow on branch: {ref}")
if is_amd_stage:
inputs = {"target_stage": stage_name}
else:
inputs = {"version": "release", "target_stage": stage_name}
# Record dispatch time before triggering
dispatch_time = time.time()
# Use requests directly as PyGithub's create_dispatch only accepts HTTP 204
dispatch_url = f"https://api.github.com/repos/{gh_repo.full_name}/actions/workflows/{target_workflow.id}/dispatches"
dispatch_resp = requests.post(
dispatch_url,
json={"ref": ref, "inputs": inputs},
headers={
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
},
)
success = dispatch_resp.status_code in (200, 204)
if not success:
print(f"Dispatch failed: {dispatch_resp.status_code} {dispatch_resp.text}")
if success:
print(f"Successfully triggered workflow for stage '{stage_name}'")
if react_on_success:
comment.create_reaction("+1")
pr.create_issue_comment(
f"✅ Triggered `{stage_name}` to run independently (skipping dependencies)."
)
# Poll for the workflow run URL and post follow-up comment
run_url = find_workflow_run_url(
gh_repo,
target_workflow.id,
ref,
stage_name,
token,
dispatch_time,
pr_head_sha=pr_head_sha,
max_wait=30,
)
if run_url:
pr.create_issue_comment(f"🔗 [View workflow run]({run_url})")
else:
pr.create_issue_comment(
f"⚠️ Could not retrieve workflow run URL. "
f"Check the [Actions tab](https://github.com/{gh_repo.full_name}/actions) for progress."
)
return True
else:
print("Failed to trigger workflow_dispatch")
return False
except Exception as e:
print(f"Error triggering workflow_dispatch: {e}")
comment.create_reaction("confused")
pr.create_issue_comment(
f"❌ Failed to trigger workflow: {str(e)}\n\n"
f"Please check the logs or contact maintainers."
)
return False
def main():
# 1. Load Environment Variables
token = get_env_var("GITHUB_TOKEN")
repo_name = get_env_var("REPO_FULL_NAME")
pr_number = int(get_env_var("PR_NUMBER"))
comment_id = int(get_env_var("COMMENT_ID"))
comment_body = get_env_var("COMMENT_BODY").strip()
user_login = get_env_var("USER_LOGIN")
# 2. Load Permissions (Local Check)
user_perms = load_permissions(user_login)
if not user_perms:
print(f"User {user_login} does not have any configured permissions. Exiting.")
return
# 3. Initialize GitHub API with Auth
auth = Auth.Token(token)
g = Github(auth=auth)
repo = g.get_repo(repo_name)
pr = repo.get_pull(pr_number)
comment = repo.get_issue(pr_number).get_comment(comment_id)
# 4. Parse Command and Execute
first_line = comment_body.split("\n")[0].strip()
if first_line.startswith("/tag-run-ci-label"):
handle_tag_run_ci(repo, pr, comment, user_perms)
elif first_line.startswith("/rerun-failed-ci"):
handle_rerun_failed_ci(repo, pr, comment, user_perms)
elif first_line.startswith("/tag-and-rerun-ci"):
# Perform both actions, but suppress individual reactions
print("Processing combined command: /tag-and-rerun-ci")
tagged = handle_tag_run_ci(
repo, pr, comment, user_perms, react_on_success=False
)
# Wait for the label to propagate before triggering rerun
if tagged:
print("Waiting 5 seconds for label to propagate...")
time.sleep(5)
rerun = handle_rerun_failed_ci(
repo, pr, comment, user_perms, react_on_success=False
)
# If at least one action was successful, add the reaction here
if tagged or rerun:
comment.create_reaction("+1")
print("Combined command processed successfully; reaction added.")
else:
print("Combined command finished, but no actions were taken.")
elif first_line.startswith("/rerun-stage"):
# Extract stage name from command
parts = first_line.split(maxsplit=1)
stage_name = parts[1].strip() if len(parts) > 1 else None
handle_rerun_stage(repo, pr, comment, user_perms, stage_name, token)
else:
print(f"Unknown or ignored command: {first_line}")
if __name__ == "__main__":
main()