diff --git a/docs/caliper/test-labels-format.md b/docs/caliper/test-labels-format.md index bc2fbec3c..44ddb3edb 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`** *(optional)*: Pre-created MLflow run for artifact upload + - **`run_id`**: MLflow run ID (assigned by the server during pre-creation) + - **`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 @@ -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/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 b3edc9ed4..c058237aa 100644 --- a/projects/caliper/engine/file_export/mlflow_backend.py +++ b/projects/caliper/engine/file_export/mlflow_backend.py @@ -416,13 +416,15 @@ 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 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) @@ -535,6 +537,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, @@ -596,11 +599,15 @@ def _run(uri: str | None) -> tuple[str, dict[str, Any] | None]: client = mlflow.tracking.MlflowClient() start_kw: dict[str, Any] = {} + if run_id: + start_kw["run_id"] = run_id if parent_run_name: start_kw["run_name"] = parent_run_name 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 59a0f4c96..c9a5b4569 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__) @@ -206,6 +207,20 @@ def run_from_orchestration_config( run_dirs = discover_run_dirs(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( run_dirs, mlflow_config_data, fallback_run_name=export_cfg.mlflow_run_name @@ -226,6 +241,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=mlflow_run_id, child_run_names=naming.get("child_run_names") or {}, verbose=export_cfg.verbose, status_yaml_path=status_yaml, @@ -238,7 +254,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, } @@ -260,3 +276,207 @@ def run_from_orchestration_config( with open(status_yaml) as f: return yaml.safe_load(f.read()) + + +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, + experiment: str | None = None, + workspace: str | None = None, +) -> dict[str, str]: + """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 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``. + """ + import mlflow + + from projects.caliper.public.file_export 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") + prev_tracking_uri = mlflow.get_tracking_uri() + 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) + mlflow.set_tracking_uri(prev_tracking_uri) + + meta = {"run_id": run_id, "experiment_id": experiment_id} + + logger.info("Pre-created MLflow run %s (experiment=%s)", run_id, experiment_id) + + return meta + + +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 destination from test labels") + return "", "" + 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 "", "" + + +@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, + workspace: str | None = None, +) -> str: + """Construct the MLflow run URL from vault secrets and the marker file. + + 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 + + from projects.caliper.public.file_export import ( + assert_tracking_uri_has_no_userinfo, + load_mlflow_secrets_yaml, + ) + + 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 test labels") + return "" + + if not secrets_path.exists(): + logger.warning("Cannot build MLflow URL: secrets file %s not found", secrets_path) + return "" + + 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: + """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(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: + 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 test labels %s: %s", labels_file, e) + + return None 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/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/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/test_phase.py b/projects/rhaiis/orchestration/test_phase.py index 5c51d60b0..74892f1b1 100644 --- a/projects/rhaiis/orchestration/test_phase.py +++ b/projects/rhaiis/orchestration/test_phase.py @@ -53,6 +53,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 @@ -295,6 +298,14 @@ def _run_test( logger.exception("Profiler trace upload failed") _warnings.append("Profiler trace upload failed") + 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 for wl_key in workload_keys: @@ -316,6 +327,7 @@ def _run_test( version=version, cluster_tag=cluster_tag, trtllm_config=trtllm_cfg, + mlflow_destination=mlflow_destination, ) try: @@ -382,6 +394,7 @@ def _run_workload_benchmark( version: str, cluster_tag: str, trtllm_config: dict | None = None, + mlflow_destination: dict[str, str] | None = None, ) -> None: """Run benchmark and post-processing for a single workload. @@ -416,6 +429,7 @@ def _run_workload_benchmark( accelerator_chip=gpu_type.upper(), run_uuid=run_uuid, trtllm_config=trtllm_config, + mlflow_destination=mlflow_destination, ) if not run_benchmark: @@ -473,6 +487,7 @@ def _create_test_labels( accelerator_chip: str = "", run_uuid: str = "", trtllm_config: dict | 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()] @@ -500,7 +515,8 @@ def _create_test_labels( "runtime_args": runtime_args, "run_uuid": run_uuid, } - write_test_labels(env.ARTIFACT_DIR, labels) + + write_test_labels(env.ARTIFACT_DIR, labels, mlflow_destination=mlflow_destination) logger.info("Created test labels: %s", labels) @@ -593,26 +609,63 @@ 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 + + +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 + + 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", "") - 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, + 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", ""), ) 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/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 7557c8f79..0ae902860 100644 --- a/projects/rhaiis/postprocess/plugin.py +++ b/projects/rhaiis/postprocess/plugin.py @@ -18,6 +18,7 @@ logger = logging.getLogger(__name__) + # CSV columns whose KPI values are in seconds but the dashboard expects milliseconds. # GuideLLM parser converts `*_ms` metrics to seconds; the old CSV pipeline kept them in ms. _SECONDS_TO_MS_COLUMNS = frozenset( @@ -188,6 +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"] = 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..3f575ae9f 100644 --- a/projects/rhaiis/postprocess/regression.py +++ b/projects/rhaiis/postprocess/regression.py @@ -331,6 +331,18 @@ 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.""" + from projects.caliper.orchestration.export import build_mlflow_run_url_from_config + + try: + return build_mlflow_run_url_from_config() + except Exception: + logger.warning("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 +504,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 +517,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}" ) @@ -514,6 +530,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: + 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 "" + + 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, @@ -573,6 +673,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 +686,7 @@ def send_failure_notification( f"{version_line}" f"{cluster_line}" f"{profiles_line}" + f"{mlflow_line}" f"*Error:*\n```{error_text}```" )