From acf9a544f6190d0e178b86889c693d9e42acf50e Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Wed, 5 Aug 2026 14:36:56 +0100 Subject: [PATCH 01/10] refactor: Share GuideLLM dashboard postprocessing --- projects/caliper/engine/kpi/format.py | 6 +- projects/caliper/tests/test_kpi_format.py | 28 ++ .../postprocess/guidellm/dashboard.py | 450 ++++++++++++++++++ .../postprocess/guidellm/parsing/parsers.py | 36 +- .../llm_d/orchestration/config.d/cpt.yaml | 1 + projects/llm_d/orchestration/config.yaml | 6 +- .../llm_d/orchestration/presets.d/cks.yaml | 1 + .../llm_d/orchestration/presets.d/cpt.yaml | 2 + .../orchestration/presets.d/rhoai-rc.yaml | 1 + projects/llm_d/orchestration/test_phase.py | 10 + projects/llm_d/postprocess/__init__.py | 1 + projects/llm_d/postprocess/plugin.py | 260 ++++++++++ projects/llm_d/tests/test_postprocess_csv.py | 247 ++++++++++ projects/llm_d/tests/test_profiles.py | 5 + projects/rhaiis/postprocess/kpis.py | 132 +---- projects/rhaiis/postprocess/parser.py | 222 +-------- projects/rhaiis/postprocess/plugin.py | 165 +------ 17 files changed, 1083 insertions(+), 490 deletions(-) create mode 100644 projects/caliper/tests/test_kpi_format.py create mode 100644 projects/guidellm/postprocess/guidellm/dashboard.py create mode 100644 projects/llm_d/postprocess/__init__.py create mode 100644 projects/llm_d/postprocess/plugin.py create mode 100644 projects/llm_d/tests/test_postprocess_csv.py diff --git a/projects/caliper/engine/kpi/format.py b/projects/caliper/engine/kpi/format.py index 697cbe473..af5abad1b 100644 --- a/projects/caliper/engine/kpi/format.py +++ b/projects/caliper/engine/kpi/format.py @@ -63,8 +63,10 @@ def transform_kpis_to_hierarchical_format(kpis: list[dict], model) -> dict: if k not in ("higher_is_better",) and k not in varying_keys } - if not test_data["labels"]: - test_data["labels"] = test_labels + # KPI handlers may add project-specific labels only to a subset of the + # records. Preserve every label that is constant for this test instead + # of freezing the set from whichever KPI happened to be emitted first. + test_data["labels"].update(test_labels) # Store test metadata from first KPI if not test_data["metadata"]: diff --git a/projects/caliper/tests/test_kpi_format.py b/projects/caliper/tests/test_kpi_format.py new file mode 100644 index 000000000..fe0f1a56f --- /dev/null +++ b/projects/caliper/tests/test_kpi_format.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +from projects.caliper.engine.kpi.format import transform_kpis_to_hierarchical_format + + +def test_hierarchical_format_merges_common_labels_from_all_kpis(): + kpis = [ + { + "run_id": "run-1", + "kpi_id": "generic", + "value": 1, + "labels": {"model": "llama"}, + }, + { + "run_id": "run-1", + "kpi_id": "dashboard", + "value": 2, + "labels": {"model": "llama", "tensor_parallel_size": "2"}, + }, + ] + model = type("Model", (), {"plugin_module": "missing.plugin"})() + + output = transform_kpis_to_hierarchical_format(kpis, model) + + assert output["tests"][0]["labels"] == { + "model": "llama", + "tensor_parallel_size": "2", + } diff --git a/projects/guidellm/postprocess/guidellm/dashboard.py b/projects/guidellm/postprocess/guidellm/dashboard.py new file mode 100644 index 000000000..3f036bffa --- /dev/null +++ b/projects/guidellm/postprocess/guidellm/dashboard.py @@ -0,0 +1,450 @@ +"""Shared GuideLLM helpers for dashboard-compatible CSV exports.""" + +from __future__ import annotations + +import csv +import json +import logging +import re +from collections.abc import Callable +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from projects.caliper.engine.model import ( + ParseResult, + TestBaseNode, + UnifiedResultRecord, + UnifiedRunModel, +) + +logger = logging.getLogger(__name__) + + +# suffix, curve key, dashboard column, unit, higher-is-better +DASHBOARD_METRICS: tuple[tuple[str, str, str, str, bool | None], ...] = ( + ("output_tok_per_sec", "output_tok_per_sec", "output_tok/sec", "tokens/s", True), + ("total_tok_per_sec", "total_tok_per_sec", "total_tok/sec", "tokens/s", True), + ("measured_concurrency", "request_concurrency", "measured concurrency", "count", None), + ("measured_rps", "measured_rps", "measured rps", "req/s", True), + ("intended_concurrency", "intended_concurrency", "intended concurrency", "count", None), + ("completed_requests", "successful_requests", "successful_requests", "count", True), + ("failed_requests", "errored_requests", "errored_requests", "count", False), + ("ttft_median", "ttft_median", "ttft_median", "s", False), + ("ttft_p95", "ttft_p95", "ttft_p95", "s", False), + ("ttft_p99", "ttft_p99", "ttft_p99", "s", False), + ("ttft_p1", "ttft_p1", "ttft_p1", "s", False), + ("ttft_p999", "ttft_p999", "ttft_p999", "s", False), + ("ttft_mean", "ttft_mean", "ttft_mean", "s", False), + ("tpot_median", "tpot_median", "tpot_median", "s", False), + ("tpot_p95", "tpot_p95", "tpot_p95", "s", False), + ("tpot_p99", "tpot_p99", "tpot_p99", "s", False), + ("tpot_p1", "tpot_p1", "tpot_p1", "s", False), + ("tpot_p999", "tpot_p999", "tpot_p999", "s", False), + ("itl_median", "itl_median", "itl_median", "s", False), + ("itl_p95", "itl_p95", "itl_p95", "s", False), + ("itl_p99", "itl_p99", "itl_p99", "s", False), + ("itl_p1", "itl_p1", "itl_p1", "s", False), + ("itl_p999", "itl_p999", "itl_p999", "s", False), + ("itl_mean", "itl_mean", "itl_mean", "s", False), + ("request_latency_median", "request_latency_median", "request_latency_median", "s", False), + ("request_latency_min", "request_latency_min", "request_latency_min", "s", False), + ("request_latency_max", "request_latency_max", "request_latency_max", "s", False), + ( + "prompt_token_count_mean", + "prompt_token_count_mean", + "prompt_token_count_mean", + "tokens", + None, + ), + ("prompt_token_count_p99", "prompt_token_count_p99", "prompt_token_count_p99", "tokens", None), + ( + "output_token_count_mean", + "output_token_count_mean", + "output_token_count_mean", + "tokens", + None, + ), + ("output_token_count_p99", "output_token_count_p99", "output_token_count_p99", "tokens", None), +) + +SECONDS_TO_MS_COLUMNS = frozenset( + column + for _, _, column, unit, _ in DASHBOARD_METRICS + if unit == "s" and not column.startswith("request_latency_") +) + +# Metadata that can be emitted as dashboard labels. Runtime/test labels are +# merged last by ``dashboard_metadata_labels`` and therefore take precedence +# over values recovered from artifacts. +DASHBOARD_METADATA_LABEL_KEYS = frozenset( + { + "product_version", + "deployment_profile", + "model_name", + "hf_model_id", + "cluster", + "benchmark_key", + "replicas", + "tensor_parallel_size", + "runtime_args", + "image_tag", + "router_config", + "gpu_type", + } +) + + +def canonical_json(value: Any) -> str: + """Serialize structured metadata deterministically for labels and CSVs.""" + return json.dumps(value, separators=(",", ":"), sort_keys=True) + + +def normalize_product_version(value: Any) -> str: + """Normalize RHOAI/KServe versions to the dashboard naming convention.""" + text = str(value or "") + match = re.fullmatch(r"v(\d+)\.(\d+)\.\d+-ea\.(\d+)", text, re.IGNORECASE) + if match: + major, minor, early_access = match.groups() + return f"RHOAI-{major}.{minor}-EA{early_access}" + return text + + +def deployment_metadata_from_profile( + profile: dict[str, Any], *, profile_name: str | None = None +) -> dict[str, Any]: + """Extract shared deployment metadata from a resolved deployment profile.""" + metadata: dict[str, Any] = {} + if profile_name: + metadata["deployment_profile"] = profile_name + if "scheduler" in profile: + metadata["router_config"] = canonical_json(profile["scheduler"]) + elif profile.get("scheduler_manifest"): + metadata["router_config"] = canonical_json( + {"scheduler_manifest": profile["scheduler_manifest"]} + ) + return metadata + + +def dashboard_metadata_labels(record_metrics: dict[str, Any]) -> dict[str, str]: + """Build metadata labels with explicit runtime KPI labels taking precedence.""" + labels = { + key: str(value) + for key, value in record_metrics.items() + if key in DASHBOARD_METADATA_LABEL_KEYS and value is not None + } + kpi_labels = record_metrics.get("kpi_labels", {}) + if isinstance(kpi_labels, dict): + labels.update({key: str(value) for key, value in kpi_labels.items()}) + return labels + + +def validate_dashboard_fieldnames(fieldnames: list[str] | tuple[str, ...]) -> None: + """Reject CSV schemas that would silently drop a dashboard metric.""" + missing = sorted({column for *_, column, _, _ in DASHBOARD_METRICS} - set(fieldnames)) + if missing: + raise ValueError(f"Dashboard CSV fieldnames omit dashboard metrics: {', '.join(missing)}") + + +def _successful_stat(metrics: dict[str, Any], name: str, key: str) -> Any: + return metrics.get(name, {}).get("successful", {}).get(key) + + +def _successful_percentile(metrics: dict[str, Any], name: str, key: str) -> Any: + return metrics.get(name, {}).get("successful", {}).get("percentiles", {}).get(key) + + +def _milliseconds_to_seconds(value: Any) -> Any: + return value / 1000.0 if value is not None else None + + +def enrich_guidellm_parse_result( + base_result: ParseResult, nodes: list[TestBaseNode] +) -> ParseResult: + """Preserve dashboard metrics from raw GuideLLM files on parsed records.""" + nodes_by_path = {str(node.test_path): node for node in nodes} + records: list[UnifiedResultRecord] = [] + for record in base_result.records: + node = nodes_by_path.get(record.test_base_path) + if node is None or record.metrics.get("no_benchmarks_found"): + records.append(record) + continue + extra, curves = _extract_dashboard_metrics(node) + metrics = {**record.metrics, **extra} + metrics["performance_curves"] = { + **metrics.get("performance_curves", {}), + **curves, + } + records.append( + UnifiedResultRecord( + test_base_path=record.test_base_path, + distinguishing_labels=record.distinguishing_labels, + metrics=metrics, + run_identity=record.run_identity, + parse_notes=record.parse_notes, + ) + ) + return ParseResult(records=records, warnings=base_result.warnings) + + +def _extract_dashboard_metrics(node: TestBaseNode) -> tuple[dict[str, Any], dict[str, list]]: + files = sorted( + path + for path in node.artifact_paths + if path.name == "benchmarks.json" + or (path.name.startswith("benchmarks-rate-") and path.suffix == ".json") + ) + benchmarks: list[dict[str, Any]] = [] + metadata: dict[str, Any] = {} + args: dict[str, Any] = {} + for path in files: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError): + continue + benchmarks.extend(payload.get("benchmarks", [])) + metadata = metadata or payload.get("metadata", {}) + args = args or payload.get("args", {}) + if not benchmarks: + return {}, {} + + benchmarks.sort( + key=lambda benchmark: float( + benchmark.get("metrics", {}) + .get("requests_per_second", {}) + .get("successful", {}) + .get("mean", 0) + ) + ) + data_values = args.get("data", []) if isinstance(args, dict) else [] + if not data_values: + fallback_data = ( + benchmarks[0] + .get("benchmarker", {}) + .get("requests", {}) + .get("attributes", {}) + .get("data") + ) + data_values = [fallback_data] if fallback_data else [] + data_value = data_values[0] if isinstance(data_values, list) and data_values else data_values + tokens: dict[str, Any] = {} + if isinstance(data_value, dict): + tokens = data_value + elif data_value: + data_text = str(data_value) + try: + parsed_data = json.loads(data_text) + if isinstance(parsed_data, dict): + tokens = parsed_data + except json.JSONDecodeError: + tokens = dict(re.findall(r"(\w+)=([\d.]+)", data_text)) + starts = [ + b.get("scheduler_metrics", {}).get("start_time", b.get("start_time")) for b in benchmarks + ] + ends = [b.get("scheduler_metrics", {}).get("end_time", b.get("end_time")) for b in benchmarks] + starts = [value for value in starts if value is not None] + ends = [value for value in ends if value is not None] + extra = { + "guidellm_version": metadata.get("guidellm_version", ""), + "prompt_toks": int(float(tokens["prompt_tokens"])) if "prompt_tokens" in tokens else "", + "output_toks": int(float(tokens["output_tokens"])) if "output_tokens" in tokens else "", + "guidellm_start_time_ms": int(min(starts) * 1000) if starts else "", + "guidellm_end_time_ms": int(max(ends) * 1000) if ends else "", + } + curves = {curve_key: [] for _, curve_key, _, _, _ in DASHBOARD_METRICS} + run_uuids: list[str] = [] + for benchmark in benchmarks: + metrics = benchmark.get("metrics", {}) + strategy = benchmark.get("config", {}).get("strategy", {}) or benchmark.get( + "scheduler", {} + ).get("strategy", {}) + + totals = ( + benchmark.get("scheduler_metrics", {}).get("requests_made", {}) + or benchmark.get("request_totals", {}) + or benchmark.get("run_stats", {}).get("requests_made", {}) + or metrics.get("request_totals", {}) + ) + run_uuids.append( + str(benchmark.get("config", {}).get("run_id") or benchmark.get("run_id") or "") + ) + values = { + "output_tok_per_sec": metrics.get("output_tokens_per_second", {}) + .get("total", {}) + .get("mean"), + "total_tok_per_sec": metrics.get("tokens_per_second", {}).get("total", {}).get("mean"), + "request_concurrency": _successful_stat(metrics, "request_concurrency", "mean"), + "measured_rps": _successful_stat(metrics, "requests_per_second", "mean"), + "intended_concurrency": strategy.get("streams", strategy.get("max_concurrency")), + "successful_requests": totals.get("successful", 0), + "errored_requests": totals.get("errored", 0), + "ttft_median": _milliseconds_to_seconds( + _successful_stat(metrics, "time_to_first_token_ms", "median") + ), + "ttft_p95": _milliseconds_to_seconds( + _successful_percentile(metrics, "time_to_first_token_ms", "p95") + ), + "ttft_p99": _milliseconds_to_seconds( + _successful_percentile(metrics, "time_to_first_token_ms", "p99") + ), + "ttft_p1": _milliseconds_to_seconds( + _successful_percentile(metrics, "time_to_first_token_ms", "p01") + ), + "ttft_p999": _milliseconds_to_seconds( + _successful_percentile(metrics, "time_to_first_token_ms", "p999") + ), + "ttft_mean": _milliseconds_to_seconds( + _successful_stat(metrics, "time_to_first_token_ms", "mean") + ), + "tpot_median": _milliseconds_to_seconds( + _successful_stat(metrics, "time_per_output_token_ms", "median") + ), + "tpot_p95": _milliseconds_to_seconds( + _successful_percentile(metrics, "time_per_output_token_ms", "p95") + ), + "tpot_p99": _milliseconds_to_seconds( + _successful_percentile(metrics, "time_per_output_token_ms", "p99") + ), + "tpot_p1": _milliseconds_to_seconds( + _successful_percentile(metrics, "time_per_output_token_ms", "p01") + ), + "tpot_p999": _milliseconds_to_seconds( + _successful_percentile(metrics, "time_per_output_token_ms", "p999") + ), + "itl_median": _milliseconds_to_seconds( + _successful_stat(metrics, "inter_token_latency_ms", "median") + ), + "itl_p95": _milliseconds_to_seconds( + _successful_percentile(metrics, "inter_token_latency_ms", "p95") + ), + "itl_p99": _milliseconds_to_seconds( + _successful_percentile(metrics, "inter_token_latency_ms", "p99") + ), + "itl_p1": _milliseconds_to_seconds( + _successful_percentile(metrics, "inter_token_latency_ms", "p01") + ), + "itl_p999": _milliseconds_to_seconds( + _successful_percentile(metrics, "inter_token_latency_ms", "p999") + ), + "itl_mean": _milliseconds_to_seconds( + _successful_stat(metrics, "inter_token_latency_ms", "mean") + ), + "request_latency_median": _successful_stat(metrics, "request_latency", "median"), + "request_latency_min": _successful_stat(metrics, "request_latency", "min"), + "request_latency_max": _successful_stat(metrics, "request_latency", "max"), + "prompt_token_count_mean": _successful_stat(metrics, "prompt_token_count", "mean"), + "prompt_token_count_p99": _successful_percentile(metrics, "prompt_token_count", "p99"), + "output_token_count_mean": _successful_stat(metrics, "output_token_count", "mean"), + "output_token_count_p99": _successful_percentile(metrics, "output_token_count", "p99"), + } + for key in curves: + curves[key].append(values.get(key)) + extra["run_uuids"] = run_uuids + return extra, curves + + +def compute_dashboard_kpis(model: UnifiedRunModel, *, prefix: str) -> list[dict[str, Any]]: + """Emit one scalar KPI per dashboard metric and rate point.""" + timestamp = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + output: list[dict[str, Any]] = [] + for record in model.unified_result_records: + curves = record.metrics.get("performance_curves", {}) + rates = record.metrics.get("request_rate", []) + if not record.run_identity.get("guidellm") or not rates: + continue + metadata_labels = { + "guidellm_version": str(record.metrics.get("guidellm_version", "")), + "prompt_toks": str(record.metrics.get("prompt_toks", "")), + "output_toks": str(record.metrics.get("output_toks", "")), + "guidellm_start_time_ms": str(record.metrics.get("guidellm_start_time_ms", "")), + "guidellm_end_time_ms": str(record.metrics.get("guidellm_end_time_ms", "")), + } + metadata_labels.update(dashboard_metadata_labels(record.metrics)) + run_uuids = record.metrics.get("run_uuids", []) + for index in range(len(rates)): + labels = {**record.distinguishing_labels, **metadata_labels, "rate_index": str(index)} + if index < len(run_uuids): + labels["run_uuid"] = run_uuids[index] + for suffix, curve_key, _, unit, higher_is_better in DASHBOARD_METRICS: + values = curves.get(curve_key, []) + if index >= len(values) or values[index] is None: + continue + kpi_labels = dict(labels) + if higher_is_better is not None: + kpi_labels["higher_is_better"] = higher_is_better + output.append( + { + "schema_version": "1", + "kpi_id": f"{prefix}_{suffix}", + "value": float(values[index]), + "unit": unit, + "run_id": record.test_base_path, + "run_path": record.test_base_path, + "timestamp": timestamp, + "labels": kpi_labels, + "source": { + "test_base_path": record.test_base_path, + "plugin_module": model.plugin_module, + }, + } + ) + return output + + +def dashboard_kpi_catalog(*, prefix: str) -> list[dict[str, Any]]: + """Return catalog entries for the shared dashboard KPI set.""" + return [ + { + "kpi_id": f"{prefix}_{suffix}", + "name": f"{prefix}_{suffix}", + "unit": unit, + "higher_is_better": higher_is_better, + } + for suffix, _, _, unit, higher_is_better in DASHBOARD_METRICS + ] + + +def export_dashboard_kpis_to_csv( + kpi_records: list[dict[str, Any]], + output_path: Path, + *, + prefix: str, + fieldnames: list[str], + metadata_row: Callable[[dict[str, Any]], dict[str, Any]], +) -> str: + """Pivot scalar per-rate KPIs into a dashboard-compatible CSV.""" + validate_dashboard_fieldnames(fieldnames) + kpi_to_column = {f"{prefix}_{suffix}": column for suffix, _, column, _, _ in DASHBOARD_METRICS} + groups: dict[tuple[str, str], dict[str, Any]] = {} + labels_by_group: dict[tuple[str, str], dict[str, Any]] = {} + for kpi in kpi_records: + labels = kpi.get("labels", {}) + key = (str(kpi.get("run_path", "")), str(labels.get("rate_index", "0"))) + column = kpi_to_column.get(kpi.get("kpi_id", "")) + if column: + groups.setdefault(key, {})[column] = kpi.get("value") + if key not in labels_by_group or len(labels) > len(labels_by_group[key]): + labels_by_group[key] = labels + + rows: list[dict[str, Any]] = [] + for key in sorted(groups): + metrics = groups[key] + labels = labels_by_group.get(key, {}) + row: dict[str, Any] = dict.fromkeys(fieldnames, "") + row.update(metadata_row(labels)) + row.update(metrics) + if "intended concurrency" in row and row["intended concurrency"] in ("", None): + row["intended concurrency"] = labels.get("intended_concurrency", "") + for column in SECONDS_TO_MS_COLUMNS: + value = row.get(column) + if value not in ("", None): + row[column] = float(value) * 1000.0 + rows.append(row) + + output_path.parent.mkdir(parents=True, exist_ok=True) + with output_path.open("w", newline="", encoding="utf-8") as output: + writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction="ignore") + writer.writeheader() + writer.writerows(rows) + logger.info("Exported %d dashboard CSV rows to %s", len(rows), output_path) + return str(output_path) diff --git a/projects/guidellm/postprocess/guidellm/parsing/parsers.py b/projects/guidellm/postprocess/guidellm/parsing/parsers.py index b50088d7b..6602d7c5c 100644 --- a/projects/guidellm/postprocess/guidellm/parsing/parsers.py +++ b/projects/guidellm/postprocess/guidellm/parsing/parsers.py @@ -15,6 +15,10 @@ TestBaseNode, UnifiedResultRecord, ) +from projects.guidellm.postprocess.guidellm.dashboard import ( + canonical_json, + normalize_product_version, +) from .models import GuideLLMBenchmark, GuideLLMConfiguration @@ -118,14 +122,15 @@ def _is_llmisvc_artifact(path: Path) -> bool: return path.name in [ "llminferenceservice.yaml", "llminferenceservice.yml", - ] and "__capture_llmisvc_state" in str(path) + "llminferenceservice.json", + ] @staticmethod def _is_config_artifact(path: Path) -> bool: """Check if path is a config.yaml artifact.""" return path.name == "config.yaml" - def extract_fields_from_llmisvc(self, file_path: Path) -> dict[str, str]: + def extract_fields_from_llmisvc(self, file_path: Path) -> dict[str, Any]: """ Extract multiple fields from LLMInferenceService YAML file. @@ -148,7 +153,7 @@ def extract_fields_from_llmisvc(self, file_path: Path) -> dict[str, str]: if annotation_value: product_version = parse_product_version_from_annotation(annotation_value) if product_version: - result["product_version"] = product_version + result["product_version"] = normalize_product_version(product_version) logger.info(f"Extracted product_version '{product_version}' from {file_path}") # Extract deployment profile from forge annotation @@ -165,6 +170,30 @@ def extract_fields_from_llmisvc(self, file_path: Path) -> dict[str, str]: result["model_name"] = model_name logger.info(f"Extracted model_name '{model_name}' from {file_path}") + replicas = extract_field_by_jsonpath(yaml_data, "spec.replicas") + if replicas is not None: + result["replicas"] = replicas + + tensor_parallel_size = extract_field_by_jsonpath(yaml_data, "spec.parallelism.tensor") + if tensor_parallel_size is not None: + result["tensor_parallel_size"] = tensor_parallel_size + + router_config = extract_field_by_jsonpath(yaml_data, "spec.router.scheduler") + if router_config is not None: + result["router_config"] = canonical_json(router_config) + + serving_container = extract_field_by_jsonpath( + yaml_data, "spec.template.containers[0]", {} + ) + if isinstance(serving_container, dict): + image = serving_container.get("image") + if image: + result["image_tag"] = image + for env_var in serving_container.get("env", []): + if env_var.get("name") == "VLLM_ADDITIONAL_ARGS": + result["runtime_args"] = env_var.get("value", "") + break + except Exception as e: logger.warning(f"Failed to extract fields from {file_path}: {e}") @@ -550,6 +579,7 @@ def parse(self, nodes: list[TestBaseNode]) -> ParseResult: # Also look for LLMInferenceService YAML files and config.yaml files for system information llmisvc_files = [p for p in node.artifact_paths if self._is_llmisvc_artifact(p)] + llmisvc_files.sort(key=lambda path: "__capture_llmisvc_state" not in str(path)) config_files = [p for p in node.artifact_paths if self._is_config_artifact(p)] if not benchmarks_files: diff --git a/projects/llm_d/orchestration/config.d/cpt.yaml b/projects/llm_d/orchestration/config.d/cpt.yaml index bfb9b06c8..1eb4f7907 100644 --- a/projects/llm_d/orchestration/config.d/cpt.yaml +++ b/projects/llm_d/orchestration/config.d/cpt.yaml @@ -3,3 +3,4 @@ kpi: platform_name: null gpu_type: null test_harness: null + product_version: null diff --git a/projects/llm_d/orchestration/config.yaml b/projects/llm_d/orchestration/config.yaml index 05e3b5739..9a6d70a05 100644 --- a/projects/llm_d/orchestration/config.yaml +++ b/projects/llm_d/orchestration/config.yaml @@ -41,7 +41,7 @@ caliper: postprocess: enabled: true artifacts_dir: null - plugin_module: projects.guidellm.postprocess.guidellm.plugin + plugin_module: projects.llm_d.postprocess.plugin postprocess_config: null filtering: include_labels: [] @@ -63,8 +63,8 @@ caliper: output: kpis/kpis.json kpis_to_csv: enabled: true - output: kpis/kpis.csv - include_header_comments: true + output: kpis/dashboard.csv + include_header_comments: false artifacts_to_ai_data: enabled: true output_dir: ai_data diff --git a/projects/llm_d/orchestration/presets.d/cks.yaml b/projects/llm_d/orchestration/presets.d/cks.yaml index b0af352e3..f4942e1cb 100644 --- a/projects/llm_d/orchestration/presets.d/cks.yaml +++ b/projects/llm_d/orchestration/presets.d/cks.yaml @@ -18,6 +18,7 @@ cks: gpu.nvidia.com/class: H200 cpt.kpi.labels.platform_name: CKS + cpt.kpi.labels.gpu_type: H200 cks-intelligent: extends: diff --git a/projects/llm_d/orchestration/presets.d/cpt.yaml b/projects/llm_d/orchestration/presets.d/cpt.yaml index d8452bdfb..81604fd39 100644 --- a/projects/llm_d/orchestration/presets.d/cpt.yaml +++ b/projects/llm_d/orchestration/presets.d/cpt.yaml @@ -29,6 +29,8 @@ cpt-rhoai-release: cpt-release-testing-rhoai: extends: - cpt-rhoai-release + cpt.kpi.labels.gpu_type: H200 + cpt.kpi.labels.product_version: RHOAI-3.5-EA2 runtime.deployment_profile: - release-distributed-default - release-precise-prefix-cache diff --git a/projects/llm_d/orchestration/presets.d/rhoai-rc.yaml b/projects/llm_d/orchestration/presets.d/rhoai-rc.yaml index d1943a32c..330ccb106 100644 --- a/projects/llm_d/orchestration/presets.d/rhoai-rc.yaml +++ b/projects/llm_d/orchestration/presets.d/rhoai-rc.yaml @@ -4,4 +4,5 @@ rhoai-rc: rhoai-3.5-ea.2: extends: [rhoai-rc] + cpt.kpi.labels.product_version: RHOAI-3.5-EA2 platform.rhoai.custom_catalog.image: quay.io/rhoai/rhoai-fbc-fragment@sha256:e213152dc5bdaaf5724d269484914f3e1d7b4a4959d4405cb3eca9dd6297b310 diff --git a/projects/llm_d/orchestration/test_phase.py b/projects/llm_d/orchestration/test_phase.py index 6e8455b2b..d09d0a56e 100644 --- a/projects/llm_d/orchestration/test_phase.py +++ b/projects/llm_d/orchestration/test_phase.py @@ -16,6 +16,9 @@ from projects.core.library.postprocess import run_and_postprocess, write_test_labels from projects.core.library.run import SignalInterrupt from projects.core.orchestration.utils.k8s import ensure_namespace +from projects.guidellm.postprocess.guidellm.dashboard import ( + deployment_metadata_from_profile, +) from projects.guidellm.toolbox.run_guidellm_benchmark import build_guidellm_args from projects.guidellm.toolbox.run_guidellm_benchmark import main as run_guidellm_benchmark_command from projects.guidellm.toolbox.run_smoke_request import main as run_smoke_request_command @@ -163,6 +166,13 @@ def extract_kpi_labels_from_config() -> dict[str, str]: if test_harness: kpi_labels["test_harness"] = test_harness + product_version = config.project.get_config("cpt.kpi.labels.product_version") + if product_version: + kpi_labels["product_version"] = product_version + + deployment_profile = runtime_config.get_deployment_profile() + kpi_labels.update(deployment_metadata_from_profile(deployment_profile)) + return kpi_labels diff --git a/projects/llm_d/postprocess/__init__.py b/projects/llm_d/postprocess/__init__.py new file mode 100644 index 000000000..1dd94efdb --- /dev/null +++ b/projects/llm_d/postprocess/__init__.py @@ -0,0 +1 @@ +"""llm-d Caliper post-processing.""" diff --git a/projects/llm_d/postprocess/plugin.py b/projects/llm_d/postprocess/plugin.py new file mode 100644 index 000000000..9842ac266 --- /dev/null +++ b/projects/llm_d/postprocess/plugin.py @@ -0,0 +1,260 @@ +"""GuideLLM post-processing with the llm-d dashboard CSV schema.""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any + +import yaml + +from projects.caliper.engine.model import ( + ParseResult, + PostProcessingPlugin, + TestBaseNode, + UnifiedRunModel, +) +from projects.guidellm.postprocess.guidellm.ai_eval import GuideLLMAIEvaluator +from projects.guidellm.postprocess.guidellm.dashboard import ( + compute_dashboard_kpis, + dashboard_kpi_catalog, + deployment_metadata_from_profile, + enrich_guidellm_parse_result, + export_dashboard_kpis_to_csv, + normalize_product_version, + validate_dashboard_fieldnames, +) +from projects.guidellm.postprocess.guidellm.parsing import GuideLLMKpiHandler, GuideLLMParser +from projects.llm_d.orchestration.render_inference_service import _build_vllm_args +from projects.llm_d.orchestration.runtime_config import deep_merge + +FIELDNAMES = [ + "run", + "accelerator", + "model", + "version", + "prompt toks", + "output toks", + "TP", + "DP", + "EP", + "replicas", + "prefill_pod_count", + "decode_pod_count", + "router_config", + "measured concurrency", + "intended concurrency", + "measured rps", + "output_tok/sec", + "total_tok/sec", + "prompt_token_count_mean", + "prompt_token_count_p99", + "output_token_count_mean", + "output_token_count_p99", + "ttft_median", + "ttft_p95", + "ttft_p1", + "ttft_p999", + "tpot_median", + "tpot_p95", + "tpot_p99", + "tpot_p999", + "tpot_p1", + "itl_median", + "itl_p95", + "itl_p999", + "itl_p1", + "request_latency_median", + "request_latency_min", + "request_latency_max", + "successful_requests", + "errored_requests", + "uuid", + "ttft_mean", + "ttft_p99", + "itl_mean", + "itl_p99", + "runtime_args", + "guidellm_start_time_ms", + "guidellm_end_time_ms", + "image_tag", + "guidellm_version", + "notes", +] +validate_dashboard_fieldnames(FIELDNAMES) + + +class LlmDGuideLLMPlugin(PostProcessingPlugin): + """Keep generic GuideLLM outputs and add the llm-d dashboard projection.""" + + def __init__(self) -> None: + self.parser = GuideLLMParser() + self.kpi_handler = GuideLLMKpiHandler() + self.ai_evaluator = GuideLLMAIEvaluator() + + def parse(self, nodes: list[TestBaseNode]) -> ParseResult: + parsed = enrich_guidellm_parse_result(self.parser.parse(nodes), nodes) + nodes_by_path = {str(node.test_path): node for node in nodes} + records = [] + for record in parsed.records: + node = nodes_by_path.get(record.test_base_path) + deployment_metadata = _extract_deployment_metadata(node) if node else {} + test_labels = node.test_labels.get("labels", {}) if node else {} + hf_model_id = test_labels.get("model_name") + if hf_model_id: + record.metrics["hf_model_id"] = hf_model_id + for key, value in deployment_metadata.items(): + record.metrics.setdefault(key, value) + records.append(record) + return ParseResult(records=records, warnings=parsed.warnings) + + def kpi_catalog(self) -> list[dict[str, Any]]: + return self.kpi_handler.get_catalog() + dashboard_kpi_catalog(prefix="llmd") + + def compute_kpis(self, model: UnifiedRunModel) -> list[dict[str, Any]]: + return self.kpi_handler.compute_kpis(model) + compute_dashboard_kpis(model, prefix="llmd") + + def export_kpis_to_csv( + self, + kpi_records: list[dict[str, Any]], + output_path: Path, + include_header_comments: bool = True, + ) -> str: + def metadata_row(labels: dict[str, Any]) -> dict[str, Any]: + accelerator = labels.get("gpu_type") or labels.get("accelerator", "") + model = labels.get("hf_model_id") or labels.get("model_name", "") + run_model = model.replace("/", "-") + tp = labels.get("tensor_parallel_size") or labels.get("TP", "") + replicas = labels.get("replicas", "") + version = normalize_product_version( + labels.get("product_version") or labels.get("version", "") + ) + deployment_profile = labels.get("deployment_profile", "") + if version and deployment_profile: + version = f"{version}-{deployment_profile}" + return { + "run": "-".join(str(value) for value in (accelerator, run_model, tp) if value), + "accelerator": accelerator, + "model": model, + "version": version, + "prompt toks": labels.get("prompt_toks", ""), + "output toks": labels.get("output_toks", ""), + "TP": tp, + "DP": labels.get("DP") or labels.get("data_parallel_size") or 0, + "EP": labels.get("EP") or labels.get("expert_parallel_size") or 0, + "replicas": replicas, + "prefill_pod_count": labels.get("prefill_pod_count", 0), + "decode_pod_count": labels.get("decode_pod_count", 0), + "router_config": labels.get("router_config", ""), + "uuid": labels.get("run_uuid", ""), + "runtime_args": labels.get("runtime_args", ""), + "guidellm_start_time_ms": labels.get("guidellm_start_time_ms", ""), + "guidellm_end_time_ms": labels.get("guidellm_end_time_ms", ""), + "image_tag": labels.get("image_tag", ""), + "guidellm_version": labels.get("guidellm_version", ""), + "notes": labels.get("notes", ""), + } + + return export_dashboard_kpis_to_csv( + kpi_records, + output_path, + prefix="llmd", + fieldnames=FIELDNAMES, + metadata_row=metadata_row, + ) + + def build_ai_data_payload(self, model: UnifiedRunModel) -> dict[str, Any]: + return self.ai_evaluator.build_payload(model, self) + + +def get_plugin() -> PostProcessingPlugin: + return LlmDGuideLLMPlugin() + + +def _extract_deployment_metadata(node: TestBaseNode) -> dict[str, Any]: + """Recover llm-d deployment metadata when only config.yaml was exported.""" + config_path = next((path for path in node.artifact_paths if path.name == "config.yaml"), None) + if config_path is None: + return {} + try: + config = yaml.safe_load(config_path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return {} + if not isinstance(config, dict): + return {} + + runtime = config.get("runtime", {}) + deployments = config.get("deployments", {}) + profile_name = runtime.get("deployment_profile") + defaults = deployments.get("defaults", {}) + profile_override = deployments.get("profiles", {}).get(profile_name, {}) + profile = deep_merge(defaults, profile_override) + + metadata = { + "model_name": runtime.get("model_name"), + "hf_model_id": runtime.get("model_name"), + "replicas": profile.get("replicas"), + "tensor_parallel_size": profile.get("tensor_parallelism"), + } + configured_labels = config.get("cpt", {}).get("kpi", {}).get("labels", {}) + metadata["gpu_type"] = configured_labels.get("gpu_type") or _extract_accelerator(node) + metadata["product_version"] = configured_labels.get("product_version") + metadata.update(deployment_metadata_from_profile(profile, profile_name=profile_name)) + vllm_args = profile.get("vllm_extra", {}).get("args", {}) + if vllm_args: + metadata["runtime_args"] = " ".join(_build_vllm_args(vllm_args)) + serving_image = profile.get("serving_image") + if not serving_image: + serving_image = _extract_serving_image(node) + if serving_image: + metadata["image_tag"] = serving_image + return {key: value for key, value in metadata.items() if value not in (None, "")} + + +def _extract_serving_image(node: TestBaseNode) -> str | None: + deployment_path = next( + ( + path + for path in node.artifact_paths + if path.name + in { + "llminferenceservice.deployments.json", + "llminferenceservice.deployments.yaml", + } + ), + None, + ) + if deployment_path is None: + return None + try: + deployments = yaml.safe_load(deployment_path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return None + for deployment in deployments.get("items", []) if isinstance(deployments, dict) else []: + containers = ( + deployment.get("spec", {}).get("template", {}).get("spec", {}).get("containers", []) + ) + for container in containers: + if container.get("name") == "main" and container.get("image"): + return str(container["image"]) + return None + + +def _extract_accelerator(node: TestBaseNode) -> str | None: + """Infer the GPU family from captured serving-pod placement.""" + pods_path = next( + (path for path in node.artifact_paths if path.name == "llminferenceservice.pods.yaml"), + None, + ) + if pods_path is None: + return None + try: + pods = yaml.safe_load(pods_path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return None + for pod in pods.get("items", []) if isinstance(pods, dict) else []: + node_name = str(pod.get("spec", {}).get("nodeName", "")) + match = re.search(r"(?:^|-)gpu-([a-z]+\d+[a-z0-9]*)(?:-|$)", node_name, re.IGNORECASE) + if match: + return match.group(1).upper() + return None diff --git a/projects/llm_d/tests/test_postprocess_csv.py b/projects/llm_d/tests/test_postprocess_csv.py new file mode 100644 index 000000000..e2b29cb2d --- /dev/null +++ b/projects/llm_d/tests/test_postprocess_csv.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +import csv +import json + +import yaml + +from projects.caliper.engine.model import TestBaseNode as CaliperTestBaseNode +from projects.caliper.engine.model import UnifiedRunModel +from projects.llm_d.postprocess.plugin import FIELDNAMES, LlmDGuideLLMPlugin + + +def _metric(**values): + return {"successful": values} + + +def test_llmd_plugin_exports_dashboard_compatible_csv(tmp_path): + benchmark_path = tmp_path / "benchmarks.json" + benchmark_path.write_text( + json.dumps( + { + "metadata": {"guidellm_version": "0.5.3"}, + "args": {"data": ["prompt_tokens=1000,output_tokens=1000"]}, + "benchmarks": [ + { + "config": { + "run_id": "guidellm-run-1", + "strategy": {"type_": "concurrent", "streams": 8}, + }, + "scheduler": {"state": {"start_time": 10, "end_time": 20}}, + "scheduler_metrics": { + "start_time": 10, + "end_time": 20, + "requests_made": {"successful": 80, "errored": 2}, + }, + "metrics": { + "requests_per_second": _metric(mean=7.5), + "request_concurrency": _metric(mean=7.8), + "output_tokens_per_second": { + "successful": {"mean": 900}, + "total": {"mean": 900}, + }, + "tokens_per_second": {"total": {"mean": 1800}}, + "time_to_first_token_ms": _metric( + mean=12, + median=10, + percentiles={"p01": 2, "p95": 20, "p99": 25, "p999": 30}, + ), + "time_per_output_token_ms": _metric( + median=3, + percentiles={"p01": 1, "p95": 5, "p99": 6, "p999": 7}, + ), + "inter_token_latency_ms": _metric( + mean=4, + median=3, + percentiles={"p01": 1, "p95": 6, "p99": 7, "p999": 8}, + ), + "request_latency": _metric(median=100, min=80, max=140), + "prompt_token_count": _metric(mean=1000, percentiles={"p99": 1000}), + "output_token_count": _metric(mean=1000, percentiles={"p99": 1000}), + }, + } + ], + } + ), + encoding="utf-8", + ) + llmisvc_path = tmp_path / "llminferenceservice.yaml" + llmisvc_path.write_text( + yaml.safe_dump( + { + "metadata": { + "annotations": {"forge.openshift.io/deployment-profile": "precise-prefix-cache"} + }, + "spec": { + "replicas": 4, + "model": {"name": "redhatai-llama-3-3-70b-instruct"}, + "parallelism": {"tensor": 2}, + "router": {"scheduler": {"config": "precise"}}, + "template": { + "containers": [ + { + "image": "registry.example/vllm:ea2", + "env": [ + { + "name": "VLLM_ADDITIONAL_ARGS", + "value": "--enable-prefix-caching", + } + ], + } + ] + }, + }, + } + ), + encoding="utf-8", + ) + node = CaliperTestBaseNode( + directory=tmp_path, + test_path=tmp_path, + artifact_paths=[benchmark_path, llmisvc_path], + test_labels={ + "labels": { + "load_shape": "concurrent-1k-1k", + "model_name": "RedHatAI/Llama-3.3-70B-Instruct-FP8-dynamic", + }, + "kpi_labels": { + "gpu_type": "H200", + "product_version": "RHOAI-3.5-EA2", + "test_harness": "rhoai-release", + }, + }, + ) + plugin = LlmDGuideLLMPlugin() + parsed = plugin.parse([node]) + model = UnifiedRunModel( + plugin_module="projects.llm_d.postprocess.plugin", + base_directory=str(tmp_path), + test_nodes=[node], + unified_result_records=parsed.records, + ) + + output_path = tmp_path / "dashboard.csv" + plugin.export_kpis_to_csv(plugin.compute_kpis(model), output_path) + + with output_path.open(newline="", encoding="utf-8") as output: + reader = csv.DictReader(output) + rows = list(reader) + assert reader.fieldnames == FIELDNAMES + assert len(rows) == 1 + row = rows[0] + assert row["run"] == "H200-RedHatAI-Llama-3.3-70B-Instruct-FP8-dynamic-2" + assert row["accelerator"] == "H200" + assert row["model"] == "RedHatAI/Llama-3.3-70B-Instruct-FP8-dynamic" + assert row["version"] == "RHOAI-3.5-EA2-precise-prefix-cache" + assert row["TP"] == "2" + assert row["DP"] == "0" + assert row["EP"] == "0" + assert row["replicas"] == "4" + assert row["router_config"] == '{"config":"precise"}' + assert row["prompt toks"] == "1000" + assert row["successful_requests"] == "80.0" + assert row["uuid"] == "guidellm-run-1" + assert row["ttft_median"] == "10.0" + assert row["runtime_args"] == "--enable-prefix-caching" + + +def test_llmd_plugin_recovers_deployment_metadata_from_config(tmp_path): + config_path = tmp_path / "config.yaml" + config_path.write_text( + yaml.safe_dump( + { + "cpt": { + "kpi": { + "labels": { + "gpu_type": "H200", + "product_version": "RHOAI-3.5-EA2", + } + } + }, + "runtime": { + "model_name": "RedHatAI/Llama-3.3-70B-Instruct-FP8-dynamic", + "deployment_profile": "precise-prefix-cache", + }, + "deployments": { + "defaults": { + "replicas": 1, + "tensor_parallelism": 1, + "vllm_extra": { + "args": { + "gpu_memory_utilization": 0.92, + "enable_prefix_caching": True, + } + }, + }, + "profiles": { + "precise-prefix-cache": { + "replicas": 4, + "tensor_parallelism": 2, + "scheduler": {}, + "vllm_extra": {"args": {"block_size": 64}}, + } + }, + }, + } + ), + encoding="utf-8", + ) + node = CaliperTestBaseNode( + directory=tmp_path, + test_path=tmp_path, + artifact_paths=[config_path], + test_labels={"labels": {}}, + ) + + from projects.llm_d.postprocess.plugin import _extract_deployment_metadata + + metadata = _extract_deployment_metadata(node) + + assert metadata["replicas"] == 4 + assert metadata["tensor_parallel_size"] == 2 + assert metadata["router_config"] == "{}" + assert metadata["gpu_type"] == "H200" + assert metadata["product_version"] == "RHOAI-3.5-EA2" + assert set(metadata["runtime_args"].split()) == { + "--gpu-memory-utilization=0.92", + "--enable-prefix-caching", + "--block-size=64", + } + + +def test_llmd_plugin_infers_accelerator_from_serving_pod_node(tmp_path): + config_path = tmp_path / "config.yaml" + config_path.write_text( + yaml.safe_dump( + { + "runtime": {"deployment_profile": "default"}, + "deployments": {"defaults": {}, "profiles": {"default": {}}}, + } + ), + encoding="utf-8", + ) + pods_path = tmp_path / "llminferenceservice.pods.yaml" + pods_path.write_text( + yaml.safe_dump( + { + "items": [ + { + "spec": { + "nodeName": "psap-worker-2-gpu-h200-k6qsd", + } + } + ] + } + ), + encoding="utf-8", + ) + node = CaliperTestBaseNode( + directory=tmp_path, + test_path=tmp_path, + artifact_paths=[config_path, pods_path], + test_labels={"labels": {}}, + ) + + from projects.llm_d.postprocess.plugin import _extract_deployment_metadata + + assert _extract_deployment_metadata(node)["gpu_type"] == "H200" diff --git a/projects/llm_d/tests/test_profiles.py b/projects/llm_d/tests/test_profiles.py index f1473cd21..a4c3d8189 100644 --- a/projects/llm_d/tests/test_profiles.py +++ b/projects/llm_d/tests/test_profiles.py @@ -175,6 +175,11 @@ def test_release_preset_expands_benchmark_list_and_merges_workload_args() -> Non core_config.project.get_config("cpt.kpi.labels.test_harness", print=False) == "rhoai-release" ) + assert core_config.project.get_config("cpt.kpi.labels.gpu_type", print=False) == "H200" + assert ( + core_config.project.get_config("cpt.kpi.labels.product_version", print=False) + == "RHOAI-3.5-EA2" + ) assert core_config.project.get_config("runtime.deployment_profile", print=False) == [ "release-distributed-default", diff --git a/projects/rhaiis/postprocess/kpis.py b/projects/rhaiis/postprocess/kpis.py index 1e8c7ece2..fbcab1cc4 100644 --- a/projects/rhaiis/postprocess/kpis.py +++ b/projects/rhaiis/postprocess/kpis.py @@ -1,139 +1,19 @@ from __future__ import annotations -from datetime import UTC, datetime from typing import Any from projects.caliper.engine.model import UnifiedRunModel - -# All per-rate-point metrics we emit as KPIs. -# (kpi_id, performance_curves key, unit, higher_is_better) -_CURVE_KPI_MAPPINGS: list[tuple[str, str, str, bool | None]] = [ - # Throughput - ("rhaiis_output_tok_per_sec", "output_tok_per_sec", "tokens/s", True), - ("rhaiis_total_tok_per_sec", "total_tok_per_sec", "tokens/s", True), - # Concurrency / rate - ("rhaiis_measured_concurrency", "request_concurrency", "count", None), - ("rhaiis_measured_rps", "measured_rps", "req/s", True), - ("rhaiis_intended_concurrency", "intended_concurrency", "count", None), - # Request counts - ("rhaiis_completed_requests", "successful_requests", "count", True), - ("rhaiis_failed_requests", "errored_requests", "count", False), - # TTFT - ("rhaiis_ttft_median", "ttft_median", "s", False), - ("rhaiis_ttft_p95", "ttft_p95", "s", False), - ("rhaiis_ttft_p99", "ttft_p99", "s", False), - ("rhaiis_ttft_p1", "ttft_p1", "s", False), - ("rhaiis_ttft_p999", "ttft_p999", "s", False), - ("rhaiis_ttft_mean", "ttft_mean", "s", False), - # TPOT - ("rhaiis_tpot_median", "tpot_median", "s", False), - ("rhaiis_tpot_p95", "tpot_p95", "s", False), - ("rhaiis_tpot_p99", "tpot_p99", "s", False), - ("rhaiis_tpot_p1", "tpot_p1", "s", False), - ("rhaiis_tpot_p999", "tpot_p999", "s", False), - # ITL - ("rhaiis_itl_median", "itl_median", "s", False), - ("rhaiis_itl_p95", "itl_p95", "s", False), - ("rhaiis_itl_p99", "itl_p99", "s", False), - ("rhaiis_itl_p1", "itl_p1", "s", False), - ("rhaiis_itl_p999", "itl_p999", "s", False), - ("rhaiis_itl_mean", "itl_mean", "s", False), - # Request latency - ("rhaiis_request_latency_median", "request_latency_median", "s", False), - ("rhaiis_request_latency_min", "request_latency_min", "s", False), - ("rhaiis_request_latency_max", "request_latency_max", "s", False), - # Token counts - ("rhaiis_prompt_token_count_mean", "prompt_token_count_mean", "tokens", None), - ("rhaiis_prompt_token_count_p99", "prompt_token_count_p99", "tokens", None), - ("rhaiis_output_token_count_mean", "output_token_count_mean", "tokens", None), - ("rhaiis_output_token_count_p99", "output_token_count_p99", "tokens", None), -] +from projects.guidellm.postprocess.guidellm.dashboard import ( + compute_dashboard_kpis, + dashboard_kpi_catalog, +) class RhaiisKpiHandler: @staticmethod def get_catalog() -> list[dict[str, Any]]: - return [ - {"kpi_id": kpi_id, "name": kpi_id, "unit": unit, "higher_is_better": hib} - for kpi_id, _, unit, hib in _CURVE_KPI_MAPPINGS - ] + return dashboard_kpi_catalog(prefix="rhaiis") @staticmethod def compute_kpis(model: UnifiedRunModel) -> list[dict[str, Any]]: - ts = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") - out: list[dict[str, Any]] = [] - - mlflow_dest = {} - for node in model.test_nodes: - dest = node.test_labels.get("mlflow_destination") - if isinstance(dest, dict) and dest.get("run_id"): - mlflow_dest = dest - break - - for r in model.unified_result_records: - if not r.run_identity.get("guidellm"): - continue - if r.metrics.get("no_benchmarks_found"): - continue - - base_labels = {**r.distinguishing_labels} - curves = r.metrics.get("performance_curves", {}) - request_rates = r.metrics.get("request_rate", []) - - # Report-level metadata — same for every rate point in this record - meta_labels = { - "guidellm_version": str(r.metrics.get("guidellm_version", "")), - "prompt_toks": str(r.metrics.get("prompt_toks", "")), - "output_toks": str(r.metrics.get("output_toks", "")), - "guidellm_start_time_ms": str(r.metrics.get("guidellm_start_time_ms", "")), - "guidellm_end_time_ms": str(r.metrics.get("guidellm_end_time_ms", "")), - "mlflow_run_id": mlflow_dest.get("run_id", ""), - "mlflow_experiment_id": mlflow_dest.get("experiment_id", ""), - } - - if not curves or not request_rates: - continue - - n_rates = len(request_rates) - - for idx in range(n_rates): - rate_labels = { - **base_labels, - **meta_labels, - "rate_index": str(idx), - } - - for kpi_id, curve_key, unit, higher_is_better in _CURVE_KPI_MAPPINGS: - curve_values = curves.get(curve_key, []) - if idx >= len(curve_values): - continue - raw_value = curve_values[idx] - if raw_value is None: - continue - try: - value = float(raw_value) - except (TypeError, ValueError): - continue - - labels = {**rate_labels} - if higher_is_better is not None: - labels["higher_is_better"] = higher_is_better - - out.append( - { - "schema_version": "1", - "kpi_id": kpi_id, - "value": value, - "unit": unit, - "run_id": r.test_base_path, - "run_path": r.test_base_path, - "timestamp": ts, - "labels": labels, - "source": { - "test_base_path": r.test_base_path, - "plugin_module": model.plugin_module, - }, - } - ) - - return out + return compute_dashboard_kpis(model, prefix="rhaiis") diff --git a/projects/rhaiis/postprocess/parser.py b/projects/rhaiis/postprocess/parser.py index a607811f8..e982b1fac 100644 --- a/projects/rhaiis/postprocess/parser.py +++ b/projects/rhaiis/postprocess/parser.py @@ -1,227 +1,15 @@ from __future__ import annotations -import json -import logging -import re -from typing import Any - -from projects.caliper.engine.model import ( - ParseResult, - TestBaseNode, - UnifiedResultRecord, -) -from projects.guidellm.postprocess.guidellm.parsing.parsers import ( - GuideLLMParser, -) - -logger = logging.getLogger(__name__) +from projects.caliper.engine.model import ParseResult, TestBaseNode +from projects.guidellm.postprocess.guidellm.dashboard import enrich_guidellm_parse_result +from projects.guidellm.postprocess.guidellm.parsing.parsers import GuideLLMParser class RhaiisParser: - """Extends GuideLLMParser with additional metrics for model-furnace parity.""" + """Parse GuideLLM results and retain dashboard-specific metrics.""" def __init__(self) -> None: self._base_parser = GuideLLMParser() def parse(self, nodes: list[TestBaseNode]) -> ParseResult: - base_result = self._base_parser.parse(nodes) - - enriched_records = [] - for record in base_result.records: - if record.metrics.get("no_benchmarks_found"): - enriched_records.append(record) - continue - - node = _find_node_for_record(record, nodes) - if node: - extra_metrics, extra_curves = _extract_extra_metrics(node) - merged_metrics = {**record.metrics, **extra_metrics} - if extra_curves: - existing_curves = merged_metrics.get("performance_curves", {}) - existing_curves.update(extra_curves) - merged_metrics["performance_curves"] = existing_curves - enriched_records.append( - UnifiedResultRecord( - test_base_path=record.test_base_path, - distinguishing_labels=record.distinguishing_labels, - metrics=merged_metrics, - run_identity=record.run_identity, - parse_notes=record.parse_notes, - ) - ) - else: - enriched_records.append(record) - - return ParseResult(records=enriched_records, warnings=base_result.warnings) - - -def _find_node_for_record( - record: UnifiedResultRecord, - nodes: list[TestBaseNode], -) -> TestBaseNode | None: - for node in nodes: - if str(node.test_path) == record.test_base_path: - return node - return None - - -def _bench_sort_key(bench: dict) -> float: - """Sort key matching GuideLLMParser: requests_per_second.successful.mean.""" - return float( - bench.get("metrics", {}).get("requests_per_second", {}).get("successful", {}).get("mean", 0) - ) - - -def _extract_extra_metrics(node: TestBaseNode) -> tuple[dict[str, Any], dict[str, list]]: - """Return (scalar_metrics, extra_curves) extracted from raw benchmarks.json files.""" - extra: dict[str, Any] = {} - benchmarks_files = sorted( - [ - p - for p in node.artifact_paths - if p.name == "benchmarks.json" - or (p.name.startswith("benchmarks-rate-") and p.suffix == ".json") - ], - key=lambda p: p.name, - ) - if not benchmarks_files: - return extra, {} - - all_benchmarks: list[dict] = [] - report_metadata: dict[str, Any] = {} - report_args: dict[str, Any] = {} - - for bf in benchmarks_files: - try: - data = json.loads(bf.read_text(encoding="utf-8")) - except (json.JSONDecodeError, OSError): - continue - all_benchmarks.extend(data.get("benchmarks", [])) - if not report_metadata: - report_metadata = data.get("metadata", {}) - if not report_args: - report_args = data.get("args", {}) - - if not all_benchmarks: - return extra, {} - - # Sort by request_rate to match GuideLLMParser curve ordering - all_benchmarks.sort(key=_bench_sort_key) - - # Report-level metadata - extra["guidellm_version"] = report_metadata.get("guidellm_version", "") - - data_str = "" - if isinstance(report_args, dict): - data_list = report_args.get("data", []) - if data_list: - data_str = data_list[0] if isinstance(data_list, list) else str(data_list) - tokens = dict(re.findall(r"(\w+)=([\d.]+)", data_str)) - extra["prompt_toks"] = int(float(tokens["prompt_tokens"])) if "prompt_tokens" in tokens else "" - extra["output_toks"] = int(float(tokens["output_tokens"])) if "output_tokens" in tokens else "" - - start_times = [] - end_times = [] - for bench in all_benchmarks: - sched = bench.get("scheduler_metrics", {}) - if "start_time" in sched: - start_times.append(sched["start_time"]) - if "end_time" in sched: - end_times.append(sched["end_time"]) - extra["guidellm_start_time_ms"] = int(min(start_times) * 1000) if start_times else "" - extra["guidellm_end_time_ms"] = int(max(end_times) * 1000) if end_times else "" - - # Scalar extras from first benchmark (backward compat) - bench0 = all_benchmarks[0] - m0 = bench0.get("metrics", {}) - - def _percentile0(metric_name: str, pct: str, default: float = 0.0) -> float: - return float( - m0.get(metric_name, {}).get("successful", {}).get("percentiles", {}).get(pct, default) - ) - - def _stat0(metric_name: str, stat: str, default: float = 0.0) -> float: - return float(m0.get(metric_name, {}).get("successful", {}).get(stat, default)) - - extra["ttft_p99"] = _percentile0("time_to_first_token_ms", "p99") / 1000.0 - extra["tpot_p99"] = _percentile0("time_per_output_token_ms", "p99") / 1000.0 - extra["itl_p99"] = _percentile0("inter_token_latency_ms", "p99") / 1000.0 - - request_totals0 = m0.get("request_totals", {}) - extra["completed_requests"] = int(request_totals0.get("successful", 0)) - extra["failed_requests"] = int(request_totals0.get("errored", 0)) - extra["prompt_token_count_mean"] = _stat0("prompt_token_count", "mean") - - concurrency = _stat0("request_concurrency", "mean") - if concurrency > 0: - extra["request_concurrency"] = concurrency - - # Per-benchmark extra curves — indices align with GuideLLM's sorted curves - extra_curves: dict[str, list] = { - "ttft_p1": [], - "ttft_p999": [], - "ttft_mean": [], - "tpot_p1": [], - "tpot_p999": [], - "itl_p1": [], - "itl_p999": [], - "itl_mean": [], - "request_latency_min": [], - "request_latency_max": [], - "measured_rps": [], - "prompt_token_count_mean": [], - "prompt_token_count_p99": [], - "output_token_count_mean": [], - "output_token_count_p99": [], - "output_tok_per_sec": [], - "total_tok_per_sec": [], - "successful_requests": [], - "errored_requests": [], - "intended_concurrency": [], - } - - for bench in all_benchmarks: - metrics = bench.get("metrics", {}) - strategy = bench.get("config", {}).get("strategy", {}) - - def _pct(metric_name: str, pct: str, _m: dict = metrics): - return _m.get(metric_name, {}).get("successful", {}).get("percentiles", {}).get(pct) - - def _pct_s(metric_name: str, pct: str): - v = _pct(metric_name, pct) - return v / 1000.0 if v is not None else None - - def _stat(metric_name: str, stat: str, _m: dict = metrics): - return _m.get(metric_name, {}).get("successful", {}).get(stat) - - def _stat_s(metric_name: str, stat: str): - v = _stat(metric_name, stat) - return v / 1000.0 if v is not None else None - - def _total_stat(metric_name: str, stat: str, _m: dict = metrics): - return _m.get(metric_name, {}).get("total", {}).get(stat) - - request_totals = metrics.get("request_totals", {}) - - extra_curves["ttft_p1"].append(_pct_s("time_to_first_token_ms", "p01")) - extra_curves["ttft_p999"].append(_pct_s("time_to_first_token_ms", "p999")) - extra_curves["ttft_mean"].append(_stat_s("time_to_first_token_ms", "mean")) - extra_curves["tpot_p1"].append(_pct_s("time_per_output_token_ms", "p01")) - extra_curves["tpot_p999"].append(_pct_s("time_per_output_token_ms", "p999")) - extra_curves["itl_p1"].append(_pct_s("inter_token_latency_ms", "p01")) - extra_curves["itl_p999"].append(_pct_s("inter_token_latency_ms", "p999")) - extra_curves["itl_mean"].append(_stat_s("inter_token_latency_ms", "mean")) - extra_curves["request_latency_min"].append(_stat("request_latency", "min")) - extra_curves["request_latency_max"].append(_stat("request_latency", "max")) - extra_curves["measured_rps"].append(_stat("requests_per_second", "mean")) - extra_curves["prompt_token_count_mean"].append(_stat("prompt_token_count", "mean")) - extra_curves["prompt_token_count_p99"].append(_pct("prompt_token_count", "p99")) - extra_curves["output_token_count_mean"].append(_stat("output_token_count", "mean")) - extra_curves["output_token_count_p99"].append(_pct("output_token_count", "p99")) - extra_curves["output_tok_per_sec"].append(_total_stat("output_tokens_per_second", "mean")) - extra_curves["total_tok_per_sec"].append(_total_stat("tokens_per_second", "mean")) - extra_curves["successful_requests"].append(request_totals.get("successful", 0)) - extra_curves["errored_requests"].append(request_totals.get("errored", 0)) - extra_curves["intended_concurrency"].append(strategy.get("streams")) - - return extra, extra_curves + return enrich_guidellm_parse_result(self._base_parser.parse(nodes), nodes) diff --git a/projects/rhaiis/postprocess/plugin.py b/projects/rhaiis/postprocess/plugin.py index 0ae902860..531cc5f80 100644 --- a/projects/rhaiis/postprocess/plugin.py +++ b/projects/rhaiis/postprocess/plugin.py @@ -1,8 +1,5 @@ from __future__ import annotations -import csv -import logging -from collections import defaultdict from pathlib import Path from typing import Any @@ -12,72 +9,11 @@ TestBaseNode, UnifiedRunModel, ) +from projects.guidellm.postprocess.guidellm.dashboard import export_dashboard_kpis_to_csv from .kpis import RhaiisKpiHandler from .parser import RhaiisParser -logger = logging.getLogger(__name__) - - -# CSV columns whose KPI values are in seconds but the dashboard expects milliseconds. -# GuideLLM parser converts `*_ms` metrics to seconds; the old CSV pipeline kept them in ms. -_SECONDS_TO_MS_COLUMNS = frozenset( - { - "ttft_median", - "ttft_p95", - "ttft_p99", - "ttft_p1", - "ttft_p999", - "ttft_mean", - "tpot_median", - "tpot_p95", - "tpot_p99", - "tpot_p1", - "tpot_p999", - "itl_median", - "itl_p95", - "itl_p99", - "itl_p1", - "itl_p999", - "itl_mean", - } -) - -# KPI ID → dashboard CSV column name mapping (complete) -_KPI_TO_CSV_COLUMN = { - "rhaiis_output_tok_per_sec": "output_tok/sec", - "rhaiis_total_tok_per_sec": "total_tok/sec", - "rhaiis_measured_concurrency": "measured concurrency", - "rhaiis_measured_rps": "measured rps", - "rhaiis_intended_concurrency": "intended concurrency", - "rhaiis_completed_requests": "successful_requests", - "rhaiis_failed_requests": "errored_requests", - "rhaiis_ttft_median": "ttft_median", - "rhaiis_ttft_p95": "ttft_p95", - "rhaiis_ttft_p99": "ttft_p99", - "rhaiis_ttft_p1": "ttft_p1", - "rhaiis_ttft_p999": "ttft_p999", - "rhaiis_ttft_mean": "ttft_mean", - "rhaiis_tpot_median": "tpot_median", - "rhaiis_tpot_p95": "tpot_p95", - "rhaiis_tpot_p99": "tpot_p99", - "rhaiis_tpot_p1": "tpot_p1", - "rhaiis_tpot_p999": "tpot_p999", - "rhaiis_itl_median": "itl_median", - "rhaiis_itl_p95": "itl_p95", - "rhaiis_itl_p99": "itl_p99", - "rhaiis_itl_p1": "itl_p1", - "rhaiis_itl_p999": "itl_p999", - "rhaiis_itl_mean": "itl_mean", - "rhaiis_request_latency_median": "request_latency_median", - "rhaiis_request_latency_min": "request_latency_min", - "rhaiis_request_latency_max": "request_latency_max", - "rhaiis_prompt_token_count_mean": "prompt_token_count_mean", - "rhaiis_prompt_token_count_p99": "prompt_token_count_p99", - "rhaiis_output_token_count_mean": "output_token_count_mean", - "rhaiis_output_token_count_p99": "output_token_count_p99", -} - class RhaiisPlugin(PostProcessingPlugin): def __init__(self) -> None: @@ -118,89 +54,40 @@ def export_kpis_to_csv( output_path: Path, include_header_comments: bool = True, ) -> str: - """Export KPI records to dashboard-format CSV. - - Groups per-rate-point KPIs by (run_path, rate_index) and pivots them - into rows matching the RHAIIS dashboard CSV schema with all fields - populated. - """ + """Export KPI records to the RHAIIS dashboard schema.""" from projects.rhaiis.postprocess.csv_export import FIELDNAMES - # Group KPIs by (run_path, rate_index) - groups: dict[tuple[str, str], dict[str, Any]] = defaultdict(dict) - group_labels: dict[tuple[str, str], dict[str, Any]] = {} - - for kpi in kpi_records: - labels = kpi.get("labels", {}) - run_path = kpi.get("run_path", "") - rate_index = labels.get("rate_index", "0") - key = (run_path, rate_index) - - csv_col = _KPI_TO_CSV_COLUMN.get(kpi.get("kpi_id", "")) - if csv_col: - groups[key][csv_col] = kpi.get("value") - - # Capture the richest set of labels for this group - if key not in group_labels or len(labels) > len(group_labels[key]): - group_labels[key] = labels - - # Build CSV rows - rows = [] - for key in sorted(groups): - metrics = groups[key] - labels = group_labels.get(key, {}) - + def metadata_row(labels: dict[str, Any]) -> dict[str, Any]: acc = labels.get("accelerator", "").upper() cluster_tag = labels.get("cluster_tag", "") model_id = labels.get("hf_model_id", "") tp = labels.get("tensor_parallel_size", "1") - version = labels.get("version", "") - run_name = ( f"{acc}-{cluster_tag}-{model_id}-{tp}" if cluster_tag else f"{acc}-{model_id}-{tp}" ) - - row: dict[str, Any] = dict.fromkeys(FIELDNAMES, "") - # Populate all metric columns from KPI values - row.update(metrics) - # Convert latency columns from seconds back to ms for dashboard compat - for col in _SECONDS_TO_MS_COLUMNS: - val = row.get(col) - if val not in ("", None): - try: - row[col] = float(val) * 1000.0 - except (TypeError, ValueError): - pass - # Populate metadata columns from labels - row["run"] = run_name - row["accelerator"] = acc - row["model"] = model_id - row["version"] = version - row["TP"] = tp - row["prompt toks"] = labels.get("prompt_toks", "") - row["output toks"] = labels.get("output_toks", "") - # intended concurrency is populated from the rhaiis_intended_concurrency KPI; - # only fall back to labels if the KPI didn't set it - if not row.get("intended concurrency"): - row["intended concurrency"] = labels.get("intended_concurrency", "") - row["image_tag"] = labels.get("image_tag", "") - row["runtime_args"] = labels.get("runtime_args", "") - row["uuid"] = labels.get("run_uuid", "") - row["guidellm_start_time_ms"] = labels.get("guidellm_start_time_ms", "") - row["guidellm_end_time_ms"] = labels.get("guidellm_end_time_ms", "") - row["guidellm_version"] = labels.get("guidellm_version", "") - row["mlflow_run_id"] = labels.get("mlflow_run_id", "") - row["mlflow_experiment_id"] = labels.get("mlflow_experiment_id", "") - rows.append(row) - - output_path.parent.mkdir(parents=True, exist_ok=True) - with open(output_path, "w", newline="", encoding="utf-8") as f: - writer = csv.DictWriter(f, fieldnames=FIELDNAMES, extrasaction="ignore") - writer.writeheader() - writer.writerows(rows) - - logger.info("Exported %d dashboard CSV rows to %s", len(rows), output_path) - return str(output_path) + return { + "run": run_name, + "accelerator": acc, + "model": model_id, + "version": labels.get("version", ""), + "TP": tp, + "prompt toks": labels.get("prompt_toks", ""), + "output toks": labels.get("output_toks", ""), + "image_tag": labels.get("image_tag", ""), + "runtime_args": labels.get("runtime_args", ""), + "uuid": labels.get("run_uuid", ""), + "guidellm_start_time_ms": labels.get("guidellm_start_time_ms", ""), + "guidellm_end_time_ms": labels.get("guidellm_end_time_ms", ""), + "guidellm_version": labels.get("guidellm_version", ""), + } + + return export_dashboard_kpis_to_csv( + kpi_records, + output_path, + prefix="rhaiis", + fieldnames=FIELDNAMES, + metadata_row=metadata_row, + ) def build_ai_data_payload(self, model: UnifiedRunModel) -> dict[str, Any]: return {} From 6dc0cf61402e068aa7c84e35e1e6cbe5b488d384 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Tue, 11 Aug 2026 11:04:00 +0100 Subject: [PATCH 02/10] chore: Review comments Signed-off-by: Alberto Perdomo --- projects/guidellm/postprocess/guidellm/dashboard.py | 2 +- projects/llm_d/orchestration/config.yaml | 2 +- projects/llm_d/postprocess/llm_d/__init__.py | 1 + projects/llm_d/postprocess/{ => llm_d}/plugin.py | 2 -- projects/llm_d/tests/test_postprocess_csv.py | 8 ++++---- 5 files changed, 7 insertions(+), 8 deletions(-) create mode 100644 projects/llm_d/postprocess/llm_d/__init__.py rename projects/llm_d/postprocess/{ => llm_d}/plugin.py (99%) diff --git a/projects/guidellm/postprocess/guidellm/dashboard.py b/projects/guidellm/postprocess/guidellm/dashboard.py index 3f036bffa..670f665ab 100644 --- a/projects/guidellm/postprocess/guidellm/dashboard.py +++ b/projects/guidellm/postprocess/guidellm/dashboard.py @@ -427,7 +427,7 @@ def export_dashboard_kpis_to_csv( labels_by_group[key] = labels rows: list[dict[str, Any]] = [] - for key in sorted(groups): + for key in sorted(groups, key=lambda k: (k[0], int(k[1]) if k[1].isdigit() else k[1])): metrics = groups[key] labels = labels_by_group.get(key, {}) row: dict[str, Any] = dict.fromkeys(fieldnames, "") diff --git a/projects/llm_d/orchestration/config.yaml b/projects/llm_d/orchestration/config.yaml index 9a6d70a05..15da6bfa2 100644 --- a/projects/llm_d/orchestration/config.yaml +++ b/projects/llm_d/orchestration/config.yaml @@ -41,7 +41,7 @@ caliper: postprocess: enabled: true artifacts_dir: null - plugin_module: projects.llm_d.postprocess.plugin + plugin_module: projects.llm_d.postprocess.llm_d.plugin postprocess_config: null filtering: include_labels: [] diff --git a/projects/llm_d/postprocess/llm_d/__init__.py b/projects/llm_d/postprocess/llm_d/__init__.py new file mode 100644 index 000000000..2a59d0151 --- /dev/null +++ b/projects/llm_d/postprocess/llm_d/__init__.py @@ -0,0 +1 @@ +"""llm-d specific Caliper post-processing plugin.""" diff --git a/projects/llm_d/postprocess/plugin.py b/projects/llm_d/postprocess/llm_d/plugin.py similarity index 99% rename from projects/llm_d/postprocess/plugin.py rename to projects/llm_d/postprocess/llm_d/plugin.py index 9842ac266..019911fb5 100644 --- a/projects/llm_d/postprocess/plugin.py +++ b/projects/llm_d/postprocess/llm_d/plugin.py @@ -22,7 +22,6 @@ enrich_guidellm_parse_result, export_dashboard_kpis_to_csv, normalize_product_version, - validate_dashboard_fieldnames, ) from projects.guidellm.postprocess.guidellm.parsing import GuideLLMKpiHandler, GuideLLMParser from projects.llm_d.orchestration.render_inference_service import _build_vllm_args @@ -81,7 +80,6 @@ "guidellm_version", "notes", ] -validate_dashboard_fieldnames(FIELDNAMES) class LlmDGuideLLMPlugin(PostProcessingPlugin): diff --git a/projects/llm_d/tests/test_postprocess_csv.py b/projects/llm_d/tests/test_postprocess_csv.py index e2b29cb2d..c68473b1c 100644 --- a/projects/llm_d/tests/test_postprocess_csv.py +++ b/projects/llm_d/tests/test_postprocess_csv.py @@ -7,7 +7,7 @@ from projects.caliper.engine.model import TestBaseNode as CaliperTestBaseNode from projects.caliper.engine.model import UnifiedRunModel -from projects.llm_d.postprocess.plugin import FIELDNAMES, LlmDGuideLLMPlugin +from projects.llm_d.postprocess.llm_d.plugin import FIELDNAMES, LlmDGuideLLMPlugin def _metric(**values): @@ -114,7 +114,7 @@ def test_llmd_plugin_exports_dashboard_compatible_csv(tmp_path): plugin = LlmDGuideLLMPlugin() parsed = plugin.parse([node]) model = UnifiedRunModel( - plugin_module="projects.llm_d.postprocess.plugin", + plugin_module="projects.llm_d.postprocess.llm_d.plugin", base_directory=str(tmp_path), test_nodes=[node], unified_result_records=parsed.records, @@ -193,7 +193,7 @@ def test_llmd_plugin_recovers_deployment_metadata_from_config(tmp_path): test_labels={"labels": {}}, ) - from projects.llm_d.postprocess.plugin import _extract_deployment_metadata + from projects.llm_d.postprocess.llm_d.plugin import _extract_deployment_metadata metadata = _extract_deployment_metadata(node) @@ -242,6 +242,6 @@ def test_llmd_plugin_infers_accelerator_from_serving_pod_node(tmp_path): test_labels={"labels": {}}, ) - from projects.llm_d.postprocess.plugin import _extract_deployment_metadata + from projects.llm_d.postprocess.llm_d.plugin import _extract_deployment_metadata assert _extract_deployment_metadata(node)["gpu_type"] == "H200" From ee95a5a11746ba0250acbb678d4877ee4b5c5785 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Tue, 11 Aug 2026 14:41:34 +0100 Subject: [PATCH 03/10] fix: Ascending concurrency order Signed-off-by: Alberto Perdomo --- projects/llm_d/orchestration/config.d/workloads.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/projects/llm_d/orchestration/config.d/workloads.yaml b/projects/llm_d/orchestration/config.d/workloads.yaml index 09bed7c4b..835b058fb 100644 --- a/projects/llm_d/orchestration/config.d/workloads.yaml +++ b/projects/llm_d/orchestration/config.d/workloads.yaml @@ -31,7 +31,7 @@ benchmarks: args: backend_type: openai_http rate_type: concurrent - rate: [300, 200, 100, 50, 1] + rate: [1, 50, 100, 200, 300] data: prompt_tokens=1000,output_tokens=1000 max_seconds: 600 @@ -42,7 +42,7 @@ benchmarks: args: backend_type: openai_http rate_type: concurrent - rate: [300, 200, 100, 50, 1] + rate: [1, 50, 100, 200, 300] data: prompt_tokens=8000,prompt_tokens_stdev=8500,prompt_tokens_min=50,prompt_tokens_max=30000,output_tokens=800,output_tokens_stdev=1500,output_tokens_min=20,output_tokens_max=8000 max_seconds: 600 From aba278b20328c30b81cfeabcc765128c3e3e2b5f Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Wed, 12 Aug 2026 08:27:07 +0100 Subject: [PATCH 04/10] fix: llm-d post processing plugin Signed-off-by: Alberto Perdomo --- projects/llm_d/postprocess/llm_d/plugin.py | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/projects/llm_d/postprocess/llm_d/plugin.py b/projects/llm_d/postprocess/llm_d/plugin.py index 019911fb5..5a3693a18 100644 --- a/projects/llm_d/postprocess/llm_d/plugin.py +++ b/projects/llm_d/postprocess/llm_d/plugin.py @@ -14,7 +14,6 @@ TestBaseNode, UnifiedRunModel, ) -from projects.guidellm.postprocess.guidellm.ai_eval import GuideLLMAIEvaluator from projects.guidellm.postprocess.guidellm.dashboard import ( compute_dashboard_kpis, dashboard_kpi_catalog, @@ -23,7 +22,8 @@ export_dashboard_kpis_to_csv, normalize_product_version, ) -from projects.guidellm.postprocess.guidellm.parsing import GuideLLMKpiHandler, GuideLLMParser +from projects.guidellm.postprocess.guidellm.plugin import GuideLLMPlugin +from projects.guidellm.postprocess.guidellm.plugin import analysis_config as analysis_config from projects.llm_d.orchestration.render_inference_service import _build_vllm_args from projects.llm_d.orchestration.runtime_config import deep_merge @@ -82,16 +82,11 @@ ] -class LlmDGuideLLMPlugin(PostProcessingPlugin): +class LlmDGuideLLMPlugin(GuideLLMPlugin): """Keep generic GuideLLM outputs and add the llm-d dashboard projection.""" - def __init__(self) -> None: - self.parser = GuideLLMParser() - self.kpi_handler = GuideLLMKpiHandler() - self.ai_evaluator = GuideLLMAIEvaluator() - def parse(self, nodes: list[TestBaseNode]) -> ParseResult: - parsed = enrich_guidellm_parse_result(self.parser.parse(nodes), nodes) + parsed = enrich_guidellm_parse_result(super().parse(nodes), nodes) nodes_by_path = {str(node.test_path): node for node in nodes} records = [] for record in parsed.records: @@ -110,7 +105,7 @@ def kpi_catalog(self) -> list[dict[str, Any]]: return self.kpi_handler.get_catalog() + dashboard_kpi_catalog(prefix="llmd") def compute_kpis(self, model: UnifiedRunModel) -> list[dict[str, Any]]: - return self.kpi_handler.compute_kpis(model) + compute_dashboard_kpis(model, prefix="llmd") + return super().compute_kpis(model) + compute_dashboard_kpis(model, prefix="llmd") def export_kpis_to_csv( self, @@ -161,9 +156,6 @@ def metadata_row(labels: dict[str, Any]) -> dict[str, Any]: metadata_row=metadata_row, ) - def build_ai_data_payload(self, model: UnifiedRunModel) -> dict[str, Any]: - return self.ai_evaluator.build_payload(model, self) - def get_plugin() -> PostProcessingPlugin: return LlmDGuideLLMPlugin() From b3823559ce042548ba2f4476e3271fe923aecf04 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Wed, 12 Aug 2026 10:01:52 +0100 Subject: [PATCH 05/10] chore: Minor fixes Signed-off-by: Alberto Perdomo --- projects/guidellm/postprocess/guidellm/dashboard.py | 1 + projects/guidellm/postprocess/guidellm/parsing/parsers.py | 7 +++++-- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/projects/guidellm/postprocess/guidellm/dashboard.py b/projects/guidellm/postprocess/guidellm/dashboard.py index 670f665ab..b4149e939 100644 --- a/projects/guidellm/postprocess/guidellm/dashboard.py +++ b/projects/guidellm/postprocess/guidellm/dashboard.py @@ -214,6 +214,7 @@ def _extract_dashboard_metrics(node: TestBaseNode) -> tuple[dict[str, Any], dict .get("requests_per_second", {}) .get("successful", {}) .get("mean", 0) + or 0 ) ) data_values = args.get("data", []) if isinstance(args, dict) else [] diff --git a/projects/guidellm/postprocess/guidellm/parsing/parsers.py b/projects/guidellm/postprocess/guidellm/parsing/parsers.py index 6602d7c5c..f5a3975a9 100644 --- a/projects/guidellm/postprocess/guidellm/parsing/parsers.py +++ b/projects/guidellm/postprocess/guidellm/parsing/parsers.py @@ -153,8 +153,11 @@ def extract_fields_from_llmisvc(self, file_path: Path) -> dict[str, Any]: if annotation_value: product_version = parse_product_version_from_annotation(annotation_value) if product_version: - result["product_version"] = normalize_product_version(product_version) - logger.info(f"Extracted product_version '{product_version}' from {file_path}") + normalized = normalize_product_version(product_version) + result["product_version"] = normalized + logger.info( + f"Extracted product_version '{product_version}' (normalized to '{normalized}') from {file_path}" + ) # Extract deployment profile from forge annotation deployment_profile = extract_field_by_jsonpath( From bc837be8bbcd8dfee38d0f85e4db5e5f35c5f606 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Wed, 12 Aug 2026 10:23:35 +0100 Subject: [PATCH 06/10] chore: Lazy imports Signed-off-by: Alberto Perdomo --- .../guidellm/postprocess/guidellm/plugin.py | 87 +++++++++++-------- 1 file changed, 51 insertions(+), 36 deletions(-) diff --git a/projects/guidellm/postprocess/guidellm/plugin.py b/projects/guidellm/postprocess/guidellm/plugin.py index 4f8f7cedd..930ed3732 100644 --- a/projects/guidellm/postprocess/guidellm/plugin.py +++ b/projects/guidellm/postprocess/guidellm/plugin.py @@ -16,11 +16,6 @@ from .ai_eval import GuideLLMAIEvaluator from .parsing import GuideLLMKpiHandler, GuideLLMParser -from .plotting.kpi_report import generate_kpi_report -from .plotting.performance_analysis import ( - generate_comprehensive_performance_report, - generate_deployment_profile_report, -) logger = logging.getLogger(__name__) @@ -35,36 +30,50 @@ ) -# Plot registry - maps report names to their generator functions and parameters -PLOT_REGISTRY = { - "report_performance_analysis": { - "function": generate_comprehensive_performance_report, - "type": "report", - "kwargs": { - "report_number": 0, - "report_title": "GuideLLM Performance Analysis", - }, - "description": "comprehensive performance analysis report (recommended)", - }, - "report_kpi_summary": { - "function": generate_kpi_report, - "type": "report", - "kwargs": { - "report_number": 1, - "report_title": "GuideLLM KPI Summary", - }, - "description": "KPI summary with test conditions and metrics", - }, - "report_deployment_profile": { - "function": generate_deployment_profile_report, - "type": "report", - "kwargs": { - "report_number": 2, - "report_title": "GuideLLM Deployment Profile Analysis", - }, - "description": "performance analysis comparing different product versions/models under identical test conditions", - }, -} +PLOT_REGISTRY: dict[str, dict[str, Any]] = {} + + +def _ensure_plot_registry() -> None: + """Populate PLOT_REGISTRY on first use, keeping pandas out of module load.""" + if PLOT_REGISTRY: + return + from .plotting.kpi_report import generate_kpi_report + from .plotting.performance_analysis import ( + generate_comprehensive_performance_report, + generate_deployment_profile_report, + ) + + PLOT_REGISTRY.update( + { + "report_performance_analysis": { + "function": generate_comprehensive_performance_report, + "type": "report", + "kwargs": { + "report_number": 0, + "report_title": "GuideLLM Performance Analysis", + }, + "description": "comprehensive performance analysis report (recommended)", + }, + "report_kpi_summary": { + "function": generate_kpi_report, + "type": "report", + "kwargs": { + "report_number": 1, + "report_title": "GuideLLM KPI Summary", + }, + "description": "KPI summary with test conditions and metrics", + }, + "report_deployment_profile": { + "function": generate_deployment_profile_report, + "type": "report", + "kwargs": { + "report_number": 2, + "report_title": "GuideLLM Deployment Profile Analysis", + }, + "description": "performance analysis comparing different product versions/models under identical test conditions", + }, + } + ) class GuideLLMPlugin(PostProcessingPlugin): @@ -83,6 +92,7 @@ def parse(self, nodes: list[TestBaseNode]) -> ParseResult: def get_available_reports(self) -> dict[str, dict[str, str]]: """Get a structured dictionary of available reports and plots with their types and descriptions.""" + _ensure_plot_registry() return { name: { "type": config["type"], @@ -93,6 +103,7 @@ def get_available_reports(self) -> dict[str, dict[str, str]]: def get_available_reports_by_type(self) -> dict[str, dict[str, str]]: """Get reports and plots grouped by type.""" + _ensure_plot_registry() result = {"reports": {}, "plots": {}} for name, config in PLOT_REGISTRY.items(): type_key = "reports" if config["type"] == "report" else "plots" @@ -101,6 +112,7 @@ def get_available_reports_by_type(self) -> dict[str, dict[str, str]]: def get_reports_only(self) -> dict[str, str]: """Get only comprehensive reports (HTML files with multiple plots).""" + _ensure_plot_registry() return { name: config["description"] for name, config in PLOT_REGISTRY.items() @@ -109,6 +121,7 @@ def get_reports_only(self) -> dict[str, str]: def get_plots_only(self) -> dict[str, str]: """Get only individual plots (single visualizations).""" + _ensure_plot_registry() return { name: config["description"] for name, config in PLOT_REGISTRY.items() @@ -116,7 +129,7 @@ def get_plots_only(self) -> dict[str, str]: } @staticmethod - def register_plot( + def register_plot( # noqa: FBT001 name: str, function: callable, description: str, type_: str = "plot", **kwargs ) -> None: """Register a new plot or report generator function. @@ -137,6 +150,7 @@ def register_plot( report_number=10 ) """ + _ensure_plot_registry() PLOT_REGISTRY[name] = { "function": function, "type": type_, @@ -153,6 +167,7 @@ def visualize( visualize_config: dict[str, Any] | None, ) -> list[str]: """Generate visualization reports for GuideLLM benchmarks.""" + _ensure_plot_registry() output_dir.mkdir(parents=True, exist_ok=True) paths: list[str] = [] wanted = frozenset(report_ids or ()) From 3671024e8acc6ed5e8edcaedf3619220d919091d Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Wed, 12 Aug 2026 10:35:48 +0100 Subject: [PATCH 07/10] fix: Concurrency order in tests Signed-off-by: Alberto Perdomo --- projects/llm_d/tests/test_profiles.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/projects/llm_d/tests/test_profiles.py b/projects/llm_d/tests/test_profiles.py index a4c3d8189..a54a315ea 100644 --- a/projects/llm_d/tests/test_profiles.py +++ b/projects/llm_d/tests/test_profiles.py @@ -107,7 +107,7 @@ def test_benchmark_workloads_are_available() -> None: assert benchmark["timeout_seconds"] == 3600 assert multi_turn["timeout_seconds"] == 7200 - assert concurrent["args"]["rate"] == [300, 200, 100, 50, 1] + assert concurrent["args"]["rate"] == [1, 50, 100, 200, 300] assert heavy["args"]["max_seconds"] == 600 assert "prompt_tokens_stdev=8500" in heavy["args"]["data"] assert "output_tokens_max=8000" in heavy["args"]["data"] From 8679bdd25880210633db161245cf34a95783efa2 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Wed, 12 Aug 2026 21:15:46 +0100 Subject: [PATCH 08/10] feat: GuideLLM PVC storageClass support Signed-off-by: Alberto Perdomo --- projects/guidellm/toolbox/run_guidellm_benchmark/main.py | 2 ++ .../templates/guidellm_pvc.yaml.j2 | 7 +++++-- .../guidellm/toolbox/run_guidellm_benchmark/utils.py | 9 ++++++++- projects/llm_d/orchestration/config.d/workloads.yaml | 1 + projects/llm_d/orchestration/runtime_config.py | 4 ++-- projects/llm_d/orchestration/test_phase.py | 1 + 6 files changed, 19 insertions(+), 5 deletions(-) diff --git a/projects/guidellm/toolbox/run_guidellm_benchmark/main.py b/projects/guidellm/toolbox/run_guidellm_benchmark/main.py index af3501f57..3308bbb11 100644 --- a/projects/guidellm/toolbox/run_guidellm_benchmark/main.py +++ b/projects/guidellm/toolbox/run_guidellm_benchmark/main.py @@ -54,6 +54,7 @@ def run( image: str = "ghcr.io/vllm-project/guidellm:v0.6.0", timeout: int = 900, pvc_size: str = "1Gi", + pvc_storage_class: str | None = None, guidellm_args: list[str] | None = None, hf_token_secret: str = "", fs_group: int | None = None, @@ -189,6 +190,7 @@ def create_guidellm_resources_task(args, ctx): namespace=ctx.target_namespace, name=ctx.benchmark_name, pvc_size=args.pvc_size, + pvc_storage_class=args.pvc_storage_class, owner_reference=owner_reference, ), ) diff --git a/projects/guidellm/toolbox/run_guidellm_benchmark/templates/guidellm_pvc.yaml.j2 b/projects/guidellm/toolbox/run_guidellm_benchmark/templates/guidellm_pvc.yaml.j2 index ab46192a7..cfc6b4d72 100644 --- a/projects/guidellm/toolbox/run_guidellm_benchmark/templates/guidellm_pvc.yaml.j2 +++ b/projects/guidellm/toolbox/run_guidellm_benchmark/templates/guidellm_pvc.yaml.j2 @@ -9,7 +9,10 @@ metadata: forge.openshift.io/project: llm_d spec: accessModes: - - ReadWriteOnce + - {{ "ReadWriteMany" if pvc_storage_class else "ReadWriteOnce" }} +{% if pvc_storage_class %} + storageClassName: {{ pvc_storage_class }} +{% endif %} resources: requests: - storage: {{ pvc_size }} \ No newline at end of file + storage: {{ pvc_size }} diff --git a/projects/guidellm/toolbox/run_guidellm_benchmark/utils.py b/projects/guidellm/toolbox/run_guidellm_benchmark/utils.py index f63ab8280..2f85729a6 100644 --- a/projects/guidellm/toolbox/run_guidellm_benchmark/utils.py +++ b/projects/guidellm/toolbox/run_guidellm_benchmark/utils.py @@ -147,7 +147,12 @@ def _build_multi_run_script(*, endpoint_url: str, runs: list[GuideLLMRun]) -> st def render_guidellm_pvc_from_parts( - *, namespace: str, name: str, pvc_size: str, owner_reference: dict[str, Any] | None = None + *, + namespace: str, + name: str, + pvc_size: str, + pvc_storage_class: str | None = None, + owner_reference: dict[str, Any] | None = None, ) -> dict[str, Any]: """Render a GuideLL-M PVC manifest from individual components. @@ -155,6 +160,7 @@ def render_guidellm_pvc_from_parts( namespace: Target namespace name: Name of the benchmark job and PVC pvc_size: Size of the PVC + pvc_storage_class: Optional storage class name for the PVC owner_reference: Optional owner reference to set (e.g., for job ownership) Returns: @@ -166,6 +172,7 @@ def render_guidellm_pvc_from_parts( "namespace": namespace, "name": name, "pvc_size": pvc_size, + "pvc_storage_class": pvc_storage_class, }, ) manifest = yaml.safe_load(rendered_yaml) diff --git a/projects/llm_d/orchestration/config.d/workloads.yaml b/projects/llm_d/orchestration/config.d/workloads.yaml index 835b058fb..60e96711b 100644 --- a/projects/llm_d/orchestration/config.d/workloads.yaml +++ b/projects/llm_d/orchestration/config.d/workloads.yaml @@ -1,6 +1,7 @@ job_name: guidellm-benchmark image: ghcr.io/vllm-project/guidellm:v0.5.4 pvc_size: 1Gi +pvc_storage_class: null timeout_seconds: 3600 args: {} diff --git a/projects/llm_d/orchestration/runtime_config.py b/projects/llm_d/orchestration/runtime_config.py index ab2b8f72a..1fb0aa54a 100644 --- a/projects/llm_d/orchestration/runtime_config.py +++ b/projects/llm_d/orchestration/runtime_config.py @@ -355,7 +355,7 @@ def _resolve_benchmark_config(benchmark_name: str) -> dict[str, Any]: ) workload_defaults = copy.deepcopy(config.project.get_config("workloads", print=False)) - default_keys = ("job_name", "image", "pvc_size", "timeout_seconds") + default_keys = ("job_name", "image", "pvc_size", "pvc_storage_class", "timeout_seconds") for key in default_keys: if key in workload_defaults and key not in benchmark: benchmark[key] = workload_defaults[key] @@ -401,7 +401,7 @@ def get_workload_config() -> dict[str, Any] | None: return None # Apply same merging logic as _resolve_benchmark_config - default_keys = ("job_name", "image", "pvc_size", "timeout_seconds") + default_keys = ("job_name", "image", "pvc_size", "pvc_storage_class", "timeout_seconds") for key in default_keys: if key in workload_defaults and key not in default_benchmark: default_benchmark[key] = workload_defaults[key] diff --git a/projects/llm_d/orchestration/test_phase.py b/projects/llm_d/orchestration/test_phase.py index d09d0a56e..ee2eb3550 100644 --- a/projects/llm_d/orchestration/test_phase.py +++ b/projects/llm_d/orchestration/test_phase.py @@ -725,6 +725,7 @@ def run_guidellm_benchmark(*, endpoint_url: str) -> None: image=benchmark.get("image"), timeout=benchmark.get("timeout_seconds"), pvc_size=benchmark.get("pvc_size"), + pvc_storage_class=benchmark.get("pvc_storage_class"), guidellm_args=guidellm_args, ) From f0fd525497f84eddee97a0c77b5b5b05bbe98103 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Tue, 18 Aug 2026 10:21:32 +0100 Subject: [PATCH 09/10] feat: Enable early MLFlow run Signed-off-by: Alberto Perdomo --- .../postprocess/guidellm/dashboard.py | 2 ++ projects/llm_d/orchestration/test_phase.py | 27 ++++++++++++++++--- projects/llm_d/postprocess/llm_d/plugin.py | 10 +++++++ projects/rhaiis/postprocess/plugin.py | 14 +++++++++- 4 files changed, 49 insertions(+), 4 deletions(-) diff --git a/projects/guidellm/postprocess/guidellm/dashboard.py b/projects/guidellm/postprocess/guidellm/dashboard.py index b4149e939..db8af46ac 100644 --- a/projects/guidellm/postprocess/guidellm/dashboard.py +++ b/projects/guidellm/postprocess/guidellm/dashboard.py @@ -91,6 +91,8 @@ "image_tag", "router_config", "gpu_type", + "mlflow_run_id", + "mlflow_experiment_id", } ) diff --git a/projects/llm_d/orchestration/test_phase.py b/projects/llm_d/orchestration/test_phase.py index ee2eb3550..43b0c251a 100644 --- a/projects/llm_d/orchestration/test_phase.py +++ b/projects/llm_d/orchestration/test_phase.py @@ -176,7 +176,9 @@ def extract_kpi_labels_from_config() -> dict[str, str]: return kpi_labels -def create_test_labels() -> None: +def create_test_labels( + mlflow_destination: dict[str, str] | None = None, +) -> None: """Create __test_labels__.yaml with model name and guidellm configuration.""" model_name = runtime_config.get_model_name() @@ -194,7 +196,12 @@ def create_test_labels() -> None: # Extract kpi_labels from config kpi_labels = extract_kpi_labels_from_config() - write_test_labels(env.ARTIFACT_DIR, labels, kpi_labels=kpi_labels if kpi_labels else None) + write_test_labels( + env.ARTIFACT_DIR, + labels, + kpi_labels=kpi_labels if kpi_labels else None, + mlflow_destination=mlflow_destination, + ) logger.info("Created test labels: %s", labels) # Dump config.project to config.yaml @@ -374,6 +381,20 @@ def do_test() -> int: # Delete all existing resources if configured cleanup_existing_resources(namespace) + try: + from projects.caliper.orchestration.export import precreate_mlflow_run_if_configured + + mlflow_destination = precreate_mlflow_run_if_configured() + except Exception: + logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) + mlflow_destination = None + + if mlflow_destination: + import yaml as _yaml + + mlflow_marker = env.ARTIFACT_DIR / "_mlflow_destination.yaml" + mlflow_marker.write_text(_yaml.safe_dump(mlflow_destination, sort_keys=False)) + endpoint_url: str | None = None primary_exc: tuple[type[BaseException], BaseException, Any] | None = None finalizer_exc: tuple[type[BaseException], BaseException, Any] | None = None @@ -381,7 +402,7 @@ def do_test() -> int: actual_llmisvc_name = "llmisvc-na-not-computed" try: # Create test labels with actual model and profile information - create_test_labels() + create_test_labels(mlflow_destination=mlflow_destination) # Generate the LLMInferenceService name before deployment # so we have it available even if deployment fails diff --git a/projects/llm_d/postprocess/llm_d/plugin.py b/projects/llm_d/postprocess/llm_d/plugin.py index 5a3693a18..b5893e54a 100644 --- a/projects/llm_d/postprocess/llm_d/plugin.py +++ b/projects/llm_d/postprocess/llm_d/plugin.py @@ -78,6 +78,8 @@ "guidellm_end_time_ms", "image_tag", "guidellm_version", + "mlflow_run_id", + "mlflow_experiment_id", "notes", ] @@ -96,6 +98,12 @@ def parse(self, nodes: list[TestBaseNode]) -> ParseResult: hf_model_id = test_labels.get("model_name") if hf_model_id: record.metrics["hf_model_id"] = hf_model_id + mlflow_dest = node.test_labels.get("mlflow_destination", {}) if node else {} + if mlflow_dest: + record.metrics.setdefault("mlflow_run_id", mlflow_dest.get("run_id", "")) + record.metrics.setdefault( + "mlflow_experiment_id", mlflow_dest.get("experiment_id", "") + ) for key, value in deployment_metadata.items(): record.metrics.setdefault(key, value) records.append(record) @@ -145,6 +153,8 @@ def metadata_row(labels: dict[str, Any]) -> dict[str, Any]: "guidellm_end_time_ms": labels.get("guidellm_end_time_ms", ""), "image_tag": labels.get("image_tag", ""), "guidellm_version": labels.get("guidellm_version", ""), + "mlflow_run_id": labels.get("mlflow_run_id", ""), + "mlflow_experiment_id": labels.get("mlflow_experiment_id", ""), "notes": labels.get("notes", ""), } diff --git a/projects/rhaiis/postprocess/plugin.py b/projects/rhaiis/postprocess/plugin.py index 531cc5f80..eca4fd400 100644 --- a/projects/rhaiis/postprocess/plugin.py +++ b/projects/rhaiis/postprocess/plugin.py @@ -21,7 +21,17 @@ def __init__(self) -> None: self.kpi_handler = RhaiisKpiHandler() def parse(self, nodes: list[TestBaseNode]) -> ParseResult: - return self.parser.parse(nodes) + parsed = self.parser.parse(nodes) + nodes_by_path = {str(node.test_path): node for node in nodes} + for record in parsed.records: + node = nodes_by_path.get(record.test_base_path) + mlflow_dest = node.test_labels.get("mlflow_destination", {}) if node else {} + if mlflow_dest: + record.metrics.setdefault("mlflow_run_id", mlflow_dest.get("run_id", "")) + record.metrics.setdefault( + "mlflow_experiment_id", mlflow_dest.get("experiment_id", "") + ) + return parsed def get_available_reports(self) -> dict[str, dict[str, str]]: return {} @@ -79,6 +89,8 @@ def metadata_row(labels: dict[str, Any]) -> dict[str, Any]: "guidellm_start_time_ms": labels.get("guidellm_start_time_ms", ""), "guidellm_end_time_ms": labels.get("guidellm_end_time_ms", ""), "guidellm_version": labels.get("guidellm_version", ""), + "mlflow_run_id": labels.get("mlflow_run_id", ""), + "mlflow_experiment_id": labels.get("mlflow_experiment_id", ""), } return export_dashboard_kpis_to_csv( From bbac826d816dee45e4005aaf1ff5510ea4056c03 Mon Sep 17 00:00:00 2001 From: Alberto Perdomo Date: Tue, 18 Aug 2026 11:16:20 +0100 Subject: [PATCH 10/10] fix: Remove duplicate implementation Signed-off-by: Alberto Perdomo --- projects/llm_d/orchestration/test_phase.py | 6 --- projects/rhaiis/orchestration/test_phase.py | 6 --- projects/rhaiis/postprocess/regression.py | 49 +-------------------- 3 files changed, 1 insertion(+), 60 deletions(-) diff --git a/projects/llm_d/orchestration/test_phase.py b/projects/llm_d/orchestration/test_phase.py index 43b0c251a..9872a0e20 100644 --- a/projects/llm_d/orchestration/test_phase.py +++ b/projects/llm_d/orchestration/test_phase.py @@ -389,12 +389,6 @@ def do_test() -> int: logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) mlflow_destination = None - if mlflow_destination: - import yaml as _yaml - - mlflow_marker = env.ARTIFACT_DIR / "_mlflow_destination.yaml" - mlflow_marker.write_text(_yaml.safe_dump(mlflow_destination, sort_keys=False)) - endpoint_url: str | None = None primary_exc: tuple[type[BaseException], BaseException, Any] | None = None finalizer_exc: tuple[type[BaseException], BaseException, Any] | None = None diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 20c93af62..fe94a409d 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -201,12 +201,6 @@ def _run_test( logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) mlflow_destination = None - if mlflow_destination: - import yaml as _yaml - - mlflow_marker = env.ARTIFACT_DIR / "_mlflow_destination.yaml" - mlflow_marker.write_text(_yaml.safe_dump(mlflow_destination, sort_keys=False)) - try: isvc_labels = {"opendatahub.io/dashboard": "true"} if profiler_enabled and engine == "vllm": diff --git a/projects/rhaiis/postprocess/regression.py b/projects/rhaiis/postprocess/regression.py index dfb735626..939c4feee 100644 --- a/projects/rhaiis/postprocess/regression.py +++ b/projects/rhaiis/postprocess/regression.py @@ -337,56 +337,9 @@ def _build_mlflow_run_url() -> str: from projects.caliper.orchestration.export import build_mlflow_run_url_from_config try: - url = build_mlflow_run_url_from_config() - if url: - return url + return build_mlflow_run_url_from_config() or "" except Exception: logger.warning("Failed to build MLflow run URL from test labels", exc_info=True) - - # Fallback: check pre-created marker written before deployment - try: - from pathlib import Path - from urllib.parse import quote - - import yaml - - from projects.caliper.public.file_export import load_mlflow_secrets_yaml - from projects.core.library import config as _cfg - from projects.core.library import env - from projects.core.library import vault as vault_lib - - marker = Path(env.ARTIFACT_DIR) / "_mlflow_destination.yaml" - if not marker.exists(): - return "" - dest = yaml.safe_load(marker.read_text()) - if not isinstance(dest, dict) or not dest.get("run_id"): - return "" - - run_id = dest["run_id"] - experiment_id = dest.get("experiment_id", "") - if not experiment_id: - return "" - - vault_name = _cfg.project.get_config("caliper.export.backend.mlflow.secrets.vault.name", "") - vault_key = _cfg.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", "" - ) - if not vault_name or not vault_key: - return "" - secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) - if not secrets_path or not secrets_path.exists(): - return "" - - secrets_data = load_mlflow_secrets_yaml(secrets_path) - tracking_uri = secrets_data.get("tracking_uri", "").rstrip("/") - if not tracking_uri.startswith(("http://", "https://")): - return "" - - workspace = _cfg.project.get_config("caliper.export.backend.mlflow.config.workspace", "") - qs = f"?workspace={quote(workspace, safe='')}" if workspace else "" - return f"{tracking_uri}/#/experiments/{experiment_id}/runs/{run_id}/artifacts{qs}" - except Exception: - logger.warning("Failed to build MLflow run URL from marker", exc_info=True) return ""