From 6e9dd4437a85d07a4786642ad401dc18b0db259f Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Thu, 30 Jul 2026 23:03:36 -0400 Subject: [PATCH 01/24] feat: pre-create MLflow run to embed run ID in dashboard CSV and Slack notifications Pre-create the MLflow run during the test step (before CSV generation) so the run_id and experiment_id are available for: - Dashboard CSV columns (mlflow_run_id, mlflow_experiment_id) - Slack notifications (MLflow run URL constructed at runtime from vault) The export step resumes the pre-created run instead of creating a new one, via the existing run_id parameter on both single-run and multi-run paths. Co-authored-by: Cursor --- .gitignore | 4 +- .../file_export/artifacts_export_run.py | 2 + .../engine/file_export/mlflow_backend.py | 5 +- projects/caliper/orchestration/export.py | 1 + .../rhaiis/orchestration/config.d/rhaiis.yaml | 2 +- projects/rhaiis/orchestration/test_phase.py | 82 +++++++++++++++++++ projects/rhaiis/postprocess/csv_export.py | 4 + projects/rhaiis/postprocess/plugin.py | 2 + projects/rhaiis/postprocess/regression.py | 57 +++++++++++++ 9 files changed, 156 insertions(+), 3 deletions(-) diff --git a/.gitignore b/.gitignore index 80ea2d510..7a476c764 100644 --- a/.gitignore +++ b/.gitignore @@ -162,4 +162,6 @@ Thumbs.db temp/ # FORGE launcher configuration (personal settings) -projects/core/launcher/launcher_config.yaml \ No newline at end of file +projects/core/launcher/launcher_config.yaml + +fournos-job-*.yaml \ No newline at end of file diff --git a/projects/caliper/engine/file_export/artifacts_export_run.py b/projects/caliper/engine/file_export/artifacts_export_run.py index 049a69a9b..bf95beb81 100644 --- a/projects/caliper/engine/file_export/artifacts_export_run.py +++ b/projects/caliper/engine/file_export/artifacts_export_run.py @@ -324,6 +324,7 @@ def run_multi_run_artifacts_export( run_dirs: list[Path], backend: tuple[str, ...] | list[str], mlflow_experiment: str | None = None, + mlflow_run_id: str | None = None, mlflow_run_name: str | None = None, mlflow_secrets_path: Path | None = None, mlflow_config_data: dict[str, Any] | None = None, @@ -411,6 +412,7 @@ def run_multi_run_artifacts_export( parameters_file="parameters.json", tracking_uri=tracking_uri, experiment=experiment, + run_id=mlflow_run_id, parent_run_name=run_name, insecure_tls=insecure_tls, connection=mlflow_connection, diff --git a/projects/caliper/engine/file_export/mlflow_backend.py b/projects/caliper/engine/file_export/mlflow_backend.py index 2d199e34c..4173f9a08 100644 --- a/projects/caliper/engine/file_export/mlflow_backend.py +++ b/projects/caliper/engine/file_export/mlflow_backend.py @@ -491,6 +491,7 @@ def log_multi_run_artifacts( parameters_file: str, tracking_uri: str | None, experiment: str | None, + run_id: str | None = None, parent_run_name: str | None = None, insecure_tls: bool = False, connection: dict[str, Any] | None = None, @@ -552,7 +553,9 @@ def _run(uri: str | None) -> tuple[str, dict[str, Any] | None]: client = mlflow.tracking.MlflowClient() start_kw: dict[str, Any] = {} - if parent_run_name: + if run_id: + start_kw["run_id"] = run_id + elif parent_run_name: start_kw["run_name"] = parent_run_name with mlflow.start_run(**start_kw) as parent: diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index 59a0f4c96..cb6859f40 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -226,6 +226,7 @@ def run_from_orchestration_config( mlflow_run_name=naming.get("parent_run_name"), mlflow_secrets_path=mlflow_secrets_path, mlflow_config_data=mlflow_config_data, + mlflow_run_id=export_cfg.mlflow_run_id, child_run_names=naming.get("child_run_names") or {}, verbose=export_cfg.verbose, status_yaml_path=status_yaml, diff --git a/projects/rhaiis/orchestration/config.d/rhaiis.yaml b/projects/rhaiis/orchestration/config.d/rhaiis.yaml index 89b904e48..574cbec1c 100644 --- a/projects/rhaiis/orchestration/config.d/rhaiis.yaml +++ b/projects/rhaiis/orchestration/config.d/rhaiis.yaml @@ -94,4 +94,4 @@ profiler: agent_analysis: enabled: false severity_threshold: 10 - url: "http://agent-gateway.psap-ai-analysis-agent-gateway.svc.cluster.local:8443/v1/stream" + url: "" diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 246aa4195..04d7f24b2 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -295,6 +295,18 @@ def _run_test( logger.exception("Profiler trace upload failed") _warnings.append("Profiler trace upload failed") + # Pre-create MLflow run so its URL can be embedded in test labels / CSV + mlflow_run_meta: dict[str, str] = {} + try: + mlflow_run_meta = _precreate_mlflow_run() + if mlflow_run_meta.get("run_id"): + config.project.set_config("caliper.export.mlflow_run_id", mlflow_run_meta["run_id"]) + config.project.set_config( + "caliper.export.mlflow_experiment_id", mlflow_run_meta["experiment_id"] + ) + except Exception: + logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) + # Phase 2: benchmark + post-processing for ALL workloads trtllm_cfg = runtime_config.get_trtllm_config() if engine == "trtllm" else None for wl_key in workload_keys: @@ -316,6 +328,8 @@ def _run_test( version=version, cluster_tag=cluster_tag, trtllm_config=trtllm_cfg, + mlflow_run_id=mlflow_run_meta.get("run_id", ""), + mlflow_experiment_id=mlflow_run_meta.get("experiment_id", ""), ) try: @@ -382,6 +396,8 @@ def _run_workload_benchmark( version: str, cluster_tag: str, trtllm_config: dict | None = None, + mlflow_run_id: str = "", + mlflow_experiment_id: str = "", ) -> None: """Run benchmark and post-processing for a single workload. @@ -415,6 +431,8 @@ def _run_workload_benchmark( accelerator_chip=gpu_type.upper(), run_uuid=run_uuid, trtllm_config=trtllm_config, + mlflow_run_id=mlflow_run_id, + mlflow_experiment_id=mlflow_experiment_id, ) if not run_benchmark: @@ -471,6 +489,8 @@ def _create_test_labels( accelerator_chip: str = "", run_uuid: str = "", trtllm_config: dict | None = None, + mlflow_run_id: str = "", + mlflow_experiment_id: str = "", ) -> None: _, image_tag = runtime_config.split_image_tag(serving_image) if serving_image else ("", "") parts = [f"{k}: {v}" for k, v in engine_args.items()] @@ -497,11 +517,73 @@ def _create_test_labels( "cluster_tag": cluster_tag, "runtime_args": runtime_args, "run_uuid": run_uuid, + "mlflow_run_id": mlflow_run_id, + "mlflow_experiment_id": mlflow_experiment_id, } write_test_labels(env.ARTIFACT_DIR, labels) logger.info("Created test labels: %s", labels) +def _precreate_mlflow_run() -> dict[str, str]: + """Pre-create an MLflow run so its IDs can be embedded in test labels before CSV generation. + + The run is created and immediately ended (status FINISHED). The export step + will resume it via ``mlflow.start_run(run_id=...)`` to upload artifacts. + + Returns a dict with ``run_id`` and ``experiment_id`` on success, + or an empty dict if MLflow is unavailable. + """ + from projects.core.library import config + from projects.core.library import vault as vault_lib + + vault_name = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False + ) + vault_key = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", None, print=False, warn=False + ) + if not vault_name or not vault_key: + logger.info("MLflow vault not configured, skipping run pre-creation") + return {} + + secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) + if not secrets_path or not secrets_path.exists(): + logger.info("MLflow secrets file not found, skipping run pre-creation") + return {} + + experiment = config.project.get_config( + "caliper.export.backend.mlflow.config.experiment", None, print=False, warn=False + ) + + import mlflow + + from projects.caliper.engine.file_export.mlflow_secrets import ( + load_mlflow_secrets_yaml, + mlflow_connection_env, + ) + + secrets_data = load_mlflow_secrets_yaml(secrets_path) + tracking_uri = secrets_data.get("tracking_uri", "") + + with mlflow_connection_env(secrets_data): + if tracking_uri: + mlflow.set_tracking_uri(tracking_uri) + if experiment: + mlflow.set_experiment(experiment) + + with mlflow.start_run(): + active = mlflow.active_run() + run_id = active.info.run_id + experiment_id = str(active.info.experiment_id) + + logger.info("Pre-created MLflow run %s (experiment=%s)", run_id, experiment_id) + + return { + "run_id": run_id, + "experiment_id": experiment_id, + } + + def _set_mlflow_metadata( model_key: str, workload_key: str, diff --git a/projects/rhaiis/postprocess/csv_export.py b/projects/rhaiis/postprocess/csv_export.py index 5f86396dc..4ac38a588 100644 --- a/projects/rhaiis/postprocess/csv_export.py +++ b/projects/rhaiis/postprocess/csv_export.py @@ -66,6 +66,8 @@ "prefix_tokens", "prefix_count", "request_type", + "mlflow_run_id", + "mlflow_experiment_id", ] @@ -279,4 +281,6 @@ def _total_stat(metric_name: str, stat: str): "prefix_tokens": "", "prefix_count": "", "request_type": "", + "mlflow_run_id": "", + "mlflow_experiment_id": "", } diff --git a/projects/rhaiis/postprocess/plugin.py b/projects/rhaiis/postprocess/plugin.py index 3c6765053..803cbf68a 100644 --- a/projects/rhaiis/postprocess/plugin.py +++ b/projects/rhaiis/postprocess/plugin.py @@ -191,6 +191,8 @@ def export_kpis_to_csv( 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) diff --git a/projects/rhaiis/postprocess/regression.py b/projects/rhaiis/postprocess/regression.py index b71bec30d..0cf2e9a54 100644 --- a/projects/rhaiis/postprocess/regression.py +++ b/projects/rhaiis/postprocess/regression.py @@ -331,6 +331,55 @@ def run_regression_analysis( DASHBOARD_BASE_URL = "https://staging-aidash.apps.ocp4.intlab.redhat.com/" + +def _build_mlflow_run_url() -> str: + """Construct the MLflow run URL at runtime from vault secrets and config.""" + try: + from projects.core.library import config + from projects.core.library import vault as vault_lib + + run_id = config.project.get_config( + "caliper.export.mlflow_run_id", None, print=False, warn=False + ) + experiment_id = config.project.get_config( + "caliper.export.mlflow_experiment_id", None, print=False, warn=False + ) + if not run_id or not experiment_id: + return "" + + vault_name = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False + ) + vault_key = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", + None, + print=False, + warn=False, + ) + 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 "" + + from projects.caliper.engine.file_export.mlflow_secrets import load_mlflow_secrets_yaml + + 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 = config.project.get_config( + "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False + ) + qs = f"?workspace={workspace}" if workspace else "" + return f"{tracking_uri}/#/experiments/{experiment_id}/runs/{run_id}/artifacts{qs}" + except Exception: + logger.debug("Failed to build MLflow run URL", exc_info=True) + return "" + + PROFILE_DISPLAY_NAMES = { "profile1": "Profile A: Balanced (1k/1k)", "profile2": "Profile B: Variable Workload (512/2k)", @@ -492,6 +541,9 @@ def send_regression_notification( ) dashboard_line = f"*Dashboard:* <{dashboard_url}|View Dashboard>\n" + mlflow_url = _build_mlflow_run_url() + mlflow_line = f"*MLflow:* <{mlflow_url}|View Run>\n" if mlflow_url else "" + message = ( f"{icon} *{headline}*\n" f"{user_line}" @@ -502,6 +554,7 @@ def send_regression_notification( f"*Versions:* {current_version} vs {compare_version} (baseline)\n" f"{report_line}" f"{dashboard_line}" + f"{mlflow_line}" f"*Changes:*\n{details}" ) @@ -573,6 +626,9 @@ def send_failure_notification( error_text = error if len(error) <= 500 else error[:500] + "..." + mlflow_url = _build_mlflow_run_url() + mlflow_line = f"*MLflow:* <{mlflow_url}|View Run>\n" if mlflow_url else "" + message = ( f":x: *RHAIIS Pipeline Failed*\n" f"{user_line}" @@ -583,6 +639,7 @@ def send_failure_notification( f"{version_line}" f"{cluster_line}" f"{profiles_line}" + f"{mlflow_line}" f"*Error:*\n```{error_text}```" ) From 411108070565a7ef7c52fe62bade70c27d9f9755 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Fri, 31 Jul 2026 00:02:52 -0400 Subject: [PATCH 02/24] fix: persist pre-created MLflow run_id to disk for cross-step resume config.project.set_config() is in-memory only and doesn't survive across pipeline steps (separate process invocations). Write the pre-created run_id to a __mlflow_precreated_run__.yaml marker file that the export step discovers by scanning the artifact tree. Co-authored-by: Cursor --- projects/caliper/orchestration/export.py | 26 +++++++++++++++++++-- projects/rhaiis/orchestration/test_phase.py | 14 +++++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index cb6859f40..eada0487e 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -206,6 +206,9 @@ def run_from_orchestration_config( run_dirs = discover_run_dirs(from_path) + # Resume a pre-created MLflow run if the test step left a marker + mlflow_run_id = export_cfg.mlflow_run_id or _discover_precreated_mlflow_run_id(from_path) + # Resolve descriptive run names from labels + run_naming config naming = resolve_run_names( run_dirs, mlflow_config_data, fallback_run_name=export_cfg.mlflow_run_name @@ -226,7 +229,7 @@ def run_from_orchestration_config( mlflow_run_name=naming.get("parent_run_name"), mlflow_secrets_path=mlflow_secrets_path, mlflow_config_data=mlflow_config_data, - mlflow_run_id=export_cfg.mlflow_run_id, + mlflow_run_id=mlflow_run_id, child_run_names=naming.get("child_run_names") or {}, verbose=export_cfg.verbose, status_yaml_path=status_yaml, @@ -239,7 +242,7 @@ def run_from_orchestration_config( mlflow_kwargs: dict[str, Any] = { "mlflow_experiment": export_cfg.mlflow_experiment, - "mlflow_run_id": export_cfg.mlflow_run_id, + "mlflow_run_id": mlflow_run_id, "mlflow_run_name": effective_name, "mlflow_secrets_path": mlflow_secrets_path, } @@ -261,3 +264,22 @@ def run_from_orchestration_config( with open(status_yaml) as f: return yaml.safe_load(f.read()) + + +MLFLOW_PRECREATED_RUN_MARKER = "__mlflow_precreated_run__.yaml" + + +def _discover_precreated_mlflow_run_id(from_path: Path) -> str | None: + """Find a pre-created MLflow run_id marker written by the test step.""" + markers = sorted(from_path.rglob(MLFLOW_PRECREATED_RUN_MARKER)) + if not markers: + return None + try: + data = yaml.safe_load(markers[0].read_text(encoding="utf-8")) + run_id = data.get("run_id") if isinstance(data, dict) else None + if run_id: + logger.info("Found pre-created MLflow run_id: %s (from %s)", run_id, markers[0]) + return run_id + except (OSError, yaml.YAMLError) as e: + logger.warning("Failed to read MLflow pre-created run marker: %s", e) + return None diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 04d7f24b2..159f1e53c 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -304,6 +304,7 @@ def _run_test( config.project.set_config( "caliper.export.mlflow_experiment_id", mlflow_run_meta["experiment_id"] ) + _write_mlflow_precreated_run_marker(mlflow_run_meta) except Exception: logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) @@ -584,6 +585,19 @@ def _precreate_mlflow_run() -> dict[str, str]: } +MLFLOW_PRECREATED_RUN_MARKER = "__mlflow_precreated_run__.yaml" + + +def _write_mlflow_precreated_run_marker(meta: dict[str, str]) -> None: + """Persist the pre-created MLflow run_id to disk so the export step can resume it.""" + import yaml + + marker_path = env.ARTIFACT_DIR / MLFLOW_PRECREATED_RUN_MARKER + marker_path.parent.mkdir(parents=True, exist_ok=True) + marker_path.write_text(yaml.safe_dump(meta, sort_keys=False), encoding="utf-8") + logger.info("Wrote MLflow pre-created run marker: %s", marker_path) + + def _set_mlflow_metadata( model_key: str, workload_key: str, From 0ccd1d65473f504084fda321891cf267920757d1 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Fri, 31 Jul 2026 00:26:06 -0400 Subject: [PATCH 03/24] fix: set MLFLOW_WORKSPACE during run pre-creation The pre-created run was landing in the wrong workspace/experiment (233 instead of 264) because MLFLOW_WORKSPACE was not set. The export step sets it, so the same experiment name resolved to a different experiment ID, and the export couldn't resume the pre-created run. Co-authored-by: Cursor --- projects/rhaiis/orchestration/test_phase.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 159f1e53c..1205cc833 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -555,6 +555,11 @@ def _precreate_mlflow_run() -> dict[str, str]: experiment = config.project.get_config( "caliper.export.backend.mlflow.config.experiment", None, print=False, warn=False ) + workspace = config.project.get_config( + "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False + ) + + import os import mlflow @@ -566,9 +571,12 @@ def _precreate_mlflow_run() -> dict[str, str]: secrets_data = load_mlflow_secrets_yaml(secrets_path) tracking_uri = secrets_data.get("tracking_uri", "") + prev_workspace = os.environ.get("MLFLOW_WORKSPACE") with mlflow_connection_env(secrets_data): if tracking_uri: mlflow.set_tracking_uri(tracking_uri) + if workspace: + os.environ["MLFLOW_WORKSPACE"] = workspace if experiment: mlflow.set_experiment(experiment) @@ -577,6 +585,11 @@ def _precreate_mlflow_run() -> dict[str, str]: run_id = active.info.run_id experiment_id = str(active.info.experiment_id) + if prev_workspace is not None: + os.environ["MLFLOW_WORKSPACE"] = prev_workspace + else: + os.environ.pop("MLFLOW_WORKSPACE", None) + logger.info("Pre-created MLflow run %s (experiment=%s)", run_id, experiment_id) return { From 9d6b5a4306eb469e5a96d71c0eaeaf843bc39993 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Fri, 31 Jul 2026 01:10:36 -0400 Subject: [PATCH 04/24] fix: read MLflow run_id from test labels as fallback for export resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The marker file approach was fragile — if _write_mlflow_precreated_run_marker() failed, the exception was silently caught and the export step created a new MLflow run instead of resuming the pre-created one. Two changes: - export.py: _discover_precreated_mlflow_run_id() now falls back to reading mlflow_run_id from __test_labels__.yaml in already-discovered run_dirs - test_phase.py: marker write is isolated in its own try/except so failures are logged separately from the pre-creation itself Co-authored-by: Cursor --- .gitignore | 3 +- projects/caliper/orchestration/export.py | 60 +++++++++++++++------ projects/rhaiis/orchestration/test_phase.py | 16 +++--- 3 files changed, 57 insertions(+), 22 deletions(-) diff --git a/.gitignore b/.gitignore index 7a476c764..565ad2d5b 100644 --- a/.gitignore +++ b/.gitignore @@ -164,4 +164,5 @@ temp/ # FORGE launcher configuration (personal settings) projects/core/launcher/launcher_config.yaml -fournos-job-*.yaml \ No newline at end of file +fournos-job-*.yaml +kubeconfig* \ No newline at end of file diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index eada0487e..8a5a85eb3 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -206,8 +206,10 @@ def run_from_orchestration_config( run_dirs = discover_run_dirs(from_path) - # Resume a pre-created MLflow run if the test step left a marker - mlflow_run_id = export_cfg.mlflow_run_id or _discover_precreated_mlflow_run_id(from_path) + # Resume a pre-created MLflow run if the test step left a marker or test labels + mlflow_run_id = export_cfg.mlflow_run_id or _discover_precreated_mlflow_run_id( + from_path, run_dirs + ) # Resolve descriptive run names from labels + run_naming config naming = resolve_run_names( @@ -269,17 +271,45 @@ def run_from_orchestration_config( MLFLOW_PRECREATED_RUN_MARKER = "__mlflow_precreated_run__.yaml" -def _discover_precreated_mlflow_run_id(from_path: Path) -> str | None: - """Find a pre-created MLflow run_id marker written by the test step.""" +def _discover_precreated_mlflow_run_id( + from_path: Path, run_dirs: list[Path] | None = None +) -> str | None: + """Find a pre-created MLflow run_id from the marker file or test labels. + + Strategy: + 1. Look for the dedicated ``__mlflow_precreated_run__.yaml`` marker. + 2. If not found, fall back to reading ``mlflow_run_id`` from the first + ``__test_labels__.yaml`` that contains it (same files that + ``_discover_run_dirs`` already found). + """ + # --- attempt 1: dedicated marker file --- markers = sorted(from_path.rglob(MLFLOW_PRECREATED_RUN_MARKER)) - if not markers: - return None - try: - data = yaml.safe_load(markers[0].read_text(encoding="utf-8")) - run_id = data.get("run_id") if isinstance(data, dict) else None - if run_id: - logger.info("Found pre-created MLflow run_id: %s (from %s)", run_id, markers[0]) - return run_id - except (OSError, yaml.YAMLError) as e: - logger.warning("Failed to read MLflow pre-created run marker: %s", e) - return None + for marker in markers: + try: + data = yaml.safe_load(marker.read_text(encoding="utf-8")) + run_id = data.get("run_id") if isinstance(data, dict) else None + if run_id: + logger.info("Found pre-created MLflow run_id: %s (from marker %s)", run_id, marker) + return run_id + except (OSError, yaml.YAMLError) as e: + logger.warning("Failed to read MLflow pre-created run marker %s: %s", marker, e) + + # --- attempt 2: read from test labels --- + TEST_LABELS_MARKER = "__test_labels__.yaml" + dirs_to_scan = run_dirs if run_dirs else [] + for run_dir in dirs_to_scan: + labels_path = run_dir / TEST_LABELS_MARKER + if not labels_path.is_file(): + continue + try: + labels = yaml.safe_load(labels_path.read_text(encoding="utf-8")) + run_id = labels.get("mlflow_run_id") if isinstance(labels, dict) else None + if run_id: + logger.info( + "Found pre-created MLflow run_id: %s (from test labels %s)", run_id, labels_path + ) + return run_id + except (OSError, yaml.YAMLError) as e: + logger.warning("Failed to read test labels %s: %s", labels_path, e) + + return None diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 1205cc833..18c809bec 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -299,15 +299,19 @@ def _run_test( mlflow_run_meta: dict[str, str] = {} try: mlflow_run_meta = _precreate_mlflow_run() - if mlflow_run_meta.get("run_id"): - config.project.set_config("caliper.export.mlflow_run_id", mlflow_run_meta["run_id"]) - config.project.set_config( - "caliper.export.mlflow_experiment_id", mlflow_run_meta["experiment_id"] - ) - _write_mlflow_precreated_run_marker(mlflow_run_meta) except Exception: logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) + if mlflow_run_meta.get("run_id"): + config.project.set_config("caliper.export.mlflow_run_id", mlflow_run_meta["run_id"]) + config.project.set_config( + "caliper.export.mlflow_experiment_id", mlflow_run_meta["experiment_id"] + ) + try: + _write_mlflow_precreated_run_marker(mlflow_run_meta) + except Exception: + logger.warning("Failed to write MLflow marker file; continuing", exc_info=True) + # Phase 2: benchmark + post-processing for ALL workloads trtllm_cfg = runtime_config.get_trtllm_config() if engine == "trtllm" else None for wl_key in workload_keys: From a3342045acf9e486eae56470a6b255b6a988c563 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Fri, 31 Jul 2026 01:23:08 -0400 Subject: [PATCH 05/24] fix: remove set_config calls for non-existent caliper.export.mlflow_run_id key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit config.project.set_config() raises KeyError when the key doesn't exist in the config schema. This was the root cause of the marker file never being written — the KeyError was caught by the outer try/except before the marker write could execute. The set_config calls are unnecessary since the run_id is communicated via test labels and the marker file, not in-memory config. Co-authored-by: Cursor --- projects/rhaiis/orchestration/test_phase.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 18c809bec..40751ae20 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -303,10 +303,6 @@ def _run_test( logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) if mlflow_run_meta.get("run_id"): - config.project.set_config("caliper.export.mlflow_run_id", mlflow_run_meta["run_id"]) - config.project.set_config( - "caliper.export.mlflow_experiment_id", mlflow_run_meta["experiment_id"] - ) try: _write_mlflow_precreated_run_marker(mlflow_run_meta) except Exception: From f2034b1fb18ac1c07b417bc1ea3cf36216d0b5ce Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Fri, 31 Jul 2026 01:44:50 -0400 Subject: [PATCH 06/24] fix: set MLflow run name to FJOB_NAME during pre-creation and resume Pre-created runs got MLflow's auto-generated name (e.g. rambunctious-fowl-43) because start_run() was called without run_name. The export step's resume path also skipped run_name when run_id was set (elif branch). Two fixes: - Pre-creation: pass FJOB_NAME as run_name to start_run() - Resume: pass both run_id and run_name so the name is updated on resume Co-authored-by: Cursor --- projects/caliper/engine/file_export/mlflow_backend.py | 4 ++-- projects/rhaiis/orchestration/test_phase.py | 3 ++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/projects/caliper/engine/file_export/mlflow_backend.py b/projects/caliper/engine/file_export/mlflow_backend.py index 4173f9a08..fb932518d 100644 --- a/projects/caliper/engine/file_export/mlflow_backend.py +++ b/projects/caliper/engine/file_export/mlflow_backend.py @@ -416,7 +416,7 @@ def _run(uri: str | None) -> tuple[str, dict[str, Any] | None]: start_kw: dict[str, Any] = {} if run_id: start_kw["run_id"] = run_id - elif run_name: + if run_name: start_kw["run_name"] = run_name meta: dict[str, Any] | None = None @@ -555,7 +555,7 @@ def _run(uri: str | None) -> tuple[str, dict[str, Any] | None]: start_kw: dict[str, Any] = {} if run_id: start_kw["run_id"] = run_id - elif parent_run_name: + if parent_run_name: start_kw["run_name"] = parent_run_name with mlflow.start_run(**start_kw) as parent: diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 40751ae20..568616636 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -580,7 +580,8 @@ def _precreate_mlflow_run() -> dict[str, str]: if experiment: mlflow.set_experiment(experiment) - with mlflow.start_run(): + run_name = os.environ.get("FJOB_NAME") + with mlflow.start_run(run_name=run_name): active = mlflow.active_run() run_id = active.info.run_id experiment_id = str(active.info.experiment_id) From a31698fe0e97b38e8e79b6cda43ed5b74542fd8c Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Sun, 2 Aug 2026 23:12:57 -0400 Subject: [PATCH 07/24] refactor: move MLflow helpers from rhaiis to caliper orchestration layer Addresses reviewer feedback: rhaiis orchestration/postprocess must not import caliper engine code directly. Moved to caliper/orchestration/export.py: - precreate_mlflow_run() (was _precreate_mlflow_run in test_phase.py) - write_mlflow_precreated_run_marker() (was _write_mlflow_precreated_run_marker) - build_mlflow_run_url() (was _build_mlflow_run_url in regression.py) rhaiis files now import from projects.caliper.orchestration.export. Co-authored-by: Cursor --- projects/caliper/orchestration/export.py | 125 ++++++++++++++++++++ projects/rhaiis/orchestration/test_phase.py | 96 ++------------- projects/rhaiis/postprocess/regression.py | 45 +------ 3 files changed, 134 insertions(+), 132 deletions(-) diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index 8a5a85eb3..2f425c461 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -268,9 +268,134 @@ def run_from_orchestration_config( return yaml.safe_load(f.read()) +def precreate_mlflow_run() -> dict[str, str]: + """Pre-create an MLflow run so its IDs can be embedded in test labels before CSV generation. + + The run is created and immediately ended (status FINISHED). The export step + will resume it via ``mlflow.start_run(run_id=...)`` to upload artifacts. + + Returns a dict with ``run_id`` and ``experiment_id`` on success, + or an empty dict if MLflow is unavailable. + """ + from projects.core.library import config + + vault_name = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False + ) + vault_key = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", None, print=False, warn=False + ) + if not vault_name or not vault_key: + logger.info("MLflow vault not configured, skipping run pre-creation") + return {} + + secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) + if not secrets_path or not secrets_path.exists(): + logger.info("MLflow secrets file not found, skipping run pre-creation") + return {} + + experiment = config.project.get_config( + "caliper.export.backend.mlflow.config.experiment", None, print=False, warn=False + ) + workspace = config.project.get_config( + "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False + ) + + import mlflow + + from projects.caliper.engine.file_export.mlflow_secrets import ( + load_mlflow_secrets_yaml, + mlflow_connection_env, + ) + + secrets_data = load_mlflow_secrets_yaml(secrets_path) + tracking_uri = secrets_data.get("tracking_uri", "") + + prev_workspace = os.environ.get("MLFLOW_WORKSPACE") + with mlflow_connection_env(secrets_data): + if tracking_uri: + mlflow.set_tracking_uri(tracking_uri) + if workspace: + os.environ["MLFLOW_WORKSPACE"] = workspace + if experiment: + mlflow.set_experiment(experiment) + + run_name = os.environ.get("FJOB_NAME") + with mlflow.start_run(run_name=run_name): + active = mlflow.active_run() + run_id = active.info.run_id + experiment_id = str(active.info.experiment_id) + + if prev_workspace is not None: + os.environ["MLFLOW_WORKSPACE"] = prev_workspace + else: + os.environ.pop("MLFLOW_WORKSPACE", None) + + logger.info("Pre-created MLflow run %s (experiment=%s)", run_id, experiment_id) + + return { + "run_id": run_id, + "experiment_id": experiment_id, + } + + MLFLOW_PRECREATED_RUN_MARKER = "__mlflow_precreated_run__.yaml" +def write_mlflow_precreated_run_marker(meta: dict[str, str]) -> None: + """Persist the pre-created MLflow run_id to disk so the export step can resume it.""" + marker_path = env.ARTIFACT_DIR / MLFLOW_PRECREATED_RUN_MARKER + marker_path.parent.mkdir(parents=True, exist_ok=True) + marker_path.write_text(yaml.safe_dump(meta, sort_keys=False), encoding="utf-8") + logger.info("Wrote MLflow pre-created run marker: %s", marker_path) + + +def build_mlflow_run_url() -> str: + """Construct the MLflow run URL at runtime from vault secrets and config.""" + try: + from projects.caliper.engine.file_export.mlflow_secrets import load_mlflow_secrets_yaml + from projects.core.library import config + + run_id = config.project.get_config( + "caliper.export.mlflow_run_id", None, print=False, warn=False + ) + experiment_id = config.project.get_config( + "caliper.export.mlflow_experiment_id", None, print=False, warn=False + ) + if not run_id or not experiment_id: + return "" + + vault_name = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False + ) + vault_key = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", + None, + print=False, + warn=False, + ) + 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 = config.project.get_config( + "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False + ) + qs = f"?workspace={workspace}" if workspace else "" + return f"{tracking_uri}/#/experiments/{experiment_id}/runs/{run_id}/artifacts{qs}" + except Exception: + logger.debug("Failed to build MLflow run URL", exc_info=True) + return "" + + def _discover_precreated_mlflow_run_id( from_path: Path, run_dirs: list[Path] | None = None ) -> str | None: diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 568616636..bbaba1d2c 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -296,15 +296,20 @@ def _run_test( _warnings.append("Profiler trace upload failed") # Pre-create MLflow run so its URL can be embedded in test labels / CSV + from projects.caliper.orchestration.export import ( + precreate_mlflow_run, + write_mlflow_precreated_run_marker, + ) + mlflow_run_meta: dict[str, str] = {} try: - mlflow_run_meta = _precreate_mlflow_run() + mlflow_run_meta = precreate_mlflow_run() except Exception: logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) if mlflow_run_meta.get("run_id"): try: - _write_mlflow_precreated_run_marker(mlflow_run_meta) + write_mlflow_precreated_run_marker(mlflow_run_meta) except Exception: logger.warning("Failed to write MLflow marker file; continuing", exc_info=True) @@ -525,93 +530,6 @@ def _create_test_labels( logger.info("Created test labels: %s", labels) -def _precreate_mlflow_run() -> dict[str, str]: - """Pre-create an MLflow run so its IDs can be embedded in test labels before CSV generation. - - The run is created and immediately ended (status FINISHED). The export step - will resume it via ``mlflow.start_run(run_id=...)`` to upload artifacts. - - Returns a dict with ``run_id`` and ``experiment_id`` on success, - or an empty dict if MLflow is unavailable. - """ - from projects.core.library import config - from projects.core.library import vault as vault_lib - - vault_name = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False - ) - vault_key = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", None, print=False, warn=False - ) - if not vault_name or not vault_key: - logger.info("MLflow vault not configured, skipping run pre-creation") - return {} - - secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) - if not secrets_path or not secrets_path.exists(): - logger.info("MLflow secrets file not found, skipping run pre-creation") - return {} - - experiment = config.project.get_config( - "caliper.export.backend.mlflow.config.experiment", None, print=False, warn=False - ) - workspace = config.project.get_config( - "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False - ) - - import os - - import mlflow - - from projects.caliper.engine.file_export.mlflow_secrets import ( - load_mlflow_secrets_yaml, - mlflow_connection_env, - ) - - secrets_data = load_mlflow_secrets_yaml(secrets_path) - tracking_uri = secrets_data.get("tracking_uri", "") - - prev_workspace = os.environ.get("MLFLOW_WORKSPACE") - with mlflow_connection_env(secrets_data): - if tracking_uri: - mlflow.set_tracking_uri(tracking_uri) - if workspace: - os.environ["MLFLOW_WORKSPACE"] = workspace - if experiment: - mlflow.set_experiment(experiment) - - run_name = os.environ.get("FJOB_NAME") - with mlflow.start_run(run_name=run_name): - active = mlflow.active_run() - run_id = active.info.run_id - experiment_id = str(active.info.experiment_id) - - if prev_workspace is not None: - os.environ["MLFLOW_WORKSPACE"] = prev_workspace - else: - os.environ.pop("MLFLOW_WORKSPACE", None) - - logger.info("Pre-created MLflow run %s (experiment=%s)", run_id, experiment_id) - - return { - "run_id": run_id, - "experiment_id": experiment_id, - } - - -MLFLOW_PRECREATED_RUN_MARKER = "__mlflow_precreated_run__.yaml" - - -def _write_mlflow_precreated_run_marker(meta: dict[str, str]) -> None: - """Persist the pre-created MLflow run_id to disk so the export step can resume it.""" - import yaml - - marker_path = env.ARTIFACT_DIR / MLFLOW_PRECREATED_RUN_MARKER - marker_path.parent.mkdir(parents=True, exist_ok=True) - marker_path.write_text(yaml.safe_dump(meta, sort_keys=False), encoding="utf-8") - logger.info("Wrote MLflow pre-created run marker: %s", marker_path) - - def _set_mlflow_metadata( model_key: str, workload_key: str, diff --git a/projects/rhaiis/postprocess/regression.py b/projects/rhaiis/postprocess/regression.py index 0cf2e9a54..a8b3f5946 100644 --- a/projects/rhaiis/postprocess/regression.py +++ b/projects/rhaiis/postprocess/regression.py @@ -334,50 +334,9 @@ def run_regression_analysis( def _build_mlflow_run_url() -> str: """Construct the MLflow run URL at runtime from vault secrets and config.""" - try: - from projects.core.library import config - from projects.core.library import vault as vault_lib - - run_id = config.project.get_config( - "caliper.export.mlflow_run_id", None, print=False, warn=False - ) - experiment_id = config.project.get_config( - "caliper.export.mlflow_experiment_id", None, print=False, warn=False - ) - if not run_id or not experiment_id: - return "" - - vault_name = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False - ) - vault_key = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", - None, - print=False, - warn=False, - ) - 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 "" - - from projects.caliper.engine.file_export.mlflow_secrets import load_mlflow_secrets_yaml + from projects.caliper.orchestration.export import build_mlflow_run_url - 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 = config.project.get_config( - "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False - ) - qs = f"?workspace={workspace}" if workspace else "" - return f"{tracking_uri}/#/experiments/{experiment_id}/runs/{run_id}/artifacts{qs}" - except Exception: - logger.debug("Failed to build MLflow run URL", exc_info=True) - return "" + return build_mlflow_run_url() PROFILE_DISPLAY_NAMES = { From 376aff36f513a4c0294505643a385d6eb57a6706 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Sun, 2 Aug 2026 23:52:48 -0400 Subject: [PATCH 08/24] feat: add slack_notify_always for success notifications + move MLflow helpers to caliper - Add send_success_notification() for clean pipeline completions - Wire tests.rhaiis.slack_notify_always config to trigger it - Add dashboard link (gated on csv_dashboard.enabled) and MLflow link - Move precreate_mlflow_run, write_mlflow_precreated_run_marker, and build_mlflow_run_url from rhaiis to caliper/orchestration/export.py Co-authored-by: Cursor --- projects/rhaiis/orchestration/analysis.py | 25 +++++- projects/rhaiis/orchestration/config.yaml | 1 + projects/rhaiis/orchestration/test_phase.py | 59 ++++++++++----- projects/rhaiis/postprocess/regression.py | 84 +++++++++++++++++++++ 4 files changed, 148 insertions(+), 21 deletions(-) diff --git a/projects/rhaiis/orchestration/analysis.py b/projects/rhaiis/orchestration/analysis.py index 724b99b5f..618d58540 100644 --- a/projects/rhaiis/orchestration/analysis.py +++ b/projects/rhaiis/orchestration/analysis.py @@ -165,6 +165,11 @@ def run_regression_check( restrict_profiles=restrict_profiles, ) + _ea = engine_args or {} + tp = _ea.get("tensor-parallel-size") or _ea.get("tp-size") or _ea.get("tp_size") or "" + dp = _ea.get("data-parallel-size") or _ea.get("dp-size") or "" + slack_user = config.project.get_config("tests.rhaiis.slack_user", "") + if analysis.get("regression_count", 0) > 0 or analysis.get("improvement_count", 0) > 0: report_url = "" agent_cfg = config.project.get_config("rhaiis.agent_analysis", {}) @@ -181,20 +186,32 @@ def run_regression_check( from projects.rhaiis.postprocess.regression import send_regression_notification - _ea = engine_args or {} - tp = _ea.get("tensor-parallel-size") or _ea.get("tp-size") or _ea.get("tp_size") or "" - dp = _ea.get("data-parallel-size") or _ea.get("dp-size") or "" send_regression_notification( analysis, model=model_cfg.get("hf_model_id", ""), accelerator=accelerator, job_id=run_uuid, - slack_user=config.project.get_config("tests.rhaiis.slack_user", ""), + slack_user=slack_user, notification_vault="psap-forge-notifications", report_url=report_url, tp=str(tp), dp=str(dp), ) + elif config.project.get_config("tests.rhaiis.slack_notify_always", False): + from projects.rhaiis.postprocess.regression import send_success_notification + + send_success_notification( + model=model_cfg.get("hf_model_id", ""), + accelerator=accelerator, + job_id=run_uuid, + slack_user=slack_user, + notification_vault="psap-forge-notifications", + tp=str(tp), + dp=str(dp), + version=current_version, + workload_keys=config.project.get_config("tests.rhaiis.workload_keys", []), + cluster=config.project.get_config("rhaiis.cluster_tag", ""), + ) except Exception: logger.warning("Regression analysis failed; continuing", exc_info=True) finally: diff --git a/projects/rhaiis/orchestration/config.yaml b/projects/rhaiis/orchestration/config.yaml index a5dffe14f..3435082eb 100644 --- a/projects/rhaiis/orchestration/config.yaml +++ b/projects/rhaiis/orchestration/config.yaml @@ -28,6 +28,7 @@ tests: version: "" compare_version: "" slack_user: "" + slack_notify_always: false caliper: postprocess: diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index bbaba1d2c..4bcc1c520 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -619,27 +619,52 @@ def _sync_postprocessed_dashboard_csv(model_key: str, workload_keys: list[str]) version = config.project.get_config("tests.rhaiis.version", "") compare_version = config.project.get_config("tests.rhaiis.compare_version", "") - if not compare_version or not version: + + if compare_version and version: + model_cfg = runtime_config.get_model(model_key) + accelerator = runtime_config.get_accelerator() + engine = runtime_config.get_engine() + engine_defaults = runtime_config.get_engine_args(engine) + first_workload = runtime_config.get_workload(workload_keys[0]) + ea = runtime_config.merge_engine_args(engine_defaults, model_cfg, first_workload, engine) + + from projects.rhaiis.orchestration.analysis import run_regression_check + + run_regression_check( + csv_path, + compare_version, + version, + model_cfg, + accelerator, + run_uuid="", + engine_args=ea, + ) return - model_cfg = runtime_config.get_model(model_key) - accelerator = runtime_config.get_accelerator() - engine = runtime_config.get_engine() - engine_defaults = runtime_config.get_engine_args(engine) - first_workload = runtime_config.get_workload(workload_keys[0]) - ea = runtime_config.merge_engine_args(engine_defaults, model_cfg, first_workload, engine) + if config.project.get_config("tests.rhaiis.slack_notify_always", False): + from projects.rhaiis.postprocess.regression import send_success_notification - from projects.rhaiis.orchestration.analysis import run_regression_check + model_cfg = runtime_config.get_model(model_key) + accelerator = runtime_config.get_accelerator() + engine = runtime_config.get_engine() + engine_defaults = runtime_config.get_engine_args(engine) + first_workload = runtime_config.get_workload(workload_keys[0]) + ea = runtime_config.merge_engine_args(engine_defaults, model_cfg, first_workload, engine) + tp = ea.get("tensor-parallel-size") or ea.get("tp-size") or ea.get("tp_size") or "" + dp = ea.get("data-parallel-size") or ea.get("dp-size") or "" - run_regression_check( - csv_path, - compare_version, - version, - model_cfg, - accelerator, - run_uuid="", - engine_args=ea, - ) + send_success_notification( + model=model_cfg.get("hf_model_id", ""), + accelerator=accelerator, + job_id=os.environ.get("FJOB_NAME", ""), + slack_user=config.project.get_config("tests.rhaiis.slack_user", ""), + notification_vault="psap-forge-notifications", + tp=str(tp), + dp=str(dp), + version=version, + workload_keys=workload_keys, + cluster=config.project.get_config("rhaiis.cluster_tag", ""), + ) def _upload_predictor_log(run_uuid: str) -> None: diff --git a/projects/rhaiis/postprocess/regression.py b/projects/rhaiis/postprocess/regression.py index a8b3f5946..565da1af1 100644 --- a/projects/rhaiis/postprocess/regression.py +++ b/projects/rhaiis/postprocess/regression.py @@ -526,6 +526,90 @@ def send_regression_notification( ) +def send_success_notification( + *, + model: str = "", + accelerator: str = "", + job_id: str = "", + slack_user: str = "", + notification_vault: str | None = None, + dry_run: bool = False, + tp: str = "", + dp: str = "", + version: str = "", + workload_keys: list[str] | None = None, + cluster: str = "", +) -> bool: + """Send a Slack notification when the RHAIIS pipeline succeeds with no regressions. + + Returns: + True if notification sent successfully + """ + if slack_user and re.match(r"^[UW][A-Z0-9]+$", slack_user): + user_line = f"*Triggered by:* <@{slack_user}>\n" + elif slack_user: + user_line = f"*Triggered by:* {slack_user}\n" + else: + user_line = "" + + parallelism_parts = [] + if tp: + parallelism_parts.append(f"TP={tp}") + if dp: + parallelism_parts.append(f"DP={dp}") + parallelism_line = ( + f"*Parallelism:* {', '.join(parallelism_parts)}\n" if parallelism_parts else "" + ) + + profiles_line = "" + if workload_keys: + profiles_line = f"*Workloads:* {', '.join(workload_keys)}\n" + + cluster_line = f"*Cluster:* {cluster}\n" if cluster else "" + version_line = f"*Version:* {version}\n" if version else "" + + dashboard_line = "" + try: + from projects.core.library import config + + if config.project.get_config("caliper.postprocess.csv_dashboard.enabled", False): + dashboard_url = _build_dashboard_url( + model=model, + accelerator=accelerator, + current_version=version, + profiles=workload_keys, + tp=tp, + ) + dashboard_line = f"*Dashboard:* <{dashboard_url}|View Dashboard>\n" + except Exception: + pass + + mlflow_url = _build_mlflow_run_url() + mlflow_line = f"*MLflow:* <{mlflow_url}|View Run>\n" if mlflow_url else "" + + message = ( + f":white_check_mark: *RHAIIS Pipeline Succeeded*\n" + f"{user_line}" + f"*Job:* `{job_id}`\n" + f"*Model:* {model}\n" + f"*Accelerator:* {accelerator}\n" + f"{parallelism_line}" + f"{version_line}" + f"{cluster_line}" + f"{profiles_line}" + f"{dashboard_line}" + f"{mlflow_line}" + ) + + if dry_run: + logger.info("DRY RUN success notification:\n%s", message) + return True + + return _send_via_topsail_bot( + message, notification_vault=notification_vault, channel_id=RHAIIS_SLACK_CHANNEL_ID + ) + + def send_failure_notification( *, error: str, From 89e7854ec18b7102770d60af78db4ae5eb0d509c Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Mon, 3 Aug 2026 00:08:18 -0400 Subject: [PATCH 09/24] fix: read MLflow IDs from marker file in build_mlflow_run_url() The config keys caliper.export.mlflow_run_id/experiment_id don't exist in the schema, so get_config returns None and the URL is never built. Now reads from the __mlflow_precreated_run__.yaml marker file on disk. Co-authored-by: Cursor --- projects/caliper/orchestration/export.py | 24 +++++++++++++++++------- 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index 2f425c461..184f5f9a8 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -350,18 +350,28 @@ def write_mlflow_precreated_run_marker(meta: dict[str, str]) -> None: logger.info("Wrote MLflow pre-created run marker: %s", marker_path) +def _read_mlflow_ids_from_marker() -> tuple[str, str]: + """Read run_id and experiment_id from the pre-created marker file on disk.""" + try: + artifact_dir = Path(env.ARTIFACT_DIR) if env.ARTIFACT_DIR else None + if not artifact_dir: + return "", "" + for marker in sorted(artifact_dir.rglob(MLFLOW_PRECREATED_RUN_MARKER)): + data = yaml.safe_load(marker.read_text(encoding="utf-8")) + if isinstance(data, dict) and data.get("run_id"): + return data["run_id"], data.get("experiment_id", "") + except Exception: + pass + return "", "" + + def build_mlflow_run_url() -> str: - """Construct the MLflow run URL at runtime from vault secrets and config.""" + """Construct the MLflow run URL at runtime from vault secrets and the marker file.""" try: from projects.caliper.engine.file_export.mlflow_secrets import load_mlflow_secrets_yaml from projects.core.library import config - run_id = config.project.get_config( - "caliper.export.mlflow_run_id", None, print=False, warn=False - ) - experiment_id = config.project.get_config( - "caliper.export.mlflow_experiment_id", None, print=False, warn=False - ) + run_id, experiment_id = _read_mlflow_ids_from_marker() if not run_id or not experiment_id: return "" From b61992be1f85facbbccb9adb345f66a4a71001f8 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Mon, 3 Aug 2026 20:09:02 -0400 Subject: [PATCH 10/24] fix: address CodeRabbit review findings - mlflow_backend.py: set mlflow.runName tag explicitly when resuming a run with run_id, since some MLflow versions ignore run_name in start_run() during resume - export.py: wrap MLFLOW_WORKSPACE restoration in finally block so it executes even if set_experiment/start_run raises - export.py: validate tracking_uri with assert_tracking_uri_has_no_userinfo() before composing the Slack-visible MLflow URL Co-authored-by: Cursor --- .../engine/file_export/mlflow_backend.py | 4 ++ projects/caliper/orchestration/export.py | 42 ++++++++++--------- 2 files changed, 27 insertions(+), 19 deletions(-) diff --git a/projects/caliper/engine/file_export/mlflow_backend.py b/projects/caliper/engine/file_export/mlflow_backend.py index fb932518d..b88b9be27 100644 --- a/projects/caliper/engine/file_export/mlflow_backend.py +++ b/projects/caliper/engine/file_export/mlflow_backend.py @@ -423,6 +423,8 @@ def _run(uri: str | None) -> tuple[str, dict[str, Any] | None]: client = mlflow.tracking.MlflowClient() with mlflow.start_run(**start_kw): rid = mlflow.active_run().info.run_id + if run_id and run_name: + mlflow.set_tag("mlflow.runName", run_name) _apply_run_metadata(effective_meta) _apply_log_model(artifact_root, effective_meta, verbose=verbose) @@ -560,6 +562,8 @@ def _run(uri: str | None) -> tuple[str, dict[str, Any] | None]: with mlflow.start_run(**start_kw) as parent: parent_rid = parent.info.run_id + if run_id and parent_run_name: + mlflow.set_tag("mlflow.runName", parent_run_name) _apply_run_metadata(effective_meta) _upload_mlflow_files_parallel( diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index 184f5f9a8..43b90af60 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -312,24 +312,25 @@ def precreate_mlflow_run() -> dict[str, str]: tracking_uri = secrets_data.get("tracking_uri", "") prev_workspace = os.environ.get("MLFLOW_WORKSPACE") - with mlflow_connection_env(secrets_data): - if tracking_uri: - mlflow.set_tracking_uri(tracking_uri) - if workspace: - os.environ["MLFLOW_WORKSPACE"] = workspace - if experiment: - mlflow.set_experiment(experiment) - - run_name = os.environ.get("FJOB_NAME") - with mlflow.start_run(run_name=run_name): - active = mlflow.active_run() - run_id = active.info.run_id - experiment_id = str(active.info.experiment_id) - - if prev_workspace is not None: - os.environ["MLFLOW_WORKSPACE"] = prev_workspace - else: - os.environ.pop("MLFLOW_WORKSPACE", None) + try: + with mlflow_connection_env(secrets_data): + if tracking_uri: + mlflow.set_tracking_uri(tracking_uri) + if workspace: + os.environ["MLFLOW_WORKSPACE"] = workspace + if experiment: + mlflow.set_experiment(experiment) + + run_name = os.environ.get("FJOB_NAME") + with mlflow.start_run(run_name=run_name): + active = mlflow.active_run() + run_id = active.info.run_id + experiment_id = str(active.info.experiment_id) + finally: + if prev_workspace is not None: + os.environ["MLFLOW_WORKSPACE"] = prev_workspace + else: + os.environ.pop("MLFLOW_WORKSPACE", None) logger.info("Pre-created MLflow run %s (experiment=%s)", run_id, experiment_id) @@ -368,7 +369,9 @@ def _read_mlflow_ids_from_marker() -> tuple[str, str]: def build_mlflow_run_url() -> str: """Construct the MLflow run URL at runtime from vault secrets and the marker file.""" try: - from projects.caliper.engine.file_export.mlflow_secrets import load_mlflow_secrets_yaml + from projects.caliper.engine.file_export.mlflow_secrets import ( + load_mlflow_secrets_yaml, + ) from projects.core.library import config run_id, experiment_id = _read_mlflow_ids_from_marker() @@ -395,6 +398,7 @@ def build_mlflow_run_url() -> str: tracking_uri = secrets_data.get("tracking_uri", "").rstrip("/") if not tracking_uri.startswith(("http://", "https://")): return "" + assert_tracking_uri_has_no_userinfo(tracking_uri) workspace = config.project.get_config( "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False From 85247981c58e62da062600aaf37af919704c7b0e Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Mon, 3 Aug 2026 20:42:45 -0400 Subject: [PATCH 11/24] simplify: drop test-labels fallback from _discover_precreated_mlflow_run_id The marker file is always written reliably; the test-labels fallback was redundant defensive code. Co-authored-by: Cursor --- projects/caliper/orchestration/export.py | 39 +++--------------------- 1 file changed, 4 insertions(+), 35 deletions(-) diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index 43b90af60..a43b5c6f3 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -207,9 +207,7 @@ def run_from_orchestration_config( run_dirs = discover_run_dirs(from_path) # Resume a pre-created MLflow run if the test step left a marker or test labels - mlflow_run_id = export_cfg.mlflow_run_id or _discover_precreated_mlflow_run_id( - from_path, run_dirs - ) + mlflow_run_id = export_cfg.mlflow_run_id or _discover_precreated_mlflow_run_id(from_path) # Resolve descriptive run names from labels + run_naming config naming = resolve_run_names( @@ -410,20 +408,9 @@ def build_mlflow_run_url() -> str: return "" -def _discover_precreated_mlflow_run_id( - from_path: Path, run_dirs: list[Path] | None = None -) -> str | None: - """Find a pre-created MLflow run_id from the marker file or test labels. - - Strategy: - 1. Look for the dedicated ``__mlflow_precreated_run__.yaml`` marker. - 2. If not found, fall back to reading ``mlflow_run_id`` from the first - ``__test_labels__.yaml`` that contains it (same files that - ``_discover_run_dirs`` already found). - """ - # --- attempt 1: dedicated marker file --- - markers = sorted(from_path.rglob(MLFLOW_PRECREATED_RUN_MARKER)) - for marker in markers: +def _discover_precreated_mlflow_run_id(from_path: Path) -> str | None: + """Find a pre-created MLflow run_id from the marker file.""" + for marker in sorted(from_path.rglob(MLFLOW_PRECREATED_RUN_MARKER)): try: data = yaml.safe_load(marker.read_text(encoding="utf-8")) run_id = data.get("run_id") if isinstance(data, dict) else None @@ -433,22 +420,4 @@ def _discover_precreated_mlflow_run_id( except (OSError, yaml.YAMLError) as e: logger.warning("Failed to read MLflow pre-created run marker %s: %s", marker, e) - # --- attempt 2: read from test labels --- - TEST_LABELS_MARKER = "__test_labels__.yaml" - dirs_to_scan = run_dirs if run_dirs else [] - for run_dir in dirs_to_scan: - labels_path = run_dir / TEST_LABELS_MARKER - if not labels_path.is_file(): - continue - try: - labels = yaml.safe_load(labels_path.read_text(encoding="utf-8")) - run_id = labels.get("mlflow_run_id") if isinstance(labels, dict) else None - if run_id: - logger.info( - "Found pre-created MLflow run_id: %s (from test labels %s)", run_id, labels_path - ) - return run_id - except (OSError, yaml.YAMLError) as e: - logger.warning("Failed to read test labels %s: %s", labels_path, e) - return None From cd6cad7865d41bb1cc20cfd7b42441e31347929e Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Mon, 3 Aug 2026 20:45:47 -0400 Subject: [PATCH 12/24] fix: add missing import for assert_tracking_uri_has_no_userinfo Import was lost during rebase when build_mlflow_run_url() was rewritten to read from the marker file. Co-authored-by: Cursor --- projects/caliper/orchestration/export.py | 1 + 1 file changed, 1 insertion(+) diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index a43b5c6f3..d6652993e 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -368,6 +368,7 @@ def build_mlflow_run_url() -> str: """Construct the MLflow run URL at runtime from vault secrets and the marker file.""" try: from projects.caliper.engine.file_export.mlflow_secrets import ( + assert_tracking_uri_has_no_userinfo, load_mlflow_secrets_yaml, ) from projects.core.library import config From 09ec49aee31e8faafcc43a77ff0280ce79612431 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Mon, 3 Aug 2026 23:36:11 -0400 Subject: [PATCH 13/24] feat: add rampup parameter to all workload profiles Co-authored-by: Cursor --- projects/rhaiis/orchestration/config.d/workloads.yaml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/projects/rhaiis/orchestration/config.d/workloads.yaml b/projects/rhaiis/orchestration/config.d/workloads.yaml index 1cee1e4c6..833cbb5e0 100644 --- a/projects/rhaiis/orchestration/config.d/workloads.yaml +++ b/projects/rhaiis/orchestration/config.d/workloads.yaml @@ -7,20 +7,24 @@ profile1: data: "prompt_tokens=1000,output_tokens=1000" rates: [1,50,100,200,300] max_seconds: 450 + rampup: 10 profile2: data: "prompt_tokens=512,prompt_tokens_stdev=128,prompt_tokens_min=1,prompt_tokens_max=1024,output_tokens=2048,output_tokens_stdev=512,output_tokens_min=1,output_tokens_max=4096" rates: [1,50,100,200,300] max_seconds: 450 + rampup: 10 profile3: data: "prompt_tokens=2048,output_tokens=128" rates: [1, 50, 100, 200, 300] max_seconds: 450 + rampup: 10 profile4: data: "prompt_tokens=8000,output_tokens=1000" rates: [1,25,50,75,100] max_seconds: 450 + rampup: 10 samples: 50 From 48aa32f41ec728b68d2501e65bc195ee83ee2497 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Mon, 3 Aug 2026 23:51:33 -0400 Subject: [PATCH 14/24] fix: wire rampup workload field through to guidellm CLI args Co-authored-by: Cursor --- projects/rhaiis/orchestration/runtime_config.py | 3 +++ projects/rhaiis/orchestration/test_phase.py | 2 ++ 2 files changed, 5 insertions(+) diff --git a/projects/rhaiis/orchestration/runtime_config.py b/projects/rhaiis/orchestration/runtime_config.py index 3f898fa81..96d563fc3 100644 --- a/projects/rhaiis/orchestration/runtime_config.py +++ b/projects/rhaiis/orchestration/runtime_config.py @@ -168,6 +168,7 @@ def build_guidellm_args( data: str, rates: list[int], max_seconds: int, + rampup: int | None = None, ) -> list[str]: guidellm_args = [] for key, value in benchmark_cfg.get("args", {}).items(): @@ -178,6 +179,8 @@ def build_guidellm_args( guidellm_args.append(f"--data={data}") guidellm_args.append(f"--rate={_format_arg_value(rates)}") guidellm_args.append(f"--max-seconds={max_seconds}") + if rampup is not None: + guidellm_args.append(f"--rampup={rampup}") return guidellm_args diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 4bcc1c520..185b63931 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -416,6 +416,7 @@ def _run_workload_benchmark( workload = runtime_config.get_workload(workload_key) rates = workload.get("rates", [1]) max_seconds = workload.get("max_seconds", 180) + rampup = workload.get("rampup") from projects.core.library import config from projects.guidellm.toolbox.run_guidellm_benchmark.main import ( @@ -467,6 +468,7 @@ def _run_workload_benchmark( data=workload["data"], rates=rates, max_seconds=max_seconds, + rampup=rampup, ) run_guidellm_benchmark( From 951a37008cdedefe8b046d0b40fbbe7f58101617 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Tue, 4 Aug 2026 13:50:06 -0400 Subject: [PATCH 15/24] revert: remove unrelated rhaiis changes from this PR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Revert agent_analysis.url change and workload rampup additions — these belong in a separate PR per reviewer feedback. Co-authored-by: Cursor --- projects/rhaiis/orchestration/config.d/rhaiis.yaml | 2 +- projects/rhaiis/orchestration/config.d/workloads.yaml | 4 ---- projects/rhaiis/orchestration/runtime_config.py | 3 --- projects/rhaiis/orchestration/test_phase.py | 2 -- 4 files changed, 1 insertion(+), 10 deletions(-) diff --git a/projects/rhaiis/orchestration/config.d/rhaiis.yaml b/projects/rhaiis/orchestration/config.d/rhaiis.yaml index 574cbec1c..89b904e48 100644 --- a/projects/rhaiis/orchestration/config.d/rhaiis.yaml +++ b/projects/rhaiis/orchestration/config.d/rhaiis.yaml @@ -94,4 +94,4 @@ profiler: agent_analysis: enabled: false severity_threshold: 10 - url: "" + url: "http://agent-gateway.psap-ai-analysis-agent-gateway.svc.cluster.local:8443/v1/stream" diff --git a/projects/rhaiis/orchestration/config.d/workloads.yaml b/projects/rhaiis/orchestration/config.d/workloads.yaml index 833cbb5e0..1cee1e4c6 100644 --- a/projects/rhaiis/orchestration/config.d/workloads.yaml +++ b/projects/rhaiis/orchestration/config.d/workloads.yaml @@ -7,24 +7,20 @@ profile1: data: "prompt_tokens=1000,output_tokens=1000" rates: [1,50,100,200,300] max_seconds: 450 - rampup: 10 profile2: data: "prompt_tokens=512,prompt_tokens_stdev=128,prompt_tokens_min=1,prompt_tokens_max=1024,output_tokens=2048,output_tokens_stdev=512,output_tokens_min=1,output_tokens_max=4096" rates: [1,50,100,200,300] max_seconds: 450 - rampup: 10 profile3: data: "prompt_tokens=2048,output_tokens=128" rates: [1, 50, 100, 200, 300] max_seconds: 450 - rampup: 10 profile4: data: "prompt_tokens=8000,output_tokens=1000" rates: [1,25,50,75,100] max_seconds: 450 - rampup: 10 samples: 50 diff --git a/projects/rhaiis/orchestration/runtime_config.py b/projects/rhaiis/orchestration/runtime_config.py index 96d563fc3..3f898fa81 100644 --- a/projects/rhaiis/orchestration/runtime_config.py +++ b/projects/rhaiis/orchestration/runtime_config.py @@ -168,7 +168,6 @@ def build_guidellm_args( data: str, rates: list[int], max_seconds: int, - rampup: int | None = None, ) -> list[str]: guidellm_args = [] for key, value in benchmark_cfg.get("args", {}).items(): @@ -179,8 +178,6 @@ def build_guidellm_args( guidellm_args.append(f"--data={data}") guidellm_args.append(f"--rate={_format_arg_value(rates)}") guidellm_args.append(f"--max-seconds={max_seconds}") - if rampup is not None: - guidellm_args.append(f"--rampup={rampup}") return guidellm_args diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 185b63931..4bcc1c520 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -416,7 +416,6 @@ def _run_workload_benchmark( workload = runtime_config.get_workload(workload_key) rates = workload.get("rates", [1]) max_seconds = workload.get("max_seconds", 180) - rampup = workload.get("rampup") from projects.core.library import config from projects.guidellm.toolbox.run_guidellm_benchmark.main import ( @@ -468,7 +467,6 @@ def _run_workload_benchmark( data=workload["data"], rates=rates, max_seconds=max_seconds, - rampup=rampup, ) run_guidellm_benchmark( From 1325e18f844475ef1cade34f3edcd0d5b407e2d8 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Tue, 4 Aug 2026 13:58:16 -0400 Subject: [PATCH 16/24] refactor: address PR review comments on caliper layering and robustness - Create projects/caliper/public/file_export.py to expose engine functions; orchestration imports from public instead of engine - Refactor precreate_mlflow_run() to accept secrets_path, experiment, workspace from caller instead of reading project config - Combine precreate_mlflow_run() and write_mlflow_precreated_run_marker() into a single call - Refactor build_mlflow_run_url() to accept secrets_path and workspace from caller - Move config reads to rhaiis callers (test_phase.py, regression.py) - Save/restore tracking URI in finally block alongside MLFLOW_WORKSPACE - URL-encode workspace in MLflow URL query string - Add logger.warning for all failure return paths - Warn on invalid marker YAML object type - Remove bare except swallowing in _read_mlflow_ids_from_marker - Add docstring explaining MLFLOW_PRECREATED_RUN_MARKER purpose - Revert unrelated rhaiis changes (agent_analysis.url, workloads rampup) Co-authored-by: Cursor --- projects/caliper/orchestration/export.py | 180 ++++++++++---------- projects/caliper/public/__init__.py | 1 + projects/caliper/public/file_export.py | 17 ++ projects/rhaiis/orchestration/test_phase.py | 46 +++-- projects/rhaiis/postprocess/regression.py | 23 ++- 5 files changed, 164 insertions(+), 103 deletions(-) create mode 100644 projects/caliper/public/__init__.py create mode 100644 projects/caliper/public/file_export.py diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index d6652993e..ac1cb0b5d 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -266,42 +266,41 @@ def run_from_orchestration_config( return yaml.safe_load(f.read()) -def precreate_mlflow_run() -> dict[str, str]: - """Pre-create an MLflow run so its IDs can be embedded in test labels before CSV generation. +MLFLOW_PRECREATED_RUN_MARKER = "__mlflow_precreated_run__.yaml" +"""Marker file written by the test step to communicate the pre-created MLflow +run ID and experiment ID to the export step. + +The test step pre-creates an MLflow run (immediately ended) so that the +``run_id`` and ``experiment_id`` are known *before* CSV generation. These IDs +are embedded in test labels and flow into the dashboard CSV. The export step +later discovers the marker via :func:`_discover_precreated_mlflow_run_id` and +resumes the same run (``mlflow.start_run(run_id=...)``) to upload artifacts, +avoiding a duplicate run. + +The marker is a YAML file with keys ``run_id`` and ``experiment_id``, written +to ``ARTIFACT_DIR`` by :func:`precreate_mlflow_run`. +""" - The run is created and immediately ended (status FINISHED). The export step - will resume it via ``mlflow.start_run(run_id=...)`` to upload artifacts. - Returns a dict with ``run_id`` and ``experiment_id`` on success, - or an empty dict if MLflow is unavailable. - """ - from projects.core.library import config +def precreate_mlflow_run( + *, + secrets_path: Path, + experiment: str | None = None, + workspace: str | None = None, +) -> dict[str, str]: + """Pre-create an MLflow run and write the marker file to ``ARTIFACT_DIR``. - vault_name = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False - ) - vault_key = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", None, print=False, warn=False - ) - if not vault_name or not vault_key: - logger.info("MLflow vault not configured, skipping run pre-creation") - return {} + The run is created and immediately ended (status FINISHED). The export step + will resume it via ``mlflow.start_run(run_id=...)`` to upload artifacts. - secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) - if not secrets_path or not secrets_path.exists(): - logger.info("MLflow secrets file not found, skipping run pre-creation") - return {} - - experiment = config.project.get_config( - "caliper.export.backend.mlflow.config.experiment", None, print=False, warn=False - ) - workspace = config.project.get_config( - "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False - ) + The caller (test harness) is responsible for reading config and vault paths; + this function does not access the project config directly. + Returns a dict with ``run_id`` and ``experiment_id``. + """ import mlflow - from projects.caliper.engine.file_export.mlflow_secrets import ( + from projects.caliper.public.file_export import ( load_mlflow_secrets_yaml, mlflow_connection_env, ) @@ -310,6 +309,7 @@ def precreate_mlflow_run() -> dict[str, str]: tracking_uri = secrets_data.get("tracking_uri", "") prev_workspace = os.environ.get("MLFLOW_WORKSPACE") + prev_tracking_uri = mlflow.get_tracking_uri() try: with mlflow_connection_env(secrets_data): if tracking_uri: @@ -329,84 +329,77 @@ def precreate_mlflow_run() -> dict[str, str]: os.environ["MLFLOW_WORKSPACE"] = prev_workspace else: os.environ.pop("MLFLOW_WORKSPACE", None) + mlflow.set_tracking_uri(prev_tracking_uri) - logger.info("Pre-created MLflow run %s (experiment=%s)", run_id, experiment_id) - - return { - "run_id": run_id, - "experiment_id": experiment_id, - } - - -MLFLOW_PRECREATED_RUN_MARKER = "__mlflow_precreated_run__.yaml" + meta = {"run_id": run_id, "experiment_id": experiment_id} - -def write_mlflow_precreated_run_marker(meta: dict[str, str]) -> None: - """Persist the pre-created MLflow run_id to disk so the export step can resume it.""" marker_path = env.ARTIFACT_DIR / MLFLOW_PRECREATED_RUN_MARKER marker_path.parent.mkdir(parents=True, exist_ok=True) marker_path.write_text(yaml.safe_dump(meta, sort_keys=False), encoding="utf-8") - logger.info("Wrote MLflow pre-created run marker: %s", marker_path) + logger.info( + "Pre-created MLflow run %s (experiment=%s), marker: %s", + run_id, + experiment_id, + marker_path, + ) + + return meta def _read_mlflow_ids_from_marker() -> tuple[str, str]: """Read run_id and experiment_id from the pre-created marker file on disk.""" - try: - artifact_dir = Path(env.ARTIFACT_DIR) if env.ARTIFACT_DIR else None - if not artifact_dir: - return "", "" - for marker in sorted(artifact_dir.rglob(MLFLOW_PRECREATED_RUN_MARKER)): - data = yaml.safe_load(marker.read_text(encoding="utf-8")) - if isinstance(data, dict) and data.get("run_id"): - return data["run_id"], data.get("experiment_id", "") - except Exception: - pass + artifact_dir = Path(env.ARTIFACT_DIR) if env.ARTIFACT_DIR else None + if not artifact_dir: + logger.warning("ARTIFACT_DIR not set, cannot read MLflow marker") + return "", "" + for marker in sorted(artifact_dir.rglob(MLFLOW_PRECREATED_RUN_MARKER)): + data = yaml.safe_load(marker.read_text(encoding="utf-8")) + if not isinstance(data, dict): + logger.warning( + "MLflow marker %s has unexpected type %s, skipping", marker, type(data).__name__ + ) + continue + if data.get("run_id"): + return data["run_id"], data.get("experiment_id", "") + logger.warning("No valid MLflow marker found under %s", artifact_dir) return "", "" -def build_mlflow_run_url() -> str: - """Construct the MLflow run URL at runtime from vault secrets and the marker file.""" - try: - from projects.caliper.engine.file_export.mlflow_secrets import ( - assert_tracking_uri_has_no_userinfo, - load_mlflow_secrets_yaml, - ) - from projects.core.library import config +def build_mlflow_run_url( + *, + secrets_path: Path, + workspace: str | None = None, +) -> str: + """Construct the MLflow run URL from vault secrets and the marker file. - run_id, experiment_id = _read_mlflow_ids_from_marker() - if not run_id or not experiment_id: - return "" + The caller (test harness) is responsible for resolving ``secrets_path`` + and ``workspace``; this function does not access the project config. + """ + from urllib.parse import quote - vault_name = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False - ) - vault_key = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", - None, - print=False, - warn=False, - ) - if not vault_name or not vault_key: - return "" + from projects.caliper.public.file_export import ( + assert_tracking_uri_has_no_userinfo, + load_mlflow_secrets_yaml, + ) - secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) - if not secrets_path or not secrets_path.exists(): - return "" + run_id, experiment_id = _read_mlflow_ids_from_marker() + if not run_id or not experiment_id: + logger.warning("Cannot build MLflow URL: run_id or experiment_id missing from marker") + 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 "" - assert_tracking_uri_has_no_userinfo(tracking_uri) + if not secrets_path.exists(): + logger.warning("Cannot build MLflow URL: secrets file %s not found", secrets_path) + return "" - workspace = config.project.get_config( - "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False - ) - qs = f"?workspace={workspace}" if workspace else "" - return f"{tracking_uri}/#/experiments/{experiment_id}/runs/{run_id}/artifacts{qs}" - except Exception: - logger.debug("Failed to build MLflow run URL", exc_info=True) + secrets_data = load_mlflow_secrets_yaml(secrets_path) + tracking_uri = secrets_data.get("tracking_uri", "").rstrip("/") + if not tracking_uri.startswith(("http://", "https://")): + logger.warning("Cannot build MLflow URL: tracking_uri has unsupported scheme") return "" + assert_tracking_uri_has_no_userinfo(tracking_uri) + + qs = f"?workspace={quote(workspace, safe='')}" if workspace else "" + return f"{tracking_uri}/#/experiments/{experiment_id}/runs/{run_id}/artifacts{qs}" def _discover_precreated_mlflow_run_id(from_path: Path) -> str | None: @@ -414,7 +407,14 @@ def _discover_precreated_mlflow_run_id(from_path: Path) -> str | None: for marker in sorted(from_path.rglob(MLFLOW_PRECREATED_RUN_MARKER)): try: data = yaml.safe_load(marker.read_text(encoding="utf-8")) - run_id = data.get("run_id") if isinstance(data, dict) else None + if not isinstance(data, dict): + logger.warning( + "MLflow marker %s has unexpected type %s, skipping", + marker, + type(data).__name__, + ) + continue + run_id = data.get("run_id") if run_id: logger.info("Found pre-created MLflow run_id: %s (from marker %s)", run_id, marker) return run_id diff --git a/projects/caliper/public/__init__.py b/projects/caliper/public/__init__.py new file mode 100644 index 000000000..6ef3f2a54 --- /dev/null +++ b/projects/caliper/public/__init__.py @@ -0,0 +1 @@ +"""Public API surface for caliper — safe for orchestration-layer imports.""" diff --git a/projects/caliper/public/file_export.py b/projects/caliper/public/file_export.py new file mode 100644 index 000000000..a8c5723a6 --- /dev/null +++ b/projects/caliper/public/file_export.py @@ -0,0 +1,17 @@ +"""Public re-exports of caliper engine file_export utilities. + +Orchestration code should import from here instead of reaching into +``projects.caliper.engine.file_export`` directly. +""" + +from projects.caliper.engine.file_export.mlflow_secrets import ( + assert_tracking_uri_has_no_userinfo, + load_mlflow_secrets_yaml, + mlflow_connection_env, +) + +__all__ = [ + "assert_tracking_uri_has_no_userinfo", + "load_mlflow_secrets_yaml", + "mlflow_connection_env", +] diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 4bcc1c520..51fc53d7f 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -9,6 +9,7 @@ import yaml from projects.core.library import env +from projects.core.library import vault as vault_lib from projects.core.library.postprocess import run_and_postprocess, write_test_labels from projects.rhaiis.orchestration import runtime_config @@ -296,23 +297,44 @@ def _run_test( _warnings.append("Profiler trace upload failed") # Pre-create MLflow run so its URL can be embedded in test labels / CSV - from projects.caliper.orchestration.export import ( - precreate_mlflow_run, - write_mlflow_precreated_run_marker, - ) - mlflow_run_meta: dict[str, str] = {} try: - mlflow_run_meta = precreate_mlflow_run() + vault_name = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False + ) + vault_key = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", + None, + print=False, + warn=False, + ) + if vault_name and vault_key: + secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) + if secrets_path and secrets_path.exists(): + from projects.caliper.orchestration.export import precreate_mlflow_run + + mlflow_run_meta = precreate_mlflow_run( + secrets_path=secrets_path, + experiment=config.project.get_config( + "caliper.export.backend.mlflow.config.experiment", + None, + print=False, + warn=False, + ), + workspace=config.project.get_config( + "caliper.export.backend.mlflow.config.workspace", + None, + print=False, + warn=False, + ), + ) + else: + logger.info("MLflow secrets file not found, skipping run pre-creation") + else: + logger.info("MLflow vault not configured, skipping run pre-creation") except Exception: logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) - if mlflow_run_meta.get("run_id"): - try: - write_mlflow_precreated_run_marker(mlflow_run_meta) - except Exception: - logger.warning("Failed to write MLflow marker file; continuing", exc_info=True) - # Phase 2: benchmark + post-processing for ALL workloads trtllm_cfg = runtime_config.get_trtllm_config() if engine == "trtllm" else None for wl_key in workload_keys: diff --git a/projects/rhaiis/postprocess/regression.py b/projects/rhaiis/postprocess/regression.py index 565da1af1..e93b84edf 100644 --- a/projects/rhaiis/postprocess/regression.py +++ b/projects/rhaiis/postprocess/regression.py @@ -335,8 +335,29 @@ def run_regression_analysis( def _build_mlflow_run_url() -> str: """Construct the MLflow run URL at runtime from vault secrets and config.""" from projects.caliper.orchestration.export import build_mlflow_run_url + from projects.core.library import config + from projects.core.library import vault as vault_lib - return build_mlflow_run_url() + vault_name = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False + ) + vault_key = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", None, print=False, warn=False + ) + if not vault_name or not vault_key: + logger.warning("Cannot build MLflow URL: vault not configured") + return "" + + secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) + if not secrets_path or not secrets_path.exists(): + logger.warning("Cannot build MLflow URL: secrets file not found") + return "" + + workspace = config.project.get_config( + "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False + ) + + return build_mlflow_run_url(secrets_path=secrets_path, workspace=workspace) PROFILE_DISPLAY_NAMES = { From 2b951ca2f616ab039c66855290cadeddd0e168fd Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Tue, 4 Aug 2026 15:05:34 -0400 Subject: [PATCH 17/24] fix: send success notification regardless of dashboard CSV config The slack_notify_always notification was inside _sync_postprocessed_dashboard_csv() which returns early when csv_dashboard.enabled is false. Move to a standalone function called from run() so it fires independently. Co-authored-by: Cursor --- projects/rhaiis/orchestration/test_phase.py | 59 +++++++++++++-------- 1 file changed, 37 insertions(+), 22 deletions(-) diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 51fc53d7f..232938c68 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -54,6 +54,9 @@ def run( logger.exception("Dashboard CSV S3 sync after postprocessing failed") ret = 1 + if ret == 0: + _maybe_send_success_notification(model_key, workload_keys) + return ret @@ -663,30 +666,42 @@ def _sync_postprocessed_dashboard_csv(model_key: str, workload_keys: list[str]) ) return - if config.project.get_config("tests.rhaiis.slack_notify_always", False): - from projects.rhaiis.postprocess.regression import send_success_notification - model_cfg = runtime_config.get_model(model_key) - accelerator = runtime_config.get_accelerator() - engine = runtime_config.get_engine() - engine_defaults = runtime_config.get_engine_args(engine) - first_workload = runtime_config.get_workload(workload_keys[0]) - ea = runtime_config.merge_engine_args(engine_defaults, model_cfg, first_workload, engine) - tp = ea.get("tensor-parallel-size") or ea.get("tp-size") or ea.get("tp_size") or "" - dp = ea.get("data-parallel-size") or ea.get("dp-size") or "" +def _maybe_send_success_notification(model_key: str, workload_keys: list[str]) -> None: + """Send a Slack success notification if slack_notify_always is set and no regression check.""" + from projects.core.library import config - send_success_notification( - model=model_cfg.get("hf_model_id", ""), - accelerator=accelerator, - job_id=os.environ.get("FJOB_NAME", ""), - slack_user=config.project.get_config("tests.rhaiis.slack_user", ""), - notification_vault="psap-forge-notifications", - tp=str(tp), - dp=str(dp), - version=version, - workload_keys=workload_keys, - cluster=config.project.get_config("rhaiis.cluster_tag", ""), - ) + compare_version = config.project.get_config("tests.rhaiis.compare_version", "") + if compare_version: + return + + if not config.project.get_config("tests.rhaiis.slack_notify_always", False): + return + + from projects.rhaiis.postprocess.regression import send_success_notification + + model_cfg = runtime_config.get_model(model_key) + accelerator = runtime_config.get_accelerator() + engine = runtime_config.get_engine() + engine_defaults = runtime_config.get_engine_args(engine) + first_workload = runtime_config.get_workload(workload_keys[0]) + ea = runtime_config.merge_engine_args(engine_defaults, model_cfg, first_workload, engine) + tp = ea.get("tensor-parallel-size") or ea.get("tp-size") or ea.get("tp_size") or "" + dp = ea.get("data-parallel-size") or ea.get("dp-size") or "" + version = config.project.get_config("tests.rhaiis.version", "") + + send_success_notification( + model=model_cfg.get("hf_model_id", ""), + accelerator=accelerator, + job_id=os.environ.get("FJOB_NAME", ""), + slack_user=config.project.get_config("tests.rhaiis.slack_user", ""), + notification_vault="psap-forge-notifications", + tp=str(tp), + dp=str(dp), + version=version, + workload_keys=workload_keys, + cluster=config.project.get_config("rhaiis.cluster_tag", ""), + ) def _upload_predictor_log(run_uuid: str) -> None: From a5c658d682d89f270b3f326160cd5e032392168b Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Tue, 4 Aug 2026 15:11:19 -0400 Subject: [PATCH 18/24] fix: log warning instead of silently swallowing exception in send_success_notification Co-authored-by: Cursor --- projects/rhaiis/postprocess/regression.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/projects/rhaiis/postprocess/regression.py b/projects/rhaiis/postprocess/regression.py index e93b84edf..6d4096b4f 100644 --- a/projects/rhaiis/postprocess/regression.py +++ b/projects/rhaiis/postprocess/regression.py @@ -603,7 +603,7 @@ def send_success_notification( ) dashboard_line = f"*Dashboard:* <{dashboard_url}|View Dashboard>\n" except Exception: - pass + logger.warning("Failed to build dashboard URL for notification", exc_info=True) mlflow_url = _build_mlflow_run_url() mlflow_line = f"*MLflow:* <{mlflow_url}|View Run>\n" if mlflow_url else "" From 01d220fa540f1e86b9fbc37208327e1178301a91 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Tue, 4 Aug 2026 15:49:16 -0400 Subject: [PATCH 19/24] revert .gitignore to upstream state Move personal ignore patterns (fournos-job-*.yaml, kubeconfig*) to .git/info/exclude per reviewer feedback. Co-authored-by: Cursor --- .gitignore | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/.gitignore b/.gitignore index 565ad2d5b..80ea2d510 100644 --- a/.gitignore +++ b/.gitignore @@ -162,7 +162,4 @@ Thumbs.db temp/ # FORGE launcher configuration (personal settings) -projects/core/launcher/launcher_config.yaml - -fournos-job-*.yaml -kubeconfig* \ No newline at end of file +projects/core/launcher/launcher_config.yaml \ No newline at end of file From 53136f1c967bc976763ec0e25574b179e26bf0ca Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Wed, 5 Aug 2026 10:39:54 -0400 Subject: [PATCH 20/24] refactor: use mlflow_destination in test labels instead of marker file - Extract MLflow pre-creation to _precreate_mlflow_run_if_configured() with guard pattern for readability - Move mlflow IDs out of labels into a separate mlflow_destination section in __test_labels__.yaml (run_id, experiment_id, workspace) - Remove __mlflow_precreated_run__.yaml marker file; export step and URL builder now read from mlflow_destination in test labels - Add conflict warning when export config and test labels both have differing run_ids - Update test-labels-format.md with mlflow_destination docs Co-authored-by: Cursor --- docs/caliper/test-labels-format.md | 8 ++ projects/caliper/orchestration/export.py | 99 ++++++++--------- projects/core/library/postprocess.py | 5 + projects/rhaiis/orchestration/test_phase.py | 111 +++++++++++--------- projects/rhaiis/postprocess/plugin.py | 30 +++++- 5 files changed, 147 insertions(+), 106 deletions(-) diff --git a/docs/caliper/test-labels-format.md b/docs/caliper/test-labels-format.md index bc2fbec3c..7f5316807 100644 --- a/docs/caliper/test-labels-format.md +++ b/docs/caliper/test-labels-format.md @@ -32,6 +32,10 @@ completion: - **`platform`**: Platform name (e.g., `"CKS"`, `"RHOAI"`) - **`gpu_type`**: GPU type (e.g., `"H100"`, `"A100"`) - **`test_harness`**: Test framework (e.g., `"guidellm"`, `"vllm"`) +- **`mlflow_destination`**: Pre-created MLflow run for artifact upload + - **`run_id`**: MLflow run ID (assigned by the server during pre-creation) + - **`experiment_id`**: MLflow experiment ID + - **`workspace`**: MLflow workspace name (optional) - **`completion`**: Test execution status - **`success`**: `true` if succeeded, `false` if failed - **`message`**: Human-readable status description @@ -89,6 +93,10 @@ kpi_labels: platform: "CKS" gpu_type: "H100" test_harness: "guidellm" +mlflow_destination: + run_id: "48e49dfc966c487cb76cf105a5314908" + experiment_id: "264" + workspace: "forge-rhaiis" completion: success: true message: "Test completed successfully" diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index ac1cb0b5d..f675f2b7c 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -206,8 +206,19 @@ def run_from_orchestration_config( run_dirs = discover_run_dirs(from_path) - # Resume a pre-created MLflow run if the test step left a marker or test labels - mlflow_run_id = export_cfg.mlflow_run_id or _discover_precreated_mlflow_run_id(from_path) + # Resume a pre-created MLflow run if the test step left mlflow_destination in test labels + discovered_run_id = _discover_precreated_mlflow_run_id(from_path) + if ( + export_cfg.mlflow_run_id + and discovered_run_id + and export_cfg.mlflow_run_id != discovered_run_id + ): + logger.error( + "Conflicting MLflow run_ids: export config has %s, test labels have %s. Using export config.", + export_cfg.mlflow_run_id, + discovered_run_id, + ) + mlflow_run_id = export_cfg.mlflow_run_id or discovered_run_id # Resolve descriptive run names from labels + run_naming config naming = resolve_run_names( @@ -266,20 +277,7 @@ def run_from_orchestration_config( return yaml.safe_load(f.read()) -MLFLOW_PRECREATED_RUN_MARKER = "__mlflow_precreated_run__.yaml" -"""Marker file written by the test step to communicate the pre-created MLflow -run ID and experiment ID to the export step. - -The test step pre-creates an MLflow run (immediately ended) so that the -``run_id`` and ``experiment_id`` are known *before* CSV generation. These IDs -are embedded in test labels and flow into the dashboard CSV. The export step -later discovers the marker via :func:`_discover_precreated_mlflow_run_id` and -resumes the same run (``mlflow.start_run(run_id=...)``) to upload artifacts, -avoiding a duplicate run. - -The marker is a YAML file with keys ``run_id`` and ``experiment_id``, written -to ``ARTIFACT_DIR`` by :func:`precreate_mlflow_run`. -""" +TEST_LABELS_FILENAME = "__test_labels__.yaml" def precreate_mlflow_run( @@ -288,13 +286,13 @@ def precreate_mlflow_run( experiment: str | None = None, workspace: str | None = None, ) -> dict[str, str]: - """Pre-create an MLflow run and write the marker file to ``ARTIFACT_DIR``. + """Pre-create an MLflow run so the export step can resume it. The run is created and immediately ended (status FINISHED). The export step will resume it via ``mlflow.start_run(run_id=...)`` to upload artifacts. - The caller (test harness) is responsible for reading config and vault paths; - this function does not access the project config directly. + The caller is responsible for persisting the returned IDs (e.g. via the + ``mlflow_destination`` section of ``__test_labels__.yaml``). Returns a dict with ``run_id`` and ``experiment_id``. """ @@ -333,35 +331,30 @@ def precreate_mlflow_run( meta = {"run_id": run_id, "experiment_id": experiment_id} - marker_path = env.ARTIFACT_DIR / MLFLOW_PRECREATED_RUN_MARKER - marker_path.parent.mkdir(parents=True, exist_ok=True) - marker_path.write_text(yaml.safe_dump(meta, sort_keys=False), encoding="utf-8") - logger.info( - "Pre-created MLflow run %s (experiment=%s), marker: %s", - run_id, - experiment_id, - marker_path, - ) + logger.info("Pre-created MLflow run %s (experiment=%s)", run_id, experiment_id) return meta -def _read_mlflow_ids_from_marker() -> tuple[str, str]: - """Read run_id and experiment_id from the pre-created marker file on disk.""" +def _read_mlflow_ids_from_test_labels() -> tuple[str, str]: + """Read run_id and experiment_id from ``mlflow_destination`` in test labels.""" artifact_dir = Path(env.ARTIFACT_DIR) if env.ARTIFACT_DIR else None if not artifact_dir: - logger.warning("ARTIFACT_DIR not set, cannot read MLflow marker") + logger.warning("ARTIFACT_DIR not set, cannot read MLflow destination from test labels") return "", "" - for marker in sorted(artifact_dir.rglob(MLFLOW_PRECREATED_RUN_MARKER)): - data = yaml.safe_load(marker.read_text(encoding="utf-8")) - if not isinstance(data, dict): - logger.warning( - "MLflow marker %s has unexpected type %s, skipping", marker, type(data).__name__ - ) - continue - if data.get("run_id"): - return data["run_id"], data.get("experiment_id", "") - logger.warning("No valid MLflow marker found under %s", artifact_dir) + for labels_file in sorted(artifact_dir.rglob(TEST_LABELS_FILENAME)): + try: + data = yaml.safe_load(labels_file.read_text(encoding="utf-8")) + if not isinstance(data, dict): + continue + dest = data.get("mlflow_destination") + if not isinstance(dest, dict): + continue + run_id = dest.get("run_id", "") + if run_id: + return run_id, dest.get("experiment_id", "") + except (OSError, yaml.YAMLError) as e: + logger.warning("Failed to read test labels %s: %s", labels_file, e) return "", "" @@ -382,9 +375,9 @@ def build_mlflow_run_url( load_mlflow_secrets_yaml, ) - run_id, experiment_id = _read_mlflow_ids_from_marker() + run_id, experiment_id = _read_mlflow_ids_from_test_labels() if not run_id or not experiment_id: - logger.warning("Cannot build MLflow URL: run_id or experiment_id missing from marker") + logger.warning("Cannot build MLflow URL: run_id or experiment_id missing from test labels") return "" if not secrets_path.exists(): @@ -403,22 +396,20 @@ def build_mlflow_run_url( def _discover_precreated_mlflow_run_id(from_path: Path) -> str | None: - """Find a pre-created MLflow run_id from the marker file.""" - for marker in sorted(from_path.rglob(MLFLOW_PRECREATED_RUN_MARKER)): + """Find a pre-created MLflow run_id from ``mlflow_destination`` in test labels.""" + for labels_file in sorted(from_path.rglob(TEST_LABELS_FILENAME)): try: - data = yaml.safe_load(marker.read_text(encoding="utf-8")) + data = yaml.safe_load(labels_file.read_text(encoding="utf-8")) if not isinstance(data, dict): - logger.warning( - "MLflow marker %s has unexpected type %s, skipping", - marker, - type(data).__name__, - ) continue - run_id = data.get("run_id") + dest = data.get("mlflow_destination") + if not isinstance(dest, dict): + continue + run_id = dest.get("run_id") if run_id: - logger.info("Found pre-created MLflow run_id: %s (from marker %s)", run_id, marker) + logger.info("Found pre-created MLflow run_id: %s (from %s)", run_id, labels_file) return run_id except (OSError, yaml.YAMLError) as e: - logger.warning("Failed to read MLflow pre-created run marker %s: %s", marker, e) + logger.warning("Failed to read test labels %s: %s", labels_file, e) return None diff --git a/projects/core/library/postprocess.py b/projects/core/library/postprocess.py index 8eccb5af4..9a2f9201d 100644 --- a/projects/core/library/postprocess.py +++ b/projects/core/library/postprocess.py @@ -39,6 +39,7 @@ def write_test_labels( version: str = "1", dump_config: bool = True, kpi_labels: dict[str, str] | None = None, + mlflow_destination: dict[str, str] | None = None, ) -> Path: """Write a __test_labels__.yaml file to mark a directory as a Caliper test base. @@ -48,6 +49,7 @@ def write_test_labels( version: Version string for the test labels format (default: "1") dump_config: Whether to save project configuration to config.yaml (default: True) kpi_labels: Optional dictionary of KPI labels for system context + mlflow_destination: Optional MLflow run destination (run_id, experiment_id, workspace) Returns: Path to the created __test_labels__.yaml file @@ -75,6 +77,9 @@ def write_test_labels( if kpi_labels: payload["kpi_labels"] = kpi_labels + if mlflow_destination: + payload["mlflow_destination"] = mlflow_destination + # Create directory and write YAML test_labels_path.parent.mkdir(parents=True, exist_ok=True) with test_labels_path.open("w", encoding="utf-8") as handle: diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 232938c68..cf9b641b1 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -9,7 +9,6 @@ import yaml from projects.core.library import env -from projects.core.library import vault as vault_lib from projects.core.library.postprocess import run_and_postprocess, write_test_labels from projects.rhaiis.orchestration import runtime_config @@ -299,44 +298,7 @@ def _run_test( logger.exception("Profiler trace upload failed") _warnings.append("Profiler trace upload failed") - # Pre-create MLflow run so its URL can be embedded in test labels / CSV - mlflow_run_meta: dict[str, str] = {} - try: - vault_name = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False - ) - vault_key = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", - None, - print=False, - warn=False, - ) - if vault_name and vault_key: - secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) - if secrets_path and secrets_path.exists(): - from projects.caliper.orchestration.export import precreate_mlflow_run - - mlflow_run_meta = precreate_mlflow_run( - secrets_path=secrets_path, - experiment=config.project.get_config( - "caliper.export.backend.mlflow.config.experiment", - None, - print=False, - warn=False, - ), - workspace=config.project.get_config( - "caliper.export.backend.mlflow.config.workspace", - None, - print=False, - warn=False, - ), - ) - else: - logger.info("MLflow secrets file not found, skipping run pre-creation") - else: - logger.info("MLflow vault not configured, skipping run pre-creation") - except Exception: - logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) + mlflow_run_meta = _precreate_mlflow_run_if_configured() # Phase 2: benchmark + post-processing for ALL workloads trtllm_cfg = runtime_config.get_trtllm_config() if engine == "trtllm" else None @@ -359,8 +321,7 @@ def _run_test( version=version, cluster_tag=cluster_tag, trtllm_config=trtllm_cfg, - mlflow_run_id=mlflow_run_meta.get("run_id", ""), - mlflow_experiment_id=mlflow_run_meta.get("experiment_id", ""), + mlflow_run_meta=mlflow_run_meta or None, ) try: @@ -427,8 +388,7 @@ def _run_workload_benchmark( version: str, cluster_tag: str, trtllm_config: dict | None = None, - mlflow_run_id: str = "", - mlflow_experiment_id: str = "", + mlflow_run_meta: dict[str, str] | None = None, ) -> None: """Run benchmark and post-processing for a single workload. @@ -462,8 +422,7 @@ def _run_workload_benchmark( accelerator_chip=gpu_type.upper(), run_uuid=run_uuid, trtllm_config=trtllm_config, - mlflow_run_id=mlflow_run_id, - mlflow_experiment_id=mlflow_experiment_id, + mlflow_run_meta=mlflow_run_meta, ) if not run_benchmark: @@ -507,6 +466,48 @@ def _run_workload_benchmark( ) +def _precreate_mlflow_run_if_configured() -> dict[str, str]: + """Pre-create an MLflow run so its IDs can be embedded in test labels. + + Uses the guard pattern with early returns to avoid deep nesting. + Returns a dict with run_id and experiment_id, or empty dict on failure. + """ + from projects.core.library import config + from projects.core.library import vault as vault_lib + + vault_name = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False + ) + vault_key = config.project.get_config( + "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", None, print=False, warn=False + ) + if not vault_name or not vault_key: + logger.info("MLflow vault not configured, skipping run pre-creation") + return {} + + secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) + if not secrets_path or not secrets_path.exists(): + logger.info("MLflow secrets file not found, skipping run pre-creation") + return {} + + experiment = config.project.get_config( + "caliper.export.backend.mlflow.config.experiment", None, print=False, warn=False + ) + workspace = config.project.get_config( + "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False + ) + + try: + from projects.caliper.orchestration.export import precreate_mlflow_run + + return precreate_mlflow_run( + secrets_path=secrets_path, experiment=experiment, workspace=workspace + ) + except Exception: + logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) + return {} + + def _create_test_labels( model_key: str, workload_key: str, @@ -520,8 +521,7 @@ def _create_test_labels( accelerator_chip: str = "", run_uuid: str = "", trtllm_config: dict | None = None, - mlflow_run_id: str = "", - mlflow_experiment_id: str = "", + mlflow_run_meta: dict[str, str] | None = None, ) -> None: _, image_tag = runtime_config.split_image_tag(serving_image) if serving_image else ("", "") parts = [f"{k}: {v}" for k, v in engine_args.items()] @@ -548,10 +548,21 @@ def _create_test_labels( "cluster_tag": cluster_tag, "runtime_args": runtime_args, "run_uuid": run_uuid, - "mlflow_run_id": mlflow_run_id, - "mlflow_experiment_id": mlflow_experiment_id, } - write_test_labels(env.ARTIFACT_DIR, labels) + + mlflow_destination = None + if mlflow_run_meta and mlflow_run_meta.get("run_id"): + from projects.core.library import config + + mlflow_destination = { + "run_id": mlflow_run_meta["run_id"], + "experiment_id": mlflow_run_meta.get("experiment_id", ""), + "workspace": config.project.get_config( + "caliper.export.backend.mlflow.config.workspace", "", print=False, warn=False + ), + } + + write_test_labels(env.ARTIFACT_DIR, labels, mlflow_destination=mlflow_destination) logger.info("Created test labels: %s", labels) diff --git a/projects/rhaiis/postprocess/plugin.py b/projects/rhaiis/postprocess/plugin.py index 803cbf68a..9071b946d 100644 --- a/projects/rhaiis/postprocess/plugin.py +++ b/projects/rhaiis/postprocess/plugin.py @@ -18,6 +18,30 @@ logger = logging.getLogger(__name__) + +def _read_mlflow_destination_from_test_labels(kpi_records: list[dict]) -> dict[str, str]: + """Extract mlflow_destination from __test_labels__.yaml near the first KPI run_path.""" + import yaml + + for kpi in kpi_records: + run_path = kpi.get("run_path", "") + if not run_path: + continue + labels_file = Path(run_path) / "__test_labels__.yaml" + if not labels_file.exists(): + labels_file = Path(run_path).parent / "__test_labels__.yaml" + if not labels_file.exists(): + continue + try: + data = yaml.safe_load(labels_file.read_text(encoding="utf-8")) + dest = data.get("mlflow_destination") if isinstance(data, dict) else None + if isinstance(dest, dict) and dest.get("run_id"): + return dest + except (OSError, yaml.YAMLError): + pass + return {} + + # 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( @@ -128,6 +152,8 @@ def export_kpis_to_csv( """ from projects.rhaiis.postprocess.csv_export import FIELDNAMES + mlflow_dest = _read_mlflow_destination_from_test_labels(kpi_records) + # 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]] = {} @@ -191,8 +217,8 @@ def export_kpis_to_csv( 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", "") + row["mlflow_run_id"] = mlflow_dest.get("run_id", "") + row["mlflow_experiment_id"] = mlflow_dest.get("experiment_id", "") rows.append(row) output_path.parent.mkdir(parents=True, exist_ok=True) From f7aa64cd2c01fbcf13d83498591c70286b610d4b Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Thu, 6 Aug 2026 11:59:08 -0400 Subject: [PATCH 21/24] refactor: move precreate_mlflow_run_if_configured to caliper with @requires - Use @requires decorator for config access, matching mlflow_verifier pattern - Return full mlflow_destination dict (run_id, experiment_id, workspace) - Remove function from rhaiis test_phase.py; import from caliper - Mark mlflow_destination and experiment_id as optional in docs Co-authored-by: Cursor --- docs/caliper/test-labels-format.md | 6 +- projects/caliper/orchestration/export.py | 42 +++++++++++++ projects/rhaiis/orchestration/test_phase.py | 70 ++++----------------- 3 files changed, 56 insertions(+), 62 deletions(-) diff --git a/docs/caliper/test-labels-format.md b/docs/caliper/test-labels-format.md index 7f5316807..44ddb3edb 100644 --- a/docs/caliper/test-labels-format.md +++ b/docs/caliper/test-labels-format.md @@ -32,10 +32,10 @@ completion: - **`platform`**: Platform name (e.g., `"CKS"`, `"RHOAI"`) - **`gpu_type`**: GPU type (e.g., `"H100"`, `"A100"`) - **`test_harness`**: Test framework (e.g., `"guidellm"`, `"vllm"`) -- **`mlflow_destination`**: Pre-created MLflow run for artifact upload +- **`mlflow_destination`** *(optional)*: Pre-created MLflow run for artifact upload - **`run_id`**: MLflow run ID (assigned by the server during pre-creation) - - **`experiment_id`**: MLflow experiment ID - - **`workspace`**: MLflow workspace name (optional) + - **`experiment_id`** *(optional)*: MLflow experiment ID + - **`workspace`** *(optional)*: MLflow workspace name - **`completion`**: Test execution status - **`success`**: `true` if succeeded, `false` if failed - **`message`**: Human-readable status description diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index f675f2b7c..7d3866ca3 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -29,6 +29,7 @@ ) from projects.core.library import env from projects.core.library import vault as vault_lib +from projects.core.library.config import requires logger = logging.getLogger(__name__) @@ -280,6 +281,47 @@ def run_from_orchestration_config( TEST_LABELS_FILENAME = "__test_labels__.yaml" +@requires( + vault_name="caliper.export.backend.mlflow.secrets.vault.name", + vault_key="caliper.export.backend.mlflow.secrets.vault.mlflow_secret", + experiment="caliper.export.backend.mlflow.config.experiment", + workspace="caliper.export.backend.mlflow.config.workspace", +) +def precreate_mlflow_run_if_configured(_cfg) -> dict[str, str] | None: + """Pre-create an MLflow run and return the ``mlflow_destination`` dict. + + Uses ``@requires`` to read vault and MLflow config from the project config. + Returns ``None`` if MLflow is not configured or pre-creation fails. + The returned dict contains ``run_id``, ``experiment_id``, and ``workspace``. + """ + vault_name = _cfg.vault_name + vault_key = _cfg.vault_key + if not vault_name or not vault_key: + logger.info("MLflow vault not configured, skipping run pre-creation") + return None + + secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) + if not secrets_path or not secrets_path.exists(): + logger.info("MLflow secrets file not found, skipping run pre-creation") + return None + + try: + meta = precreate_mlflow_run( + secrets_path=secrets_path, + experiment=_cfg.experiment or None, + workspace=_cfg.workspace or None, + ) + except Exception: + logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) + return None + + return { + "run_id": meta["run_id"], + "experiment_id": meta.get("experiment_id", ""), + "workspace": _cfg.workspace or "", + } + + def precreate_mlflow_run( *, secrets_path: Path, diff --git a/projects/rhaiis/orchestration/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 10c037a67..74892f1b1 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -298,7 +298,13 @@ def _run_test( logger.exception("Profiler trace upload failed") _warnings.append("Profiler trace upload failed") - mlflow_run_meta = _precreate_mlflow_run_if_configured() + 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 # Phase 2: benchmark + post-processing for ALL workloads trtllm_cfg = runtime_config.get_trtllm_config() if engine == "trtllm" else None @@ -321,7 +327,7 @@ def _run_test( version=version, cluster_tag=cluster_tag, trtllm_config=trtllm_cfg, - mlflow_run_meta=mlflow_run_meta or None, + mlflow_destination=mlflow_destination, ) try: @@ -388,7 +394,7 @@ def _run_workload_benchmark( version: str, cluster_tag: str, trtllm_config: dict | None = None, - mlflow_run_meta: dict[str, str] | None = None, + mlflow_destination: dict[str, str] | None = None, ) -> None: """Run benchmark and post-processing for a single workload. @@ -423,7 +429,7 @@ def _run_workload_benchmark( accelerator_chip=gpu_type.upper(), run_uuid=run_uuid, trtllm_config=trtllm_config, - mlflow_run_meta=mlflow_run_meta, + mlflow_destination=mlflow_destination, ) if not run_benchmark: @@ -468,48 +474,6 @@ def _run_workload_benchmark( ) -def _precreate_mlflow_run_if_configured() -> dict[str, str]: - """Pre-create an MLflow run so its IDs can be embedded in test labels. - - Uses the guard pattern with early returns to avoid deep nesting. - Returns a dict with run_id and experiment_id, or empty dict on failure. - """ - from projects.core.library import config - from projects.core.library import vault as vault_lib - - vault_name = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False - ) - vault_key = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", None, print=False, warn=False - ) - if not vault_name or not vault_key: - logger.info("MLflow vault not configured, skipping run pre-creation") - return {} - - secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) - if not secrets_path or not secrets_path.exists(): - logger.info("MLflow secrets file not found, skipping run pre-creation") - return {} - - experiment = config.project.get_config( - "caliper.export.backend.mlflow.config.experiment", None, print=False, warn=False - ) - workspace = config.project.get_config( - "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False - ) - - try: - from projects.caliper.orchestration.export import precreate_mlflow_run - - return precreate_mlflow_run( - secrets_path=secrets_path, experiment=experiment, workspace=workspace - ) - except Exception: - logger.warning("MLflow run pre-creation failed; continuing", exc_info=True) - return {} - - def _create_test_labels( model_key: str, workload_key: str, @@ -523,7 +487,7 @@ def _create_test_labels( accelerator_chip: str = "", run_uuid: str = "", trtllm_config: dict | None = None, - mlflow_run_meta: dict[str, str] | None = None, + mlflow_destination: dict[str, str] | None = None, ) -> None: _, image_tag = runtime_config.split_image_tag(serving_image) if serving_image else ("", "") parts = [f"{k}: {v}" for k, v in engine_args.items()] @@ -552,18 +516,6 @@ def _create_test_labels( "run_uuid": run_uuid, } - mlflow_destination = None - if mlflow_run_meta and mlflow_run_meta.get("run_id"): - from projects.core.library import config - - mlflow_destination = { - "run_id": mlflow_run_meta["run_id"], - "experiment_id": mlflow_run_meta.get("experiment_id", ""), - "workspace": config.project.get_config( - "caliper.export.backend.mlflow.config.workspace", "", print=False, warn=False - ), - } - write_test_labels(env.ARTIFACT_DIR, labels, mlflow_destination=mlflow_destination) logger.info("Created test labels: %s", labels) From d95b8604b71a9ccda9c87c42e2c4292241e50544 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Thu, 6 Aug 2026 13:20:31 -0400 Subject: [PATCH 22/24] fix: walk up directory tree to find test labels for mlflow_destination The KPI run_path points deep inside the benchmark results, several levels below where __test_labels__.yaml is written. Walk up from run_path instead of only checking the immediate parent. Co-authored-by: Cursor --- projects/rhaiis/postprocess/plugin.py | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/projects/rhaiis/postprocess/plugin.py b/projects/rhaiis/postprocess/plugin.py index b396c0e36..a52c3d521 100644 --- a/projects/rhaiis/postprocess/plugin.py +++ b/projects/rhaiis/postprocess/plugin.py @@ -20,25 +20,26 @@ def _read_mlflow_destination_from_test_labels(kpi_records: list[dict]) -> dict[str, str]: - """Extract mlflow_destination from __test_labels__.yaml near the first KPI run_path.""" + """Extract mlflow_destination from __test_labels__.yaml by walking up from KPI run_path.""" import yaml for kpi in kpi_records: run_path = kpi.get("run_path", "") if not run_path: continue - labels_file = Path(run_path) / "__test_labels__.yaml" - if not labels_file.exists(): - labels_file = Path(run_path).parent / "__test_labels__.yaml" - if not labels_file.exists(): - continue - try: - data = yaml.safe_load(labels_file.read_text(encoding="utf-8")) - dest = data.get("mlflow_destination") if isinstance(data, dict) else None - if isinstance(dest, dict) and dest.get("run_id"): - return dest - except (OSError, yaml.YAMLError): - pass + current = Path(run_path) + while current != current.parent: + labels_file = current / "__test_labels__.yaml" + if labels_file.exists(): + try: + data = yaml.safe_load(labels_file.read_text(encoding="utf-8")) + dest = data.get("mlflow_destination") if isinstance(data, dict) else None + if isinstance(dest, dict) and dest.get("run_id"): + return dest + except (OSError, yaml.YAMLError): + pass + break + current = current.parent return {} From e8df4379a02dcfcbc916ddda1ab3412aeb2d6cfe Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Thu, 6 Aug 2026 13:50:32 -0400 Subject: [PATCH 23/24] fix: pass mlflow IDs through KPI labels instead of filesystem reads run_path in KPI records is relative, so the CSV plugin subprocess cannot resolve __test_labels__.yaml on disk. Instead, extract mlflow_destination from test nodes during compute_kpis and include mlflow_run_id/mlflow_experiment_id in KPI record labels. Co-authored-by: Cursor --- projects/rhaiis/postprocess/kpis.py | 9 ++++++++ projects/rhaiis/postprocess/plugin.py | 30 ++------------------------- 2 files changed, 11 insertions(+), 28 deletions(-) diff --git a/projects/rhaiis/postprocess/kpis.py b/projects/rhaiis/postprocess/kpis.py index fc137feea..1e8c7ece2 100644 --- a/projects/rhaiis/postprocess/kpis.py +++ b/projects/rhaiis/postprocess/kpis.py @@ -63,6 +63,13 @@ 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 @@ -80,6 +87,8 @@ def compute_kpis(model: UnifiedRunModel) -> list[dict[str, Any]]: "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: diff --git a/projects/rhaiis/postprocess/plugin.py b/projects/rhaiis/postprocess/plugin.py index a52c3d521..0ae902860 100644 --- a/projects/rhaiis/postprocess/plugin.py +++ b/projects/rhaiis/postprocess/plugin.py @@ -19,30 +19,6 @@ logger = logging.getLogger(__name__) -def _read_mlflow_destination_from_test_labels(kpi_records: list[dict]) -> dict[str, str]: - """Extract mlflow_destination from __test_labels__.yaml by walking up from KPI run_path.""" - import yaml - - for kpi in kpi_records: - run_path = kpi.get("run_path", "") - if not run_path: - continue - current = Path(run_path) - while current != current.parent: - labels_file = current / "__test_labels__.yaml" - if labels_file.exists(): - try: - data = yaml.safe_load(labels_file.read_text(encoding="utf-8")) - dest = data.get("mlflow_destination") if isinstance(data, dict) else None - if isinstance(dest, dict) and dest.get("run_id"): - return dest - except (OSError, yaml.YAMLError): - pass - break - current = current.parent - return {} - - # 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( @@ -150,8 +126,6 @@ def export_kpis_to_csv( """ from projects.rhaiis.postprocess.csv_export import FIELDNAMES - mlflow_dest = _read_mlflow_destination_from_test_labels(kpi_records) - # 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]] = {} @@ -215,8 +189,8 @@ def export_kpis_to_csv( 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"] = mlflow_dest.get("run_id", "") - row["mlflow_experiment_id"] = mlflow_dest.get("experiment_id", "") + 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) From d905471a1895099c20430fc923dd7f11f75e88c7 Mon Sep 17 00:00:00 2001 From: Harshith-umesh Date: Thu, 6 Aug 2026 15:54:08 -0400 Subject: [PATCH 24/24] refactor: move _build_mlflow_run_url wrapper to caliper with @requires The config-aware wrapper only uses caliper config keys, so it belongs in caliper/orchestration/export.py where all projects can reuse it. regression.py now imports and calls build_mlflow_run_url_from_config(). Co-authored-by: Cursor --- projects/caliper/orchestration/export.py | 25 +++++++++++++++++++++ projects/rhaiis/postprocess/regression.py | 27 +++++------------------ 2 files changed, 30 insertions(+), 22 deletions(-) diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index 7d3866ca3..c9a5b4569 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -400,6 +400,31 @@ def _read_mlflow_ids_from_test_labels() -> tuple[str, str]: return "", "" +@requires( + vault_name="caliper.export.backend.mlflow.secrets.vault.name", + vault_key="caliper.export.backend.mlflow.secrets.vault.mlflow_secret", + workspace="caliper.export.backend.mlflow.config.workspace", +) +def build_mlflow_run_url_from_config(_cfg) -> str: + """Config-aware wrapper around :func:`build_mlflow_run_url`. + + Resolves vault secrets and workspace from project config via ``@requires``. + Returns an empty string if MLflow is not configured or URL cannot be built. + """ + vault_name = _cfg.vault_name + vault_key = _cfg.vault_key + if not vault_name or not vault_key: + logger.warning("Cannot build MLflow URL: vault not configured") + return "" + + secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) + if not secrets_path or not secrets_path.exists(): + logger.warning("Cannot build MLflow URL: secrets file not found") + return "" + + return build_mlflow_run_url(secrets_path=secrets_path, workspace=_cfg.workspace or None) + + def build_mlflow_run_url( *, secrets_path: Path, diff --git a/projects/rhaiis/postprocess/regression.py b/projects/rhaiis/postprocess/regression.py index 6d4096b4f..3f575ae9f 100644 --- a/projects/rhaiis/postprocess/regression.py +++ b/projects/rhaiis/postprocess/regression.py @@ -334,31 +334,14 @@ def run_regression_analysis( def _build_mlflow_run_url() -> str: """Construct the MLflow run URL at runtime from vault secrets and config.""" - from projects.caliper.orchestration.export import build_mlflow_run_url - from projects.core.library import config - from projects.core.library import vault as vault_lib + from projects.caliper.orchestration.export import build_mlflow_run_url_from_config - vault_name = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.name", None, print=False, warn=False - ) - vault_key = config.project.get_config( - "caliper.export.backend.mlflow.secrets.vault.mlflow_secret", None, print=False, warn=False - ) - if not vault_name or not vault_key: - logger.warning("Cannot build MLflow URL: vault not configured") - return "" - - secrets_path = vault_lib.get_vault_content_path(vault_name, vault_key) - if not secrets_path or not secrets_path.exists(): - logger.warning("Cannot build MLflow URL: secrets file not found") + try: + return build_mlflow_run_url_from_config() + except Exception: + logger.warning("Failed to build MLflow run URL", exc_info=True) return "" - workspace = config.project.get_config( - "caliper.export.backend.mlflow.config.workspace", None, print=False, warn=False - ) - - return build_mlflow_run_url(secrets_path=secrets_path, workspace=workspace) - PROFILE_DISPLAY_NAMES = { "profile1": "Profile A: Balanced (1k/1k)",