diff --git a/projects/rhaiis/README.md b/projects/rhaiis/README.md index 716cdea4f..e6584e1a9 100644 --- a/projects/rhaiis/README.md +++ b/projects/rhaiis/README.md @@ -281,7 +281,9 @@ Available configOverrides: | `rhaiis.engines.sglang.args.*` | SGLang CLI args (e.g. `tp-size`, `mem-fraction-static`, `context-length`) | | `rhaiis.engines.trtllm.args.*` | TRT-LLM CLI args (e.g. `tp_size`, `ep_size`, `max_batch_size`) | | `rhaiis.engines.trtllm.trtllm_config.*` | TRT-LLM server config (kv_cache, cuda_graph, moe) | -| `rhaiis.profiler.enabled` | Enable PyTorch profiler | +| `rhaiis.profiler.enabled` | Enable profiler Phase 1 | +| `rhaiis.profiler.backend` | `webhook` (default) or `native` (HTTP start/stop) | +| `rhaiis.profiler.kind` | Native kind: `torch`, `cuda`, or `proton` | | `rhaiis.agent_analysis.enabled` | Enable AI agent regression analysis | | `caliper.postprocess.csv_dashboard.enabled` | Enable dashboard CSV S3 sync | | `benchmarks.guidellm.timeout` | Benchmark timeout in seconds | @@ -322,6 +324,7 @@ oc patch fournosjob -n psap-automation \ | `enable_profiler_gate` | [rhaiis](./toolbox/enable_profiler_gate/) | Enable/disable the PyTorch profiler gate file on the vLLM pod | | `verify_profiler_prereqs` | [rhaiis](./toolbox/verify_profiler_prereqs/) | Verify profiler prerequisites (gate file, sitecustomize.py) | | `copy_profiler_traces` | [rhaiis](./toolbox/copy_profiler_traces/) | Copy Chrome trace JSON files from the vLLM pod | +| `control_native_profiler` | [rhaiis](./toolbox/control_native_profiler/) | POST `/start_profile` and `/stop_profile` on the predictor | ## Usage @@ -337,6 +340,16 @@ python3 -m projects.rhaiis.orchestration.cli test \ python3 -m projects.rhaiis.orchestration.cli test \ --model llama-4-scout-fp8 --workload profile2 --dry-run +# Native vLLM torch profiler (HTTP start/stop) +python3 -m projects.rhaiis.orchestration.cli test \ + --preset llama-8b --preset profile1 --preset profiler-native-short \ + --namespace kserve-e2e-perf --dry-run + +# SGLang native profiler (10 engine steps after 5 warmup steps) +python3 -m projects.rhaiis.orchestration.cli test \ + --preset llama-8b --preset profile1 --preset profiler-native-sglang \ + --namespace kserve-e2e-perf --dry-run + # Full E2E test python3 -m projects.rhaiis.orchestration.cli test \ --model qwen3-0_6b \ @@ -440,17 +453,52 @@ step also detects infrastructure failures that occur before the test step runs. ## PyTorch profiling -When `rhaiis.profiler.enabled: true`, the pipeline runs profiler-gated benchmarks -before the main benchmarks: +Set `rhaiis.profiler.enabled: true`. Phase 1 still copies traces and uploads them to S3 the same way; only **how capture is started** differs. + +### Backend: `webhook` (default) + +vLLM-only. Labels the InferenceService `vllm-profiler/enabled=true` so the cluster webhook injects `sitecustomize.py`. Then: + +1. Verify webhook prerequisites +2. Write `/tmp/profiler_gate` with a workload label (e.g. `isl1000_osl1000`) +3. Run GuideLLM at `rhaiis.profiler.rates` / `max_seconds` (default 200 concurrent for 200s) +4. Remove the gate file +5. Copy `/tmp/trace_*.json*` and upload rank-0 traces to S3 + +The webhook records configured `execute_model` ranges (typically 500–503). GuideLLM does **not** stop when that range ends; it always runs for `max_seconds`. + +### Backend: `native` + +vLLM 0.13+ and SGLang. No webhook. The server is started with engine profiler support, then Forge calls HTTP start/stop around the same short GuideLLM load: -1. Verify profiler prerequisites on the vLLM pod (gate file, sitecustomize.py) -2. Enable the profiler gate with a workload-specific label (e.g. `isl1000_osl1000`) -3. Run a GuideLLM benchmark at the configured profiler rates -4. Disable the profiler gate -5. Copy Chrome trace JSON files from the vLLM pod -6. Upload traces to S3 organized by accelerator/model/TP/version/profile +```bash +--preset profiler-native # torch, default profiler rates/duration +--preset profiler-native-short # torch, concurrency 1 for 60s +--preset profiler-native-window # skip 50 engine steps, record 10 +--preset profiler-native-cuda # Nsight (nsys in the serving image) +--preset profiler-native-proton # Triton Proton chrome_trace +--preset profiler-native-sglang # SGLang engine + /start_profile +``` + +1. Deploy with `--profiler-config` (vLLM) or `SGLANG_TORCH_PROFILER_DIR` (SGLang) +2. `POST /start_profile` +3. Run GuideLLM at profiler rates / max_seconds +4. `POST /stop_profile` (flush can take many minutes; `native.stop_timeout_seconds` default 1800) +5. Copy `rhaiis.profiler.traces_dir` (default `/tmp/vllm_profile`) and upload to S3 + +vLLM native knobs (under `rhaiis.profiler.native`): `delay_iterations`, `max_iterations`, `wait_iterations` / `warmup_iterations` / `active_iterations` (torch.profiler schedule), `with_stack`, `record_shapes`, `with_memory`, `ignore_frontend`. + +SGLang uses the same HTTP API with `num_steps` / `start_step` / `activities` in the start body. -Traces are viewable in `chrome://tracing` or Perfetto UI. +### Native `kind` + +| kind | What it is | +|------|------------| +| `torch` (default) | PyTorch profiler → Chrome/Perfetto `.json.gz` under `torch_profiler_dir` | +| `cuda` | CUDA Profiler API (`cudaProfilerStart/Stop`). Pair with Nsight: set `native.nsys_wrap: true` (serving image must contain `nsys`) | +| `proton` | Triton Proton (CUPTI). Needs `--enforce-eager`. Output `hatchet` tree or `chrome_trace` via `proton_output_format` | + +Traces are viewable in `chrome://tracing` or Perfetto UI (Proton hatchet trees use `proton-viewer`). ## Parallel job isolation @@ -510,6 +558,9 @@ Full list: `grep "^[a-z]" orchestration/config.d/models.yaml` | `profile2` | 512 (stdev 128) | 2048 (stdev 512) | 1, 50, 100, 200, 300 | 450 | | `profile3` | 2048 | 128 | 1, 50, 100, 200, 300 | 450 | | `profile4` | 8000 | 1000 | 1, 25, 50, 75, 100 | 450 | +| `profile5` | 100000 | 1000 | 1, 2, 5 | 450 | +| `profile6` | 1000 (5 turns, prefix 512) | 1000 | 1, 25, 50, 75, 100 | 450 | +| `profile7` | 8000 (stdev 8500, 50–30000) | 800 (stdev 1500) | 1, 50, 100, 200, 300 | 450 | ## Presets @@ -521,10 +572,18 @@ python3 -m projects.rhaiis.orchestration.cli test \ --preset llama-8b --preset profile1 \ --namespace kserve-e2e-perf +# Native torch profiler, short capture +python3 -m projects.rhaiis.orchestration.cli test \ + --preset llama-8b --preset profile1 --preset profiler-native-short \ + --namespace kserve-e2e-perf + # Available model presets: llama-8b, llama-70b, llama-405b, llama-4-scout, # llama-4-maverick, granite-8b, mistral-24b, qwen25-7b, qwen3-235b, # deepseek-r1, deepseek-v3, gpt-oss -# Workload presets: profile1, profile2, profile3, profile4 +# Workload presets: profile1 … profile7 +# Profiler presets: profiler-webhook, profiler-native, profiler-native-short, +# profiler-native-window, profiler-native-cuda, profiler-native-proton, +# profiler-native-sglang # Accelerator presets: nvidia, amd ``` diff --git a/projects/rhaiis/orchestration/config.d/rhaiis.yaml b/projects/rhaiis/orchestration/config.d/rhaiis.yaml index 358f710b1..4c77147e8 100644 --- a/projects/rhaiis/orchestration/config.d/rhaiis.yaml +++ b/projects/rhaiis/orchestration/config.d/rhaiis.yaml @@ -80,12 +80,42 @@ warmup: profiler: enabled: false + # webhook: mutating injector + /tmp/profiler_gate (vLLM only) + # native: vLLM --profiler-config + POST /start_profile /stop_profile + # (SGLang uses the same HTTP API with SGLANG_TORCH_PROFILER_DIR) + backend: webhook + # native kind: torch (Chrome/Perfetto), cuda (Nsight CUDA profiler API), + # proton (Triton Proton; requires enforce-eager) + kind: torch + traces_dir: /tmp/vllm_profile s3_bucket: psap-dashboard-data s3_prefix: pytorch-profiles/rhaiis vault: psap-forge-dashboard-s3 labels: [] rates: [200] max_seconds: 200 + native: + delay_iterations: 0 + max_iterations: 0 + wait_iterations: 0 + warmup_iterations: 0 + active_iterations: 5 + ignore_frontend: true + record_shapes: false + with_stack: true + with_memory: false + with_flops: false + use_gzip: true + dump_cuda_time_total: true + stop_timeout_seconds: 1800 + nsys_wrap: false + proton_output_format: chrome_trace + proton_hook: triton + num_steps: 0 + start_step: 0 + activities: + - CPU + - GPU agent_analysis: enabled: false diff --git a/projects/rhaiis/orchestration/manifests.py b/projects/rhaiis/orchestration/manifests.py index caa96c821..a2d7446d7 100644 --- a/projects/rhaiis/orchestration/manifests.py +++ b/projects/rhaiis/orchestration/manifests.py @@ -219,6 +219,10 @@ def _build_vllm_sglang_container( gpu_count: int, env_vars_list: list[dict], ) -> dict[str, Any]: + raw_args = dict(engine_args or {}) + nsys_wrap = bool(raw_args.pop("_nsys_wrap", False)) + nsys_output = str(raw_args.pop("_nsys_output", "/tmp/vllm_profile/nsys_capture")) + if engine == "sglang": command = ["sglang", "serve"] if storage_source == "hf": @@ -226,13 +230,28 @@ def _build_vllm_sglang_container( else: args = ["--model-path=/mnt/models", "--port=8080", "--host=0.0.0.0"] else: - command = ["python3", "-m", "vllm.entrypoints.openai.api_server"] + vllm_cmd = ["python3", "-m", "vllm.entrypoints.openai.api_server"] + if nsys_wrap: + command = [ + "nsys", + "profile", + "--trace-fork-before-exec=true", + "--cuda-graph-trace=node", + "--capture-range=cudaProfilerApi", + "--capture-range-end=repeat", + f"--output={nsys_output}", + *vllm_cmd, + ] + else: + command = vllm_cmd if storage_source == "hf": args = [f"--model={model_id}", "--port=8080"] else: args = ["--model=/mnt/models", f"--served-model-name={model_id}", "--port=8080"] - for key, val in (engine_args or {}).items(): + for key, val in raw_args.items(): + if str(key).startswith("_"): + continue if isinstance(val, bool): if val: args.append(f"--{key}") diff --git a/projects/rhaiis/orchestration/presets.d/benchmarks.yaml b/projects/rhaiis/orchestration/presets.d/benchmarks.yaml index fdb3e2070..38207f912 100644 --- a/projects/rhaiis/orchestration/presets.d/benchmarks.yaml +++ b/projects/rhaiis/orchestration/presets.d/benchmarks.yaml @@ -6,3 +6,67 @@ benchmark: rhaiis.profiler.enabled: true caliper.postprocess.csv_dashboard.enabled: true rhaiis.agent_analysis.enabled: false + +# Cluster webhook + /tmp/profiler_gate (vLLM only). This is also the +# default when rhaiis.profiler.backend is unset. +profiler-webhook: + rhaiis.profiler.enabled: true + rhaiis.profiler.backend: webhook + +# vLLM 0.13+ HTTP /start_profile + /stop_profile (torch Chrome traces). +# Example: +# --preset llama-8b --preset profile1 --preset profiler-native +profiler-native: + rhaiis.profiler.enabled: true + rhaiis.profiler.backend: native + rhaiis.profiler.kind: torch + +# Short native capture: one concurrent stream, 60s GuideLLM window. +# Prefer this over profiler-native when you only need a trace, not a 200s soak. +profiler-native-short: + rhaiis.profiler.enabled: true + rhaiis.profiler.backend: native + rhaiis.profiler.kind: torch + rhaiis.profiler.rates: [1] + rhaiis.profiler.max_seconds: 60 + +# Skip 50 engine iterations, then record 10. GuideLLM still runs for max_seconds; +# the engine stops recording after max_iterations. +profiler-native-window: + rhaiis.profiler.enabled: true + rhaiis.profiler.backend: native + rhaiis.profiler.kind: torch + rhaiis.profiler.rates: [1] + rhaiis.profiler.max_seconds: 120 + rhaiis.profiler.native.delay_iterations: 50 + rhaiis.profiler.native.max_iterations: 10 + rhaiis.profiler.native.ignore_frontend: true + +# Nsight Systems via CUDA profiler API. The serving image must contain `nsys`. +profiler-native-cuda: + rhaiis.profiler.enabled: true + rhaiis.profiler.backend: native + rhaiis.profiler.kind: cuda + rhaiis.profiler.native.nsys_wrap: true + rhaiis.profiler.rates: [1] + rhaiis.profiler.max_seconds: 60 + +# Triton Proton (eager mode). Open chrome_trace in Perfetto; hatchet needs proton-viewer. +profiler-native-proton: + rhaiis.profiler.enabled: true + rhaiis.profiler.backend: native + rhaiis.profiler.kind: proton + rhaiis.profiler.native.proton_output_format: chrome_trace + rhaiis.profiler.rates: [1] + rhaiis.profiler.max_seconds: 60 + +# SGLang native profiler (SGLANG_TORCH_PROFILER_DIR + /start_profile). +# num_steps auto-stops the engine profiler; /stop_profile is still attempted. +profiler-native-sglang: + rhaiis.engine: sglang + rhaiis.profiler.enabled: true + rhaiis.profiler.backend: native + rhaiis.profiler.rates: [1] + rhaiis.profiler.max_seconds: 60 + rhaiis.profiler.native.num_steps: 10 + rhaiis.profiler.native.start_step: 5 diff --git a/projects/rhaiis/orchestration/profiler.py b/projects/rhaiis/orchestration/profiler.py new file mode 100644 index 000000000..5a1284d3d --- /dev/null +++ b/projects/rhaiis/orchestration/profiler.py @@ -0,0 +1,143 @@ +"""Native (HTTP start/stop) profiler helpers for rhaiis. + +Kept in orchestration so toolbox commands stay free of config imports. +The webhook backend is unchanged; this module only prepares deploy-time +args/env and the /start_profile JSON body for the native path. +""" + +from __future__ import annotations + +import json +import logging + +logger = logging.getLogger(__name__) + +NATIVE_ENGINES = ("vllm", "sglang") +DEFAULT_TRACES_DIR = "/tmp/vllm_profile" +DEFAULT_RPC_TIMEOUT_MS = "1800000" + + +def profiler_backend(profiler_cfg: dict) -> str: + return str(profiler_cfg.get("backend", "webhook") or "webhook").strip().lower() + + +def profiler_kind(profiler_cfg: dict) -> str: + return str(profiler_cfg.get("kind", "torch") or "torch").strip().lower() + + +def traces_dir(profiler_cfg: dict) -> str: + return str(profiler_cfg.get("traces_dir") or DEFAULT_TRACES_DIR) + + +def native_options(profiler_cfg: dict) -> dict: + return dict(profiler_cfg.get("native") or {}) + + +def is_native_backend(profiler_cfg: dict) -> bool: + return profiler_backend(profiler_cfg) == "native" + + +def engine_supports_profiler(engine: str, profiler_cfg: dict) -> bool: + if is_native_backend(profiler_cfg): + return engine in NATIVE_ENGINES + return engine == "vllm" + + +def build_vllm_profiler_config_json(profiler_cfg: dict) -> str: + """JSON for vLLM ``--profiler-config`` (v0.13+).""" + kind = profiler_kind(profiler_cfg) + if kind not in ("torch", "cuda", "proton"): + raise ValueError(f"Unsupported rhaiis.profiler.kind: {kind}") + + native = native_options(profiler_cfg) + cfg: dict = {"profiler": kind} + directory = traces_dir(profiler_cfg) + + if kind == "torch": + cfg["torch_profiler_dir"] = directory + cfg["torch_profiler_with_stack"] = bool(native.get("with_stack", True)) + cfg["torch_profiler_record_shapes"] = bool(native.get("record_shapes", False)) + cfg["torch_profiler_with_memory"] = bool(native.get("with_memory", False)) + cfg["torch_profiler_with_flops"] = bool(native.get("with_flops", False)) + cfg["torch_profiler_use_gzip"] = bool(native.get("use_gzip", True)) + cfg["torch_profiler_dump_cuda_time_total"] = bool( + native.get("dump_cuda_time_total", True) + ) + cfg["ignore_frontend"] = bool(native.get("ignore_frontend", True)) + delay = int(native.get("delay_iterations", 0) or 0) + max_iters = int(native.get("max_iterations", 0) or 0) + if delay: + cfg["delay_iterations"] = delay + if max_iters: + cfg["max_iterations"] = max_iters + wait = int(native.get("wait_iterations", 0) or 0) + warmup = int(native.get("warmup_iterations", 0) or 0) + active = int(native.get("active_iterations", 5) or 5) + if wait or warmup: + cfg["wait_iterations"] = wait + cfg["warmup_iterations"] = warmup + cfg["active_iterations"] = active + elif kind == "proton": + output_format = str(native.get("proton_output_format") or "chrome_trace") + cfg["proton_profiler_dir"] = directory + cfg["proton_output_format"] = output_format + cfg["proton_data"] = "trace" if output_format == "chrome_trace" else "tree" + if native.get("proton_hook", "triton"): + cfg["proton_hook"] = native.get("proton_hook", "triton") + return json.dumps(cfg, separators=(",", ":")) + + +def build_sglang_start_body(profiler_cfg: dict) -> str: + """JSON body for SGLang ``POST /start_profile``.""" + native = native_options(profiler_cfg) + body: dict = {"output_dir": traces_dir(profiler_cfg)} + num_steps = native.get("num_steps", native.get("max_iterations", 0)) + num_steps = int(num_steps or 0) + start_step = int(native.get("start_step", native.get("delay_iterations", 0)) or 0) + if num_steps > 0: + body["num_steps"] = num_steps + if start_step > 0: + body["start_step"] = start_step + activities = native.get("activities") or ["CPU", "GPU"] + if isinstance(activities, str): + activities = [a.strip() for a in activities.split(",") if a.strip()] + body["activities"] = activities + return json.dumps(body, separators=(",", ":")) + + +def apply_native_profiler_deploy( + engine: str, + engine_args: dict, + env_vars: dict, + profiler_cfg: dict, +) -> None: + """Mutate ServingRuntime args/env so native start/stop profiling works. + + vLLM: ``--profiler-config`` plus a long RPC timeout for ``/stop_profile`` flush. + SGLang: ``SGLANG_TORCH_PROFILER_DIR`` (no extra serve flag required). + """ + directory = traces_dir(profiler_cfg) + kind = profiler_kind(profiler_cfg) + native = native_options(profiler_cfg) + + if engine == "vllm": + engine_args["profiler-config"] = build_vllm_profiler_config_json(profiler_cfg) + env_vars.setdefault("VLLM_RPC_TIMEOUT", DEFAULT_RPC_TIMEOUT_MS) + env_vars.setdefault("VLLM_RPC_GET_DATA_TIMEOUT_MS", DEFAULT_RPC_TIMEOUT_MS) + if kind == "proton": + engine_args["enforce-eager"] = True + if kind == "cuda" and native.get("nsys_wrap", False): + engine_args["_nsys_wrap"] = True + engine_args["_nsys_output"] = f"{directory}/nsys_capture" + logger.info("CUDA profiler: wrapping vLLM process with nsys (image must include nsys)") + elif engine == "sglang": + env_vars["SGLANG_TORCH_PROFILER_DIR"] = directory + else: + raise ValueError(f"Native profiler does not support engine={engine}") + + logger.info( + "Native profiler deploy: engine=%s kind=%s traces_dir=%s", + engine, + kind, + directory, + ) diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 20c93af62..4f1c74029 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -11,6 +11,14 @@ from projects.core.library import env from projects.core.library.postprocess import run_and_postprocess, write_test_labels from projects.rhaiis.orchestration import runtime_config +from projects.rhaiis.orchestration.profiler import ( + apply_native_profiler_deploy, + build_sglang_start_body, + engine_supports_profiler, + is_native_backend, + native_options, + traces_dir as profiler_traces_dir, +) logger = logging.getLogger(__name__) @@ -170,6 +178,7 @@ def _run_test( profiler_cfg = runtime_config.get_profiler_config() profiler_enabled = profiler_cfg.get("enabled", False) + profiler_native = is_native_backend(profiler_cfg) run_benchmark = config.project.get_config("tests.rhaiis.run_benchmark", True) # Standalone analysis only — no deployment needed @@ -209,11 +218,17 @@ def _run_test( try: isvc_labels = {"opendatahub.io/dashboard": "true"} - if profiler_enabled and engine == "vllm": - isvc_labels["vllm-profiler/enabled"] = "true" - elif profiler_enabled and engine != "vllm": - logger.warning("Profiler is only supported with vLLM engine, skipping profiler") + if profiler_enabled and not engine_supports_profiler(engine, profiler_cfg): + logger.warning( + "Profiler backend=%s is not supported with engine=%s, skipping profiler", + "native" if profiler_native else "webhook", + engine, + ) profiler_enabled = False + elif profiler_enabled and profiler_native: + apply_native_profiler_deploy(engine, engine_args, env_vars, profiler_cfg) + elif profiler_enabled and engine == "vllm": + isvc_labels["vllm-profiler/enabled"] = "true" logger.info("Deploying %s to %s/%s", model_cfg["hf_model_id"], namespace, deployment_name) ea = engine_args or {} @@ -297,6 +312,7 @@ def _run_test( workload=runtime_config.get_workload(wl_key), workload_key=wl_key, benchmark_timeout=benchmark_timeout, + engine=engine, ) if profiler_enabled: logger.info("Running profiler for workload=%s", wl_key) @@ -714,8 +730,10 @@ def _run_warmup_step( workload: dict, workload_key: str, benchmark_timeout: int, + engine: str = "vllm", ) -> None: """Run a short warmup benchmark to prime KV cache and CUDA kernels.""" + _ = engine from projects.core.library import config from projects.guidellm.toolbox.run_guidellm_benchmark.main import ( run as run_guidellm_benchmark, @@ -761,11 +779,95 @@ def _run_profiler_step( workload: dict, workload_key: str, benchmark_timeout: int, + engine: str = "vllm", +) -> None: + """Phase-1 profiler capture, then copy traces. S3 upload stays in the caller.""" + profiler_cfg = runtime_config.get_profiler_config() + if is_native_backend(profiler_cfg): + _run_native_profiler_step( + deployment_name=deployment_name, + namespace=namespace, + endpoint_url=endpoint_url, + benchmark_cfg=benchmark_cfg, + model_cfg=model_cfg, + workload=workload, + workload_key=workload_key, + benchmark_timeout=benchmark_timeout, + engine=engine, + profiler_cfg=profiler_cfg, + ) + return + _run_webhook_profiler_step( + deployment_name=deployment_name, + namespace=namespace, + endpoint_url=endpoint_url, + benchmark_cfg=benchmark_cfg, + model_cfg=model_cfg, + workload=workload, + workload_key=workload_key, + benchmark_timeout=benchmark_timeout, + profiler_cfg=profiler_cfg, + ) + + +def _profiler_labels(profiler_cfg: dict, workload: dict) -> list[str]: + labels = profiler_cfg.get("labels", []) + if labels: + return list(labels) + label = _derive_profiler_label(workload) + logger.info("Auto-generated profiler label from workload: %s", label) + return [label] + + +def _run_guidellm_profiler_load( + *, + deployment_name: str, + namespace: str, + endpoint_url: str, + benchmark_cfg: dict, + model_cfg: dict, + workload: dict, + workload_key: str, + benchmark_timeout: int, + profiler_cfg: dict, ) -> None: - """Run profiler-gated benchmarks: verify prereqs → enable gate → benchmark → disable gate → copy traces.""" from projects.guidellm.toolbox.run_guidellm_benchmark.main import ( run as run_guidellm_benchmark, ) + + guidellm_args = runtime_config.build_guidellm_args( + benchmark_cfg=benchmark_cfg, + model_id=model_cfg["hf_model_id"], + data=workload["data"], + rates=profiler_cfg.get("rates", [1]), + max_seconds=profiler_cfg.get("max_seconds", 60), + ) + run_guidellm_benchmark( + endpoint_url=f"{endpoint_url}/v1", + name=_guidellm_job_name("guidellm-profiler", workload_key, deployment_name), + namespace=namespace, + image=benchmark_cfg.get("image", "ghcr.io/vllm-project/guidellm:v0.6.0"), + timeout=benchmark_timeout, + pvc_size=benchmark_cfg.get("pvc_size", "5Gi"), + guidellm_args=guidellm_args, + hf_token_secret=benchmark_cfg.get("hf_token_secret", ""), + fs_group=benchmark_cfg.get("fs_group"), + ) + + +def _run_webhook_profiler_step( + *, + deployment_name: str, + namespace: str, + endpoint_url: str, + benchmark_cfg: dict, + model_cfg: dict, + workload: dict, + workload_key: str, + benchmark_timeout: int, + profiler_cfg: dict, +) -> None: + """Webhook path: verify prereqs → gate file → GuideLLM → disable gate → copy /tmp/trace_*.""" from projects.rhaiis.toolbox.copy_profiler_traces.main import run as copy_profiler_traces from projects.rhaiis.toolbox.enable_profiler_gate.main import run as enable_profiler_gate from projects.rhaiis.toolbox.verify_profiler_prereqs.main import run as verify_profiler_prereqs @@ -773,44 +875,25 @@ def _run_profiler_step( logger.info("Verifying profiler prerequisites") verify_profiler_prereqs(namespace=namespace) - profiler_cfg = runtime_config.get_profiler_config() - labels = profiler_cfg.get("labels", []) - if not labels: - labels = [_derive_profiler_label(workload)] - logger.info("Auto-generated profiler label from workload: %s", labels[0]) - - profiler_max_seconds = profiler_cfg.get("max_seconds", 60) - + labels = _profiler_labels(profiler_cfg, workload) for label in labels: - logger.info("Profiling label=%s", label) - - gate_value = label if isinstance(label, str) else str(label) + logger.info("Profiling label=%s (webhook gate)", label) enable_profiler_gate( name=deployment_name, namespace=namespace, - gate_value=gate_value, + gate_value=label if isinstance(label, str) else str(label), ) - - profiler_rates = profiler_cfg.get("rates", [1]) - guidellm_args = runtime_config.build_guidellm_args( - benchmark_cfg=benchmark_cfg, - model_id=model_cfg["hf_model_id"], - data=workload["data"], - rates=profiler_rates, - max_seconds=profiler_max_seconds, - ) - try: - run_guidellm_benchmark( - endpoint_url=f"{endpoint_url}/v1", - name=_guidellm_job_name("guidellm-profiler", workload_key, deployment_name), + _run_guidellm_profiler_load( + deployment_name=deployment_name, namespace=namespace, - image=benchmark_cfg.get("image", "ghcr.io/vllm-project/guidellm:v0.6.0"), - timeout=benchmark_timeout, - pvc_size=benchmark_cfg.get("pvc_size", "5Gi"), - guidellm_args=guidellm_args, - hf_token_secret=benchmark_cfg.get("hf_token_secret", ""), - fs_group=benchmark_cfg.get("fs_group"), + endpoint_url=endpoint_url, + benchmark_cfg=benchmark_cfg, + model_cfg=model_cfg, + workload=workload, + workload_key=workload_key, + benchmark_timeout=benchmark_timeout, + profiler_cfg=profiler_cfg, ) finally: enable_profiler_gate( @@ -826,6 +909,81 @@ def _run_profiler_step( logger.warning("Failed to copy profiler traces", exc_info=True) +def _run_native_profiler_step( + *, + deployment_name: str, + namespace: str, + endpoint_url: str, + benchmark_cfg: dict, + model_cfg: dict, + workload: dict, + workload_key: str, + benchmark_timeout: int, + engine: str, + profiler_cfg: dict, +) -> None: + """Native path: POST /start_profile → GuideLLM → POST /stop_profile → copy traces_dir.""" + from projects.rhaiis.toolbox.control_native_profiler.main import run as control_native_profiler + from projects.rhaiis.toolbox.copy_profiler_traces.main import run as copy_profiler_traces + + labels = _profiler_labels(profiler_cfg, workload) + native = native_options(profiler_cfg) + directory = profiler_traces_dir(profiler_cfg) + stop_timeout = int(native.get("stop_timeout_seconds", 1800) or 1800) + start_body = build_sglang_start_body(profiler_cfg) if engine == "sglang" else "" + + for label in labels: + logger.info("Profiling label=%s (native HTTP start/stop)", label) + control_native_profiler( + endpoint_url=endpoint_url, + action="start", + timeout_seconds=60, + body=start_body, + name=deployment_name, + namespace=namespace, + traces_dir=directory, + ) + try: + _run_guidellm_profiler_load( + deployment_name=deployment_name, + namespace=namespace, + endpoint_url=endpoint_url, + benchmark_cfg=benchmark_cfg, + model_cfg=model_cfg, + workload=workload, + workload_key=workload_key, + benchmark_timeout=benchmark_timeout, + profiler_cfg=profiler_cfg, + ) + finally: + try: + control_native_profiler( + endpoint_url=endpoint_url, + action="stop", + timeout_seconds=stop_timeout, + name=deployment_name, + namespace=namespace, + traces_dir=directory, + ) + except Exception: + logger.warning( + "Native /stop_profile failed (ok if num_steps already ended the session)", + exc_info=True, + ) + + logger.info("Copying native profiler traces from %s", directory) + try: + copy_profiler_traces( + name=deployment_name, + namespace=namespace, + remote_dir=directory, + file_glob="*", + run_label=labels[0] if labels else "", + ) + except Exception: + logger.warning("Failed to copy profiler traces", exc_info=True) + + def _derive_profiler_label(workload: dict) -> str: """Auto-generate a profiler label like 'isl1000_osl1000' from the workload data string.""" data = workload.get("data", "") diff --git a/projects/rhaiis/postprocess/s3_dashboard.py b/projects/rhaiis/postprocess/s3_dashboard.py index 060288284..85c383fb4 100644 --- a/projects/rhaiis/postprocess/s3_dashboard.py +++ b/projects/rhaiis/postprocess/s3_dashboard.py @@ -101,7 +101,7 @@ def upload_profiler_traces_to_s3( trace_files = [ f for f in traces_dir.iterdir() - if f.name.startswith("trace") and "rank0" in f.name and f.suffix in (".json", ".gz") + if f.is_file() and _is_uploadable_trace(f) ] if not trace_files: return {"status": "skipped", "reason": "no trace files found"} @@ -196,6 +196,21 @@ def upload_predictor_log_to_s3( return {"status": "failed", "error": str(e)} +def _is_uploadable_trace(path: Path) -> bool: + """Rank-0 Chrome/Perfetto (and Proton/Nsight) artifacts after copy_profiler_traces.""" + name = path.name + if not name.startswith("trace") or "rank0" not in name: + return False + return ( + name.endswith(".json") + or name.endswith(".json.gz") + or name.endswith(".gz") + or name.endswith(".hatchet") + or name.endswith(".nsys-rep") + or name.endswith(".qdrep") + ) + + def _accelerator_s3_folder(accelerator: str) -> str: upper = accelerator.upper() if upper.startswith("H200"): diff --git a/projects/rhaiis/test/test_profiler_native.py b/projects/rhaiis/test/test_profiler_native.py new file mode 100644 index 000000000..1f025304b --- /dev/null +++ b/projects/rhaiis/test/test_profiler_native.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from projects.rhaiis.orchestration.profiler import ( + apply_native_profiler_deploy, + build_sglang_start_body, + build_vllm_profiler_config_json, + engine_supports_profiler, + is_native_backend, +) +from projects.rhaiis.postprocess.s3_dashboard import _is_uploadable_trace +from projects.rhaiis.toolbox.copy_profiler_traces.main import normalize_trace_name + + +def test_webhook_is_default_backend() -> None: + assert is_native_backend({}) is False + assert is_native_backend({"backend": "webhook"}) is False + assert is_native_backend({"backend": "native"}) is True + + +def test_engine_support_matrix() -> None: + webhook = {"backend": "webhook"} + native = {"backend": "native"} + assert engine_supports_profiler("vllm", webhook) + assert not engine_supports_profiler("sglang", webhook) + assert engine_supports_profiler("vllm", native) + assert engine_supports_profiler("sglang", native) + assert not engine_supports_profiler("trtllm", native) + + +def test_vllm_torch_profiler_config_json() -> None: + cfg = { + "kind": "torch", + "traces_dir": "/tmp/vllm_profile", + "native": {"with_stack": True, "ignore_frontend": True}, + } + parsed = json.loads(build_vllm_profiler_config_json(cfg)) + assert parsed["profiler"] == "torch" + assert parsed["torch_profiler_dir"] == "/tmp/vllm_profile" + assert parsed["ignore_frontend"] is True + assert "wait_iterations" not in parsed + assert "max_iterations" not in parsed + + +def test_vllm_torch_schedule_included_when_warmup_set() -> None: + cfg = { + "kind": "torch", + "native": {"warmup_iterations": 2, "active_iterations": 8, "wait_iterations": 1}, + } + parsed = json.loads(build_vllm_profiler_config_json(cfg)) + assert parsed["warmup_iterations"] == 2 + assert parsed["active_iterations"] == 8 + assert parsed["wait_iterations"] == 1 + + +def test_vllm_cuda_and_proton_kinds() -> None: + cuda = json.loads(build_vllm_profiler_config_json({"kind": "cuda"})) + assert cuda == {"profiler": "cuda"} + + proton = json.loads( + build_vllm_profiler_config_json( + {"kind": "proton", "traces_dir": "/tmp/vllm_profile"} + ) + ) + assert proton["profiler"] == "proton" + assert proton["proton_profiler_dir"] == "/tmp/vllm_profile" + assert proton["proton_data"] == "trace" + + +def test_apply_native_vllm_sets_profiler_config() -> None: + args: dict = {} + env: dict = {} + apply_native_profiler_deploy("vllm", args, env, {"backend": "native", "kind": "torch"}) + assert "profiler-config" in args + assert env["VLLM_RPC_TIMEOUT"] == "1800000" + + +def test_apply_native_cuda_nsys_wrap() -> None: + args: dict = {} + env: dict = {} + apply_native_profiler_deploy( + "vllm", + args, + env, + {"backend": "native", "kind": "cuda", "native": {"nsys_wrap": True}}, + ) + assert args["_nsys_wrap"] is True + assert args["_nsys_output"].endswith("/nsys_capture") + + +def test_apply_native_sglang_sets_env_only() -> None: + args: dict = {"tp-size": 2} + env: dict = {} + apply_native_profiler_deploy("sglang", args, env, {"backend": "native"}) + assert "profiler-config" not in args + assert env["SGLANG_TORCH_PROFILER_DIR"] == "/tmp/vllm_profile" + + +def test_sglang_start_body() -> None: + body = json.loads( + build_sglang_start_body({"native": {"num_steps": 10, "start_step": 5}}) + ) + assert body["num_steps"] == 10 + assert body["start_step"] == 5 + assert body["output_dir"] == "/tmp/vllm_profile" + + +def test_normalize_trace_name_prefixes_native_files() -> None: + name = normalize_trace_name("worker.pt.trace.json.gz", "isl1000_osl1000") + assert name.startswith("trace_") + assert "rank0" in name + assert "isl1000_osl1000" in name + assert name.endswith(".pt.trace.json.gz") + + +def test_uploadable_trace_accepts_gzip_and_proton() -> None: + assert _is_uploadable_trace(Path("trace_rank0_runisl_x.pt.trace.json.gz")) + assert _is_uploadable_trace(Path("trace_rank0_runisl.hatchet")) + assert not _is_uploadable_trace(Path("trace_rank1.json")) + assert not _is_uploadable_trace(Path("notes.txt")) diff --git a/projects/rhaiis/toolbox/control_native_profiler/main.py b/projects/rhaiis/toolbox/control_native_profiler/main.py new file mode 100644 index 000000000..17e8a3ec0 --- /dev/null +++ b/projects/rhaiis/toolbox/control_native_profiler/main.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python3 +"""Start or stop engine-native HTTP profiling (vLLM / SGLang /start_profile).""" + +from __future__ import annotations + +import json + +from projects.core.dsl import ( + RetryFailure, + entrypoint, + execute_tasks, + retry, + shell, + task, +) + + +@entrypoint +def run( + *, + endpoint_url: str, + action: str = "start", + timeout_seconds: int = 1800, + body: str = "", + name: str = "", + namespace: str = "", + traces_dir: str = "/tmp/vllm_profile", +): + """POST /start_profile or /stop_profile on the predictor. + + Args: + endpoint_url: Predictor base URL (no /v1 suffix), e.g. http://name-predictor.ns.svc:8080 + action: ``start`` or ``stop`` + timeout_seconds: curl --max-time (stop/flush can take many minutes) + body: Optional JSON string for the POST body (SGLang start options) + name: InferenceService name; used to mkdir traces_dir in the pod + namespace: Namespace of the InferenceService + traces_dir: Directory created in the pod before start (vLLM torch_profiler_dir) + """ + return execute_tasks(locals()) + + +@task +def validate_action(args, context): + if args.action not in ("start", "stop"): + raise ValueError(f"action must be 'start' or 'stop', got {args.action!r}") + base = args.endpoint_url.rstrip("/") + if base.endswith("/v1"): + base = base[:-3] + context.base_url = base + context.path = "/start_profile" if args.action == "start" else "/stop_profile" + return f"Native profiler {args.action}: {context.base_url}{context.path}" + + +@task +def ensure_traces_dir(args, context): + if args.action != "start" or not args.name or not args.namespace: + return "Skip mkdir (stop, or no pod locator)" + traces_dir = args.traces_dir + if not traces_dir.startswith("/tmp"): + raise ValueError(f"traces_dir must be under /tmp, got {traces_dir}") + pod = shell.run( + f"oc get pod -oname " + f"-lserving.kserve.io/inferenceservice={args.name} " + f"-n {args.namespace} | head -1", + check=False, + ) + pod_name = pod.stdout.strip() + if not pod_name: + raise RuntimeError(f"No predictor pod found for {args.name} in {args.namespace}") + shell.run( + f"oc exec {pod_name} -n {args.namespace} -- mkdir -p {traces_dir}", + ) + return f"Created {traces_dir} on {pod_name}" + + +@retry(attempts=8, delay=5, backoff=1.0, retry_on_exceptions=True) +@task +def call_profiler_api(args, context): + cmd = [ + "curl", + "-sS", + "-X", + "POST", + "-H", + "Content-Type: application/json", + "--max-time", + str(args.timeout_seconds), + "-w", + "\nHTTP_CODE:%{http_code}", + f"{context.base_url}{context.path}", + ] + if args.body: + cmd.extend(["-d", args.body]) + + result = shell.run( + cmd, + shell=False, + check=False, + timeout_seconds=float(args.timeout_seconds) + 30, + log_stdout=True, + ) + stdout = result.stdout or "" + http_code = _parse_http_code(stdout) + if http_code is None or http_code >= 500 or http_code == 0: + raise RetryFailure( + f"{args.action} {context.path} failed: http={http_code} rc={result.returncode} " + f"{stdout[-500:]}" + ) + if http_code == 404: + raise RuntimeError( + f"{context.path} returned 404 — start the server with --profiler-config " + "(vLLM) or a recent SGLang build that exposes /start_profile" + ) + if http_code >= 400: + raise RuntimeError(f"{args.action} {context.path} HTTP {http_code}: {stdout[-500:]}") + + body = stdout.rsplit("HTTP_CODE:", 1)[0].strip() + context.response = body + try: + parsed = json.loads(body) if body else {} + context.response_json = parsed + except json.JSONDecodeError: + context.response_json = {} + return f"{args.action} {context.path} HTTP {http_code}" + + +def _parse_http_code(stdout: str) -> int | None: + marker = "HTTP_CODE:" + if marker not in stdout: + return None + tail = stdout.rsplit(marker, 1)[-1].strip() + try: + return int(tail.split()[0]) + except (ValueError, IndexError): + return None + + +if __name__ == "__main__": + run.main() diff --git a/projects/rhaiis/toolbox/copy_profiler_traces/main.py b/projects/rhaiis/toolbox/copy_profiler_traces/main.py index 147fe8dfe..c452f57e2 100644 --- a/projects/rhaiis/toolbox/copy_profiler_traces/main.py +++ b/projects/rhaiis/toolbox/copy_profiler_traces/main.py @@ -1,5 +1,11 @@ #!/usr/bin/env python3 +from __future__ import annotations + +import re +import shutil +from pathlib import Path + from projects.core.dsl import ( entrypoint, execute_tasks, @@ -7,9 +13,45 @@ task, ) +_SAFE_DIR = re.compile(r"^/tmp(/[A-Za-z0-9._-]+)*$") +_SAFE_GLOB = re.compile(r"^[A-Za-z0-9._*?\[\]-]+$") + + +def normalize_trace_name(original: str, run_label: str = "") -> str: + """Make engine-native filenames match the webhook/S3 convention. + + S3 upload looks for ``trace_*rank0*``. vLLM writes ``*.pt.trace.json.gz``; + SGLang writes ``*.trace.json.gz``. Prefix those so the existing uploader + still finds rank-0 artifacts. + """ + name = Path(original).name + if run_label and f"run{run_label}" not in name: + name = f"run{run_label}_{name}" + if "rank" not in name.lower(): + name = f"rank0_{name}" + if not name.startswith("trace"): + name = f"trace_{name}" + return name + @entrypoint -def run(*, name: str, namespace: str): +def run( + *, + name: str, + namespace: str, + remote_dir: str = "/tmp", + file_glob: str = "trace_*.json*", + run_label: str = "", +): + """Copy profiler traces from the predictor pod. + + Args: + name: KServe InferenceService name + namespace: Namespace of the InferenceService + remote_dir: Directory inside the pod (webhook: /tmp, native: torch_profiler_dir) + file_glob: Files to tar. Use ``*`` to copy the whole remote_dir. + run_label: Optional gate/profile label injected into copied filenames + """ return execute_tasks(locals()) @@ -19,6 +61,15 @@ def setup_directories(args, context): return "Traces directory created" +@task +def validate_remote_path(args, context): + if not _SAFE_DIR.match(args.remote_dir): + raise ValueError(f"remote_dir must be an absolute /tmp path, got {args.remote_dir!r}") + if args.file_glob != "*" and not _SAFE_GLOB.match(args.file_glob): + raise ValueError(f"Unsafe file_glob: {args.file_glob!r}") + return f"Copying {args.file_glob} from {args.remote_dir}" + + @task def find_predictor_pod(args, context): result = shell.run( @@ -37,33 +88,54 @@ def find_predictor_pod(args, context): @task def list_trace_files(args, context): + if args.file_glob == "*": + ls_cmd = f"ls -A {args.remote_dir} 2>/dev/null || echo NO_TRACES" + else: + ls_cmd = f"ls {args.remote_dir}/{args.file_glob} 2>/dev/null || echo NO_TRACES" result = shell.run( - f"oc exec {context.pod_name} -n {args.namespace} " - "-- sh -c 'ls /tmp/trace_*.json* 2>/dev/null || echo NO_TRACES'", + f"oc exec {context.pod_name} -n {args.namespace} -- sh -c '{ls_cmd}'", check=False, log_stdout=False, ) if "NO_TRACES" in result.stdout or not result.stdout.strip(): - raise RuntimeError(f"No profiler traces found in pod {context.pod_name}") + raise RuntimeError(f"No profiler traces found in pod {context.pod_name}:{args.remote_dir}") trace_list = result.stdout.strip() - context.trace_count = len(trace_list.splitlines()) + context.trace_count = len([line for line in trace_list.splitlines() if line.strip()]) return f"Found {context.trace_count} trace files" @task def copy_traces(args, context): traces_dir = args.artifact_dir / "artifacts/traces" + tar_args = "." if args.file_glob == "*" else args.file_glob shell.run( 'bash -o pipefail -c "' f"oc exec {context.pod_name} -n {args.namespace}" - " -- sh -c 'cd /tmp && tar cf - trace_*.json*'" + f" -- sh -c 'cd {args.remote_dir} && tar cf - {tar_args}'" f' | tar --no-same-owner -xf - -C {traces_dir}"', ) - copied = list(traces_dir.glob("trace_*.json*")) + copied = _flatten_and_normalize(traces_dir, args.run_label) return f"Copied {len(copied)} trace files to {traces_dir}" +def _flatten_and_normalize(traces_dir: Path, run_label: str) -> list[Path]: + files = [p for p in traces_dir.rglob("*") if p.is_file()] + dest_files: list[Path] = [] + for src in files: + dest = traces_dir / normalize_trace_name(src.name, run_label) + if src.resolve() != dest.resolve(): + dest.parent.mkdir(parents=True, exist_ok=True) + if dest.exists(): + dest.unlink() + shutil.move(str(src), str(dest)) + dest_files.append(dest) + for path in sorted(traces_dir.rglob("*"), reverse=True): + if path.is_dir() and path != traces_dir and not any(path.iterdir()): + path.rmdir() + return dest_files + + if __name__ == "__main__": run.main() diff --git a/pyproject.toml b/pyproject.toml index 4faa72b56..abb190edd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -143,7 +143,7 @@ ignore = [ [tool.pytest.ini_options] minversion = "7.0" addopts = "-ra -q --strict-markers --strict-config" -testpaths = ["projects/core/tests", "projects/llm_d/tests", "projects/caliper/tests"] +testpaths = ["projects/core/tests", "projects/llm_d/tests", "projects/caliper/tests", "projects/rhaiis/test"] python_files = ["test_*.py", "*_test.py"] python_classes = ["Test*"] python_functions = ["test_*"]