Merge pull request #158 from ashtarkb/mlflow-metrics

[caliper] solidify the metrics to mlflow

Author: openshift-merge-bot[bot]

projects/caliper/cli/commands.py

@@ -695,6 +695,76 @@ def kpi_csv_export(
         sys.exit(3)
 
 
+@click.command("kpis-to-mlflow")
+@click.option(
+    "--input",
+    "input_file",
+    type=click.Path(path_type=Path),
+    required=True,
+    help="Input KPI JSON file (schema v2 hierarchical format)",
+)
+@click.option(
+    "--artifacts-dir",
+    "artifacts_dir",
+    type=click.Path(path_type=Path, exists=True),
+    required=True,
+    help="Root of the artifact tree containing __test_labels__.yaml markers",
+)
+@click.option(
+    "--status-file", type=click.Path(path_type=Path), help="YAML file to write operation status"
+)
+def kpis_to_mlflow_cmd(
+    input_file: Path,
+    artifacts_dir: Path,
+    status_file: Path | None,
+) -> None:
+    """Convert kpis.json into per-run metrics.json + parameters.json for MLflow."""
+    from projects.caliper.engine.kpi.kpis_to_mlflow import generate_metrics_from_kpis
+
+    status_data: dict = {"success": False}
+
+    try:
+        result = generate_metrics_from_kpis(input_file, artifacts_dir)
+        status = result.get("status", "unknown")
+        if status == "success":
+            status_data = {
+                "success": True,
+                "tests_processed": result.get("tests_processed", 0),
+                "total_tests": result.get("total_tests", 0),
+            }
+            click.echo(
+                f"Generated metrics.json for {result.get('tests_processed', 0)}/"
+                f"{result.get('total_tests', 0)} test(s)"
+            )
+        elif status == "skipped":
+            status_data = {"success": True, "skipped": True, "reason": result.get("reason", "")}
+            click.echo(f"Skipped: {result.get('reason', '')}")
+        else:
+            status_data = {"success": False, "error": result.get("error", "unknown error")}
+            click.echo(f"kpis-to-mlflow failed: {result.get('error', 'unknown')}", err=True)
+    except Exception as e:  # noqa: BLE001
+        import traceback
+
+        full_traceback = traceback.format_exc()
+        status_data = {"success": False, "error": str(e), "traceback": full_traceback}
+        click.echo(f"kpis-to-mlflow failed: {e}", err=True)
+        click.echo(f"Full traceback:\n{full_traceback}", err=True)
+
+        if not status_file:
+            sys.exit(3)
+    finally:
+        if status_file:
+            try:
+                with open(status_file, "w", encoding="utf-8") as f:
+                    yaml.dump(status_data, f, default_flow_style=False)
+            except Exception as status_err:
+                click.echo(f"Failed to write status file {status_file}: {status_err}", err=True)
+                sys.exit(4)
+
+    if not status_data.get("success", False):
+        sys.exit(3)
+
+
 @click.command("import")
 @click.option("--snapshot", type=click.Path(path_type=Path), required=True)
 @click.pass_context
@@ -1063,7 +1133,6 @@ def artifacts_export(
     status_yaml_path: Path | None,
     upload_workers: int,
 ) -> None:
-
     # Load config from file if provided, CLI args override
     config = {}
     if mlflow_config_path:

projects/caliper/cli/main.py

@@ -18,6 +18,7 @@
     kpi_generate,
     kpi_import,
     kpi_s3_import,
+    kpis_to_mlflow_cmd,
     list_reports_cmd,
     parse_cmd,
     visualize_cmd,
@@ -212,6 +213,7 @@ def run_cli() -> None:
 kpi_group.add_command(kpi_import)
 kpi_group.add_command(analyse_kpis_cmd)
 kpi_group.add_command(kpi_s3_import)
+kpi_group.add_command(kpis_to_mlflow_cmd)
 
 # Register s3-export command under kpi group
 kpi_group.add_command(s3_export_cmd)

projects/caliper/engine/file_export/mlflow_backend.py

@@ -461,6 +461,48 @@ def _load_json_file(path: Path) -> dict[str, Any]:
         return {}
 
 
+def _log_2d_metrics(metrics_2d: dict[str, Any]) -> None:
+    """Log 2D metrics as stepped MLflow metrics.
+
+    Each key maps to a list of ``{"x": ..., "y": ...}`` dicts (already
+    sorted by x in ``metrics_from_kpis``).  Each data point is logged
+    with ``step=int(x)`` so MLflow renders the curve.
+
+    Raises:
+        ValueError: If a data point has a non-integer x value (MLflow
+            steps must be integers) or if x/y are not numeric.
+    """
+    import mlflow
+
+    for metric_name, data_points in metrics_2d.items():
+        if not isinstance(data_points, list):
+            continue
+        for i, pt in enumerate(data_points):
+            if not isinstance(pt, dict):
+                raise ValueError(
+                    f"2D metric {metric_name!r}: data_points[{i}] is {type(pt).__name__}, "
+                    f"expected dict with 'x' and 'y' keys"
+                )
+            x = pt.get("x")
+            y = pt.get("y")
+            if (
+                not isinstance(x, int | float)
+                or isinstance(x, bool)
+                or not isinstance(y, int | float)
+                or isinstance(y, bool)
+            ):
+                raise ValueError(
+                    f"2D metric {metric_name!r}: data_points[{i}] has non-numeric "
+                    f"x={x!r} or y={y!r}"
+                )
+            if x != int(x):
+                raise ValueError(
+                    f"2D metric {metric_name!r}: data_points[{i}] has non-integer "
+                    f"step x={x!r} (MLflow steps must be integers)"
+                )
+            mlflow.log_metric(str(metric_name), float(y), step=int(x))
+
+
 def _log_metrics_and_params_from_tree(artifact_root: Path) -> None:
     """Find metrics.json/parameters.json under __test_labels__.yaml-marked dirs and log them."""
     import mlflow
@@ -473,7 +515,9 @@ def _log_metrics_and_params_from_tree(artifact_root: Path) -> None:
         mf = run_dir / "metrics.json"
         if mf.is_file():
             for k, v in _load_json_file(mf).items():
-                if isinstance(v, int | float) and not isinstance(v, bool):
+                if isinstance(v, list):
+                    _log_2d_metrics({k: v})
+                elif isinstance(v, int | float) and not isinstance(v, bool):
                     mlflow.log_metric(str(k), float(v))
 
         pf = run_dir / "parameters.json"
@@ -593,7 +637,9 @@ def _run(uri: str | None) -> tuple[str, dict[str, Any] | None]:
                     mf = run_dir / metrics_file
                     if mf.is_file():
                         for k, v in _load_json_file(mf).items():
-                            if isinstance(v, int | float) and not isinstance(v, bool):
+                            if isinstance(v, list):
+                                _log_2d_metrics({k: v})
+                            elif isinstance(v, int | float) and not isinstance(v, bool):
                                 mlflow.log_metric(str(k), float(v))
 
                     pf = run_dir / parameters_file

projects/caliper/engine/kpi/kpis_to_mlflow.py

@@ -0,0 +1,168 @@
+"""Generic kpis.json -> metrics.json + parameters.json conversion.
+
+Reads a hierarchical kpis.json (schema v2) and writes per-test-run
+metrics.json and parameters.json files into the matching artifact tree
+directories. The MLflow export backend picks these up automatically via
+``_log_metrics_and_params_from_tree``.
+
+This replaces project-specific metrics.json generation (e.g. in
+mcp_gateway parsers) with a single generic caliper mechanism that works
+for every project producing a kpis.json.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+METRICS_FILE = "metrics.json"
+PARAMETERS_FILE = "parameters.json"
+TEST_LABELS_MARKER = "__test_labels__.yaml"
+
+
+def _build_run_dir_index(artifact_tree: Path) -> dict[str, Path]:
+    """Map run directory names to their paths using __test_labels__.yaml markers."""
+    index: dict[str, Path] = {}
+    for marker in sorted(artifact_tree.rglob(TEST_LABELS_MARKER)):
+        if marker.is_file():
+            run_dir = marker.parent
+            try:
+                rel = run_dir.relative_to(artifact_tree)
+            except ValueError:
+                rel = Path(run_dir.name)
+            index[str(rel)] = run_dir
+            index[run_dir.name] = run_dir
+    return index
+
+
+def _is_scalar(value: Any) -> bool:
+    """Check if a KPI value is a scalar number (not 2D data)."""
+    return isinstance(value, int | float) and not isinstance(value, bool)
+
+
+def _extract_2d_points(value: Any) -> list[dict[str, float]] | None:
+    """Extract sorted (x, y) data points from a 2D KPI value.
+
+    Returns a list of ``{"x": ..., "y": ...}`` dicts sorted by x,
+    or ``None`` if the value is not a valid 2D structure.
+    """
+    if not isinstance(value, dict):
+        return None
+    data_points = value.get("data_points")
+    if not isinstance(data_points, list) or not data_points:
+        return None
+    points = []
+    for pt in data_points:
+        if isinstance(pt, dict) and _is_scalar(pt.get("x")) and _is_scalar(pt.get("y")):
+            points.append({"x": float(pt["x"]), "y": float(pt["y"])})
+    if not points:
+        return None
+    points.sort(key=lambda p: p["x"])
+    return points
+
+
+def _write_json(path: Path, data: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, sort_keys=True)
+        f.write("\n")
+
+
+def generate_metrics_from_kpis(
+    kpis_json_path: Path,
+    artifact_tree: Path,
+) -> dict[str, Any]:
+    """Convert kpis.json into per-run metrics.json and parameters.json files.
+
+    For each test entry in kpis.json, finds the matching directory under
+    ``artifact_tree`` (via ``__test_labels__.yaml`` markers) and writes:
+
+    - ``metrics.json``: ``{kpi_id: value}`` for all scalar KPIs
+    - ``parameters.json``: test-level labels as string key-value pairs
+
+    Args:
+        kpis_json_path: Path to the kpis.json file (schema v2).
+        artifact_tree: Root of the caliper artifact tree containing
+            test run directories with ``__test_labels__.yaml`` markers.
+
+    Returns:
+        Status dict with counts and any warnings.
+    """
+    if not kpis_json_path.is_file():
+        raise FileNotFoundError(f"kpis.json not found: {kpis_json_path}")
+
+    with kpis_json_path.open(encoding="utf-8") as f:
+        data = json.load(f)
+
+    if not isinstance(data, dict) or data.get("schema_version") != "2":
+        return {"status": "skipped", "reason": "Not a schema v2 kpis.json"}
+
+    tests = data.get("tests", [])
+    if not tests:
+        return {"status": "skipped", "reason": "No tests in kpis.json"}
+
+    run_dir_index = _build_run_dir_index(artifact_tree)
+    if not run_dir_index:
+        logger.warning("No test run directories found under %s", artifact_tree)
+        return {"status": "skipped", "reason": "No run directories with __test_labels__.yaml found"}
+
+    written = 0
+    warnings: list[str] = []
+
+    for test_entry in tests:
+        run_id = test_entry.get("run_id", "")
+        test_base_path = (
+            test_entry.get("metadata", {}).get("source", {}).get("test_base_path", run_id)
+        )
+
+        run_dir = run_dir_index.get(test_base_path) or run_dir_index.get(run_id)
+        if run_dir is None:
+            warnings.append(f"No matching directory for run_id={run_id!r}")
+            continue
+
+        kpis = test_entry.get("kpis", [])
+        metrics: dict[str, Any] = {}
+        for kpi in kpis:
+            kpi_id = kpi.get("id", "")
+            if not kpi_id:
+                continue
+            value = kpi.get("value")
+            is_2d = kpi.get("is_2d", False)
+            if is_2d:
+                points = _extract_2d_points(value)
+                if points:
+                    metrics[kpi_id] = points
+            elif _is_scalar(value):
+                metrics[kpi_id] = value
+
+        if metrics:
+            _write_json(run_dir / METRICS_FILE, metrics)
+
+        labels = test_entry.get("labels", {})
+        if labels:
+            params = {str(k): ("" if v is None else str(v)) for k, v in labels.items()}
+            _write_json(run_dir / PARAMETERS_FILE, params)
+
+        written += 1
+
+    result: dict[str, Any] = {
+        "status": "success",
+        "tests_processed": written,
+        "total_tests": len(tests),
+    }
+    if warnings:
+        result["warnings"] = warnings
+        for w in warnings:
+            logger.warning("kpis-to-metrics: %s", w)
+
+    logger.info(
+        "Generated metrics.json for %d/%d test(s) from %s",
+        written,
+        len(tests),
+        kpis_json_path.name,
+    )
+    return result

projects/caliper/orchestration/cli_builder.py

@@ -299,6 +299,30 @@ def build_ai_eval_export_command(
     return cmd
 
 
+def build_kpis_to_mlflow_command(
+    tree_root: Path,
+    status_file: Path,
+    input_file: Path,
+) -> list[str]:
+    """Build CLI command for caliper kpi kpis-to-mlflow.
+
+    Args:
+        tree_root: Root of the artifact tree with __test_labels__.yaml markers
+        status_file: Where to write status YAML
+        input_file: Input kpis.json file (schema v2)
+
+    Returns:
+        List of command arguments for subprocess
+    """
+    cmd = _CALIPER_BASE_CMD + ["kpi", "kpis-to-mlflow"]
+
+    cmd.extend(["--input", str(input_file)])
+    cmd.extend(["--artifacts-dir", str(tree_root)])
+    cmd.extend(["--status-file", str(status_file)])
+
+    return cmd
+
+
 def build_s3_import_command(
     config: CaliperOrchestrationPostprocessConfig,
     status_file: Path,