diff --git a/.github/workflows/mobile-price-classification-components.yaml b/.github/workflows/mobile-price-classification-components.yaml new file mode 100644 index 0000000..6c17a48 --- /dev/null +++ b/.github/workflows/mobile-price-classification-components.yaml @@ -0,0 +1,53 @@ +name: Build Mobile Price Classification Components Image + +on: + workflow_dispatch: + inputs: + tag: + description: "Image tag (e.g. v2, v3)" + required: true + default: "v2" + +env: + DOCKERFILE_PATH: "pipelines/lightweight-python-package/Dockerfile" + BUILD_CONTEXT_DIR: "pipelines/lightweight-python-package/" + IMAGE_NAME: "mobile-price-classification" + IMAGE_PUSH_PATH: "${{ secrets.GCP_ARTIFACT_REGISTRY_PATH }}" + IMAGE_PUSH_SECRET: "${{ secrets.GCP_SA_KEY }}" + +jobs: + build-image: + runs-on: ubuntu-latest + steps: + - name: Delete huge unnecessary tools folder + run: rm -rf /opt/hostedtoolcache + + - name: Checkout code + uses: actions/checkout@v7 + + - name: Auth GCloud CLI + uses: google-github-actions/auth@v3 + with: + credentials_json: ${{ env.IMAGE_PUSH_SECRET }} + + - name: Docker Login to Google Artifact Registry + uses: docker/login-action@v4 + with: + registry: europe-west3-docker.pkg.dev + username: _json_key + password: ${{ env.IMAGE_PUSH_SECRET }} + + - name: Build and push image + run: | + TAG=${{ github.event.inputs.tag }} + VERSIONED=${{ env.IMAGE_PUSH_PATH }}/${{ env.IMAGE_NAME }}:${TAG} + LATEST=${{ env.IMAGE_PUSH_PATH }}/${{ env.IMAGE_NAME }}:latest + COMMIT=${{ env.IMAGE_PUSH_PATH }}/${{ env.IMAGE_NAME }}:commit-${{ github.sha }} + BUILD_CONTEXT="${GITHUB_WORKSPACE}/${{ env.BUILD_CONTEXT_DIR }}" + docker build -t $VERSIONED -f ${{ env.DOCKERFILE_PATH }} "${BUILD_CONTEXT}" + docker tag $VERSIONED $LATEST + docker tag $VERSIONED $COMMIT + docker push $VERSIONED + docker push $LATEST + docker push $COMMIT + echo "Pushed: $VERSIONED" diff --git a/README.md b/README.md index 05c3307..4e315ef 100644 --- a/README.md +++ b/README.md @@ -7,15 +7,21 @@ For full platform documentation, see [docs.prokube.ai](https://docs.prokube.ai/) ```py . ├── .github # workflows to build images +├── ci # CI orchestrator (run_all.py) and contributor conventions ├── hparam-tuning # hyperparameter tuning examples (Katib) ├── images # custom container images used by examples ├── mlflow # MLflow experiment tracking examples ├── notebooks # Jupyter notebook examples (Dask, MNIST VAE, etc.) ├── pipelines # Kubeflow Pipelines examples ├── rstudio # RStudio examples +├── scripts # shared utilities (credential setup, API key management) ├── serving # model serving examples (KServe, vLLM, shadow deployments) ``` +Many serving examples include an `apply.py` (deploy + smoke-test) and a `cleanup.py` +(teardown of Kubernetes resources) alongside the notebook. These are used by the CI +orchestrator and can also be run manually. + ## Note about storage Storage on Kubernetes is a complex topic and a deep dive is outside the scope of this repository. There are two types of storage you might encounter here — block and object storage. @@ -43,6 +49,12 @@ Some examples require a minimum prokube platform version. If an example is not l |---|---|---| | `serving/minimal-s3-model` | v1.7.0 | Requires s3creds secret with KServe support | +## CI + +Examples are tested end-to-end by `ci/run_all.py`, which runs inside a Kubeflow +notebook pod. See [`ci/README.md`](ci/README.md) for how the orchestrator works and +how to add a new example to CI. + ## Contributing All code contributions should go via pull requests. Make sure your code is clearly documented and that it adheres to established standards (e.g. PEP). diff --git a/ci/README.md b/ci/README.md new file mode 100644 index 0000000..73c697d --- /dev/null +++ b/ci/README.md @@ -0,0 +1,156 @@ +# CI — contributor guide + +> This document is written primarily for AI agents making changes to this +> repository. It describes the conventions that must be followed for a new +> example to be picked up by `ci/run_all.py` correctly. + +--- + +## Adding a new example + +Register it in the `_EXAMPLES` list in `run_all.py`. Everything else — +phase scheduling, opt-in gating, cleanup, dry-run output, MLflow-credential +skipping — is derived from that entry automatically. + +```python +Example( + name="serving/my-new-example", # display name and result key + steps=[ + Step("script", "serving/my-new-example/apply.py"), + # or: Step("notebook", "serving/my-new-example/example.ipynb") + # chain multiple steps if needed (executed sequentially) + ], + phase=1, # see Phase rules below + cleanup="serving/my-new-example/cleanup.py", # omit if no K8s resources + opt_in="include_foo", # omit if always enabled + mlflow_dependent=False, # True = skip when MLflow creds absent +) +``` + +### Phase rules + +| Phase | When to use | +|-------|-------------| +| 1 | Self-contained: does not depend on anything else in CI | +| 2 | Submits a KFP pipeline and returns fast; actual run is polled in Phase 4 | +| 3 | Requires a model already registered by the Phase 2 mlflow-mobile-price pipeline | + +### Opt-in flag + +Add `opt_in="include_foo"` and a corresponding `--include-foo` argument in +the `__main__` block of `run_all.py`. Use opt-in when the example requires +cluster add-ons (KEDA, postgres-operator, GPU nodes) that are not guaranteed +to be present. + +--- + +## apply.py and cleanup.py + +### cleanup.py — always add when K8s resources are created + +Any example that creates Kubernetes resources (InferenceService, Deployment, +Service, CRD instance, …) must have a `cleanup.py` in its directory. +CI runs all cleanup scripts in parallel in a `finally` block so they execute +even on failure. + +A cleanup script must: +- Delete resources idempotently (`--ignore-not-found`) +- Never raise on failure (print a warning at most) +- Read the namespace from the pod filesystem: + +```python +with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f: + ns = f.read().strip() +``` + +**Do not** add a `cleanup.py` for examples that only create in-cluster +transient objects managed by KFP (pipeline runs, artifacts) — those are +cleaned up by KFP's own retention policy. + +### apply.py — only for serving examples driven by a script + +Add an `apply.py` only when the CI execution of a serving example is driven +by a Python script rather than a notebook. This is the case when: + +- The notebook's deploy/test section requires manual inputs that cannot be + automated, so a separate script owns the full deploy-wait-smoketest cycle. +- The example has no notebook at all. + +An `apply.py` must exit non-zero on failure (CI relies on the return code). +It must print a clean error message to stderr and suppress the full traceback: + +```python +if __name__ == "__main__": + try: + deploy() + except Exception as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) +``` + +If you add an `apply.py`, always add the matching `cleanup.py` as well. + +--- + +## scripts/ utilities + +`scripts/` contains two cluster utilities importable from notebooks and apply +scripts. Both work identically whether called from a notebook cell or from an +`apply.py`. + +### setup_mlflow_credentials.py + +One-time, interactive. Stores MLflow credentials in the +`mlflow-credentials` K8s secret. **Requires human input** (the MLflow +Personal Access Token cannot be obtained programmatically). Run it once from +a JupyterLab terminal: + +```bash +python examples/scripts/setup_mlflow_credentials.py +``` + +Do not call this from CI. CI validates the secret exists in the preflight +check and skips MLflow-dependent examples if it does not. + +### get_or_create_api_key.py + +Idempotent, no human input. Creates or retrieves the `examples-key` +AIGatewayKey CR. Call it from any notebook cell or script that needs an +inference API key: + +```python +import sys, subprocess, os +_root = subprocess.check_output(["git", "rev-parse", "--show-toplevel"], text=True).strip() +sys.path.insert(0, os.path.join(_root, "scripts")) +from get_or_create_api_key import get_or_create_api_key + +API_KEY = get_or_create_api_key() +``` + +--- + +## ci-skip cell tag + +Tag a notebook cell with `ci-skip` to have CI replace it with a no-op comment +before execution. Use this **only** for cells that genuinely cannot run +headlessly: + +- Cells that require manual input (hardcoded paths, paste-your-token + placeholders, `input()` calls). +- Interactive widget cells (`ipywidgets`, `ipykernel` display-only code). + +**Do not** use `ci-skip` to work around a fixable automation problem. If a +cell fails because it needs credentials or a namespace, fix it to load those +automatically (see the MLflow credential pattern above). `ci-skip` is a last +resort, not a convenience. + +To tag a cell in JupyterLab: select the cell → Property Inspector (right +panel) → add `ci-skip` under Cell Tags. + +In raw notebook JSON the tag appears as: + +```json +"metadata": { + "tags": ["ci-skip"] +} +``` diff --git a/ci/run_all.py b/ci/run_all.py new file mode 100644 index 0000000..63fdce6 --- /dev/null +++ b/ci/run_all.py @@ -0,0 +1,986 @@ +""" +CI orchestrator — runs all automatable examples and reports results. + +Execution model +--------------- +Phase 1 (parallel, self-contained): + All examples whose phase=1 in _EXAMPLES, including opt-in ones. + +Phase 2 (parallel, pipeline submissions — return fast): + All examples whose phase=2 in _EXAMPLES. + +Phase 3 (depends on Phase 2 mlflow-mobile-price notebook completing): + All examples whose phase=3 in _EXAMPLES. + +Phase 4 (KFP run polling — runs alongside Phase 3): + Poll all KFP run IDs extracted from Phase 2 output notebooks until + Succeeded or timeout. + +Phase 5 — cleanup (always, in finally block): + All cleanup.py scripts derived from _EXAMPLES, in parallel. + +Adding a new example +-------------------- +Add one entry to _EXAMPLES below. Everything else (cleanup, phase +scheduling, opt-in gating, dry-run output) is derived from the table. + +Usage from a Jupyter notebook +------------------------------ + import subprocess + subprocess.run(["python", "ci/run_all.py"], check=True) + +CLI usage +--------- + python ci/run_all.py [--timeout-notebook 1800] [--timeout-pipeline 3600] + [--include-keda] [--include-shadow] [--include-pytorch] + [--dry-run] + +Prerequisites +------------- + pip install papermill kfp + python scripts/setup_mlflow_credentials.py +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +import time +from concurrent.futures import Future, ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path +from typing import Callable, Literal + + +# ── Repo root ───────────────────────────────────────────────────────────────── +_REPO_ROOT = Path( + subprocess.check_output(["git", "rev-parse", "--show-toplevel"], text=True).strip() +) + + +# ── Prerequisite: papermill ─────────────────────────────────────────────────── + + +def _ensure_papermill() -> None: + """Install papermill if it is not already available.""" + try: + import papermill # noqa: F401 + except ImportError: + print("papermill not found — installing...") + subprocess.run( + [sys.executable, "-m", "pip", "install", "papermill"], + check=True, + ) + print("papermill installed.") + + +# ── KFP run ID pattern (UUID v4) ───────────────────────────────────────────── +_RUN_ID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}") + + +# ── Result dataclass ────────────────────────────────────────────────────────── + + +@dataclass +class Result: + name: str + status: str = "PENDING" # PASS | FAIL | SKIP + duration: float = 0.0 + error: str = "" + kfp_run_ids: list[str] = field(default_factory=list) + + +# ── Example registry ────────────────────────────────────────────────────────── + + +@dataclass +class Step: + """One execution unit within an Example — a notebook or a Python script.""" + + kind: Literal["notebook", "script"] + path: str # relative to repo root + extra_args: list[str] = field(default_factory=list) + # True → extract KFP run IDs from output (notebooks: broad UUID regex; + # scripts: KFP_RUN_ID= lines). Set False for ISVC examples + # whose output contains UUIDs that are not KFP run IDs. + extract_run_ids: bool = True + + +@dataclass +class Example: + """One CI example. Add a new entry to _EXAMPLES to include it in CI.""" + + name: str + steps: list[Step] + phase: int # 1 = independent 2 = pipeline submit 3 = needs mlflow model + cleanup: str | None = None # relative path to cleanup.py, or None + opt_in: str | None = ( + None # argparse dest that gates this example, e.g. "include_keda" + ) + mlflow_dependent: bool = ( + False # skip automatically when MLflow credentials are unavailable + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# Registration table — add new examples here. +# ───────────────────────────────────────────────────────────────────────────── +_EXAMPLES: list[Example] = [ + # ── Phase 1: independent, self-contained ───────────────────────────────── + Example( + name="notebooks/dask", + steps=[Step("notebook", "notebooks/dask/dask_example.ipynb")], + phase=1, + cleanup="notebooks/dask/cleanup.py", + ), + Example( + name="notebooks/mobile-price-classification", + steps=[ + Step( + "notebook", + "notebooks/mobile-price-classification/mobile-price-classifications.ipynb", + ) + ], + phase=1, + ), + Example( + name="mlflow/mlflow-quickstart", + steps=[Step("notebook", "mlflow/mlflow-quickstart-example.ipynb")], + phase=1, + mlflow_dependent=True, + ), + Example( + name="mlflow/mlflow-image-example", + steps=[Step("notebook", "mlflow/mlflow-image-example.ipynb")], + phase=1, + mlflow_dependent=True, + ), + Example( + name="mlflow/mlflow-kfp-example", + steps=[Step("notebook", "mlflow/mlflow-kfp-example.ipynb")], + phase=1, + mlflow_dependent=True, + ), + Example( + name="serving/minimal-s3-model", + steps=[Step("notebook", "serving/minimal-s3-model/minimal-s3-model.ipynb")], + phase=1, + cleanup="serving/minimal-s3-model/cleanup.py", + ), + Example( + name="serving/hf-vllm-completion", + steps=[ + Step("script", "serving/hf-vllm-completion/apply.py", extract_run_ids=False) + ], + phase=1, + cleanup="serving/hf-vllm-completion/cleanup.py", + ), + Example( + name="serving/kserve-keda-autoscaling", + steps=[ + Step( + "script", + "serving/kserve-keda-autoscaling/apply.py", + extract_run_ids=False, + ) + ], + phase=1, + opt_in="include_keda", + cleanup="serving/kserve-keda-autoscaling/cleanup.py", + ), + Example( + name="serving/minimal-example-shadow-deployment", + steps=[ + Step( + "script", + "serving/minimal-example-shadow-deployment/apply.py", + extract_run_ids=False, + ) + ], + phase=1, + opt_in="include_shadow", + cleanup="serving/minimal-example-shadow-deployment/cleanup.py", + ), + Example( + name="notebooks/mnist-vae", + steps=[ + Step( + "script", + "notebooks/mnist-vae/run_training.py", + extra_args=["--max_epochs", "3"], + extract_run_ids=False, + ), + Step( + "notebook", + "notebooks/mnist-vae/visualizations.ipynb", + extract_run_ids=False, + ), + ], + phase=1, + opt_in="include_pytorch", + ), + # ── Phase 2: pipeline submissions (return fast; KFP runs polled in Phase 4) + Example( + name="mlflow/mobile-price-classification", + steps=[ + Step( + "notebook", + "mlflow/mobile-price-classification/mlflow-mobile-price-classification.ipynb", + ) + ], + phase=2, + mlflow_dependent=True, + ), + Example( + name="pipelines/lightweight-components", + steps=[ + Step( + "notebook", + "pipelines/lightweight-components/mobile-price-classifications.ipynb", + ) + ], + phase=2, + ), + Example( + name="pipelines/lightweight-python-package", + steps=[ + Step("script", "pipelines/lightweight-python-package/submit-cluster.py") + ], + phase=2, + ), + Example( + name="pipelines/minimal-container-components", + steps=[ + Step("script", "pipelines/minimal-container-components/submit-cluster.py") + ], + phase=2, + ), + # ── Phase 3: MLflow KServe ISVCs (need model registered by Phase 2) ────── + Example( + name="serving/mlflow-kserve-minimal", + steps=[ + Step( + "script", + "serving/mlflow-kserve-minimal/apply.py", + extract_run_ids=False, + ) + ], + phase=3, + mlflow_dependent=True, + cleanup="serving/mlflow-kserve-minimal/cleanup.py", + ), + Example( + name="serving/mlflow-kserve-inference-protocols", + steps=[ + Step( + "script", + "serving/mlflow-kserve-inference-protocols/apply.py", + extract_run_ids=False, + ) + ], + phase=3, + mlflow_dependent=True, + cleanup="serving/mlflow-kserve-inference-protocols/cleanup.py", + ), +] + +# Cleanup scripts for resources not covered by an Example entry above. +_EXTRA_CLEANUP_PATHS: list[str] = [ + "hparam-tuning/minimal-mnist/cleanup.py", +] + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _format_papermill_error(exc: Exception) -> str: + """Extract a human-readable summary from a PapermillExecutionError.""" + try: + from papermill.exceptions import PapermillExecutionError + + if not isinstance(exc, PapermillExecutionError): + return str(exc) + except ImportError: + return str(exc) + + lines = [ + f"Cell {exc.exec_count} raised {exc.ename}: {exc.evalue}", + ] + if exc.source: + src_lines = exc.source.strip().splitlines()[:5] + lines.append(" Cell source:") + for src_line in src_lines: + lines.append(f" {src_line}") + if len(exc.source.strip().splitlines()) > 5: + lines.append(" ...") + if exc.traceback: + tb_lines = [l for l in exc.traceback if l.strip()] + if tb_lines: + lines.append(f" Traceback (last): {tb_lines[-1].strip()}") + return "\n".join(lines) + + +def _strip_ci_skip_cells(nb_path: Path, output_dir: Path) -> Path: + """Return a copy of the notebook with 'ci-skip' tagged cells replaced by a comment.""" + with open(nb_path) as fh: + nb = json.load(fh) + skipped = 0 + for cell in nb["cells"]: + if "ci-skip" in cell.get("metadata", {}).get("tags", []): + cell["source"] = ["# [CI] cell skipped (ci-skip tag)\n"] + skipped += 1 + if skipped == 0: + return nb_path + stripped = output_dir / f"_stripped_{nb_path.name}" + stripped.parent.mkdir(parents=True, exist_ok=True) + with open(stripped, "w") as fh: + json.dump(nb, fh, indent=1) + print(f" ({skipped} ci-skip cell(s) stripped from {nb_path.name})") + return stripped + + +def _run_notebook(nb_path: Path, output_dir: Path, timeout: int) -> Path: + """Execute a notebook with papermill. Returns the output notebook path.""" + import papermill as pm + from papermill.exceptions import PapermillExecutionError + + output_dir.mkdir(parents=True, exist_ok=True) + nb_to_run = _strip_ci_skip_cells(nb_path, output_dir) + output_path = output_dir / nb_path.name + try: + pm.execute_notebook( + str(nb_to_run), + str(output_path), + kernel_name="python3", + execution_timeout=timeout, + cwd=str(nb_path.parent), + progress_bar=False, + ) + except PapermillExecutionError as exc: + raise RuntimeError(_format_papermill_error(exc)) from exc + return output_path + + +def _run_script( + script_path: Path, + timeout: int, + extra_args: list[str] | None = None, +) -> tuple[str, str]: + """Run a plain Python script. Returns (stdout, stderr).""" + cmd = [sys.executable, str(script_path)] + (extra_args or []) + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + cwd=str(script_path.parent), + ) + if result.returncode != 0: + detail = (result.stderr or result.stdout or "(no output)").strip() + raise RuntimeError(detail) + return result.stdout, result.stderr + + +def _extract_run_ids_from_notebook(output_nb: Path) -> list[str]: + """Parse a papermill output notebook for KFP run IDs.""" + with open(output_nb) as fh: + nb = json.load(fh) + ids: list[str] = [] + for cell in nb.get("cells", []): + for output in cell.get("outputs", []): + text = "".join( + output.get("text", []) + output.get("data", {}).get("text/plain", []) + ) + ids.extend(_RUN_ID_RE.findall(text)) + return list(dict.fromkeys(ids)) + + +def _extract_run_ids_from_stdout(stdout: str) -> list[str]: + """Parse stdout from a script for KFP_RUN_ID= lines.""" + ids = [] + for line in stdout.splitlines(): + if line.startswith("KFP_RUN_ID="): + ids.append(line.split("=", 1)[1].strip()) + return ids + + +_KFP_TERMINAL_STATES = {"SUCCEEDED", "FAILED", "ERROR", "CANCELED", "SKIPPED"} + + +def _get_failed_task_logs(run: object, namespace: str) -> str: + """Best-effort: tail logs from the pod(s) of the first failed KFP task.""" + try: + task_details = ( + getattr(getattr(run, "run_details", None), "task_details", None) or [] + ) + for task in task_details: + if "FAIL" not in str(getattr(task, "state", "")).upper(): + continue + task_name = getattr(task, "display_name", "unknown-task") + for ct in getattr(task, "child_tasks", None) or []: + pod = ( + ct.get("pod_name") + if isinstance(ct, dict) + else getattr(ct, "pod_name", None) + ) + if not pod: + continue + for container in ("main", "user-main"): + result = subprocess.run( + [ + "kubectl", + "logs", + pod, + "-n", + namespace, + "-c", + container, + "--tail=15", + ], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode == 0 and result.stdout.strip(): + header = f"[Task '{task_name}' / pod {pod[:30]}...]" + tail = "\n".join( + f" {l}" for l in result.stdout.strip().splitlines()[-10:] + ) + return f"{header}\n{tail}" + except Exception: # noqa: BLE001 + pass + return "" + + +def _poll_kfp_run(run_id: str, timeout: int, interval: int = 30) -> tuple[str, str]: + """Poll a KFP run until terminal state. Returns (status, error_detail).""" + try: + from kfp.client import Client + except ImportError: + return "SKIP (kfp not installed)", "" + client = Client() + try: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + namespace = fh.read().strip() + except OSError: + namespace = "" + deadline = time.time() + timeout + while time.time() < deadline: + run = client.get_run(run_id) + state = str( + getattr(run, "state", None) + or getattr(getattr(run, "run", None), "status", None) + or "UNKNOWN" + ).upper() + if state in _KFP_TERMINAL_STATES: + error_detail = "" + if state not in ("SUCCEEDED", "SKIPPED"): + err = getattr(run, "error", None) + error_detail = (getattr(err, "message", "") or "") if err else "" + if not error_detail and namespace: + error_detail = _get_failed_task_logs(run, namespace) + return state, error_detail + time.sleep(interval) + return f"TIMEOUT after {timeout}s", "" + + +def _run_cleanup(cleanup_path: Path) -> None: + """Run a cleanup.py; log but don't raise on failure.""" + try: + subprocess.run( + [sys.executable, str(cleanup_path)], + capture_output=True, + text=True, + cwd=str(cleanup_path.parent), + timeout=120, + ) + except Exception as exc: # noqa: BLE001 + print(f" WARNING: cleanup {cleanup_path.name} failed: {exc}", file=sys.stderr) + + +# ── Orchestration context ───────────────────────────────────────────────────── + + +@dataclass +class _Context: + """Shared mutable state threaded through every phase.""" + + executor: ThreadPoolExecutor + root: Path + output_dir: Path + timeout_notebook: int + timeout_pipeline: int + results: dict[str, Result] + poll_results: dict[str, str] + poll_errors: dict[str, str] + + +# ── Execution primitives ────────────────────────────────────────────────────── + + +def _make_work( + steps: list[Step], root: Path, output_dir: Path, timeout: int +) -> Callable[[Result], None]: + """Build a work(result) closure from a list of Steps.""" + + def work(result: Result) -> None: + for step in steps: + if step.kind == "notebook": + out = _run_notebook(root / step.path, output_dir, timeout) + if step.extract_run_ids: + result.kfp_run_ids.extend(_extract_run_ids_from_notebook(out)) + elif step.kind == "script": + stdout, _ = _run_script( + root / step.path, + timeout, + extra_args=step.extra_args or None, + ) + if step.extract_run_ids: + result.kfp_run_ids.extend(_extract_run_ids_from_stdout(stdout)) + + return work + + +def _timed_run( + executor: ThreadPoolExecutor, + results: dict[str, Result], + name: str, + work: Callable[[Result], None], +) -> Future: + """Submit work(result) to the executor with shared timing + error handling.""" + result = Result(name=name) + results[name] = result + print(f" [START ] {name}") + + def _run() -> None: + t0 = time.time() + try: + work(result) + result.status = "PASS" + except Exception as exc: # noqa: BLE001 + result.status = "FAIL" + result.error = str(exc) + finally: + result.duration = time.time() - t0 + + return executor.submit(_run) + + +def _drain(ctx: _Context, futures: dict[str, Future]) -> None: + """Wait on a {name: Future} map, printing each result as it completes.""" + by_future = {v: k for k, v in futures.items()} + for f in as_completed(futures.values()): + _print_result(ctx.results[by_future[f]]) + + +def _skip(ctx: _Context, name: str, reason: str) -> None: + """Record a SKIP result and print a one-line notice.""" + print(f" [SKIP ] {name}") + ctx.results[name] = Result(name=name, status="SKIP", error=reason) + + +# ── Report ──────────────────────────────────────────────────────────────────── + + +def _print_result(r: Result) -> None: + dur = f"{r.duration:.0f}s" if r.duration else "-" + print(f" [{r.status}] {r.name} ({dur})") + if r.status == "FAIL" and r.error: + for line in r.error.splitlines(): + print(f" {line}") + + +def _print_report( + results: dict[str, Result], + poll_results: dict[str, str], + poll_errors: dict[str, str], + run_id_to_name: dict[str, str], +) -> None: + col = 50 + passed = failed = 0 + for r in results.values(): + if r.status == "FAIL": + failed += 1 + else: + passed += 1 + for status in poll_results.values(): + if status.upper() == "SUCCEEDED": + passed += 1 + else: + failed += 1 + + if failed: + print("\nFailed details:") + print("-" * 70) + for r in results.values(): + if r.status == "FAIL" and r.error: + print(f"\n{r.name}:") + for line in r.error.splitlines(): + print(f" {line}") + for run_id, status in poll_results.items(): + if status.upper() != "SUCCEEDED": + source = run_id_to_name.get(run_id, "unknown") + print(f"\nKFP run {run_id[:8]}... (from {source}): {status}") + err = poll_errors.get(run_id, "") + if err: + for line in err.splitlines(): + print(f" {line}") + + print("\n" + "=" * 70) + print(f"{'EXAMPLE':<{col}} {'STATUS':<10} {'DURATION'}") + print("-" * 70) + for r in results.values(): + dur = f"{r.duration:.0f}s" if r.duration else "-" + print(f"{r.name:<{col}} {r.status:<10} {dur}") + if poll_results: + print() + print(f"{'KFP RUN (source notebook)':<{col}} {'STATUS':<10}") + print("-" * 70) + for run_id, status in poll_results.items(): + source = run_id_to_name.get(run_id, "unknown") + label = f"{run_id[:8]}... ({source})" + print(f"{label:<{col}} {status:<10}") + print("=" * 70) + print(f"PASSED: {passed} FAILED: {failed} TOTAL: {passed + failed}") + print() + + +def _print_dry_run() -> None: + by_phase: dict[int, list[str]] = {} + for ex in _EXAMPLES: + label = ex.name + if ex.opt_in: + label += f" (--{ex.opt_in.replace('_', '-')})" + if ex.mlflow_dependent: + label += " (mlflow)" + by_phase.setdefault(ex.phase, []).append(label) + phase_names = { + 1: "independent", + 2: "pipeline submissions", + 3: "MLflow KServe ISVCs", + } + print("[dry-run] Would execute the following phases:") + for phase, labels in sorted(by_phase.items()): + print(f" Phase {phase} ({phase_names.get(phase, f'phase {phase}')}):") + for label in labels: + print(f" {label}") + print(" Phase 4: KFP run polling") + print(" Phase 5: cleanup") + + +# ── Pre-flight ──────────────────────────────────────────────────────────────── + + +def _check_mlflow_credentials() -> tuple[bool, str]: + """Return (ok, reason). Checks secret existence then validates with MLflow API.""" + import base64 as _b64 + import json as _json + import urllib.error + import urllib.request + + ns = _namespace() + r = subprocess.run( + ["kubectl", "get", "secret", "mlflow-credentials", "-n", ns, "-o", "json"], + capture_output=True, + text=True, + ) + if r.returncode != 0: + return False, ( + "mlflow-credentials secret not found — " + "run scripts/setup_mlflow_credentials.py first" + ) + try: + data = _json.loads(r.stdout)["data"] + uri = _b64.b64decode(data["MLFLOW_TRACKING_URI"]).decode().rstrip("/") + username = _b64.b64decode(data["MLFLOW_TRACKING_USERNAME"]).decode() + password = _b64.b64decode(data["MLFLOW_TRACKING_PASSWORD"]).decode() + except KeyError as exc: + return ( + False, + f"mlflow-credentials secret is missing key {exc} — re-run setup_mlflow_credentials.py", + ) + + creds = _b64.b64encode(f"{username}:{password}".encode()).decode() + req = urllib.request.Request( + f"{uri}/api/2.0/mlflow/experiments/search?max_results=1", + headers={"Authorization": f"Basic {creds}"}, + ) + try: + urllib.request.urlopen(req, timeout=8) + return True, "OK" + except urllib.error.HTTPError as exc: + if exc.code in (401, 403): + return False, ( + "MLflow credentials are invalid or the PAT has expired — " + "re-run scripts/setup_mlflow_credentials.py" + ) + return ( + False, + f"MLflow API returned HTTP {exc.code} — check MLFLOW_TRACKING_URI in the secret", + ) + except Exception as exc: # noqa: BLE001 + return False, f"Could not reach MLflow at {uri}: {exc}" + + +def _preflight(results: dict[str, Result]) -> bool: + """Install papermill and validate MLflow credentials. + + MLflow-dependent examples are recorded as SKIP in `results` on failure. + Returns True if MLflow credentials are valid. + """ + _ensure_papermill() + print("Pre-flight: checking MLflow credentials...") + mlflow_ok, reason = _check_mlflow_credentials() + if mlflow_ok: + print(" [OK] MLflow credentials valid") + else: + print(f" [SKIP] {reason}") + for ex in _EXAMPLES: + if ex.mlflow_dependent: + results[ex.name] = Result(name=ex.name, status="SKIP", error=reason) + return mlflow_ok + + +# ── Phases ──────────────────────────────────────────────────────────────────── + + +def _run_phase( + ctx: _Context, + phase: int, + opts: dict[str, bool], +) -> dict[str, Future]: + """Submit all examples for a given phase; return {name: Future}.""" + futures: dict[str, Future] = {} + for ex in _EXAMPLES: + if ex.phase != phase: + continue + if ex.name in ctx.results: # already SKIP from pre-flight + print(f" [SKIP ] {ex.name}") + continue + if ex.opt_in and not opts.get(ex.opt_in, False): + _skip( + ctx, ex.name, f"opt-in: pass --{ex.opt_in.replace('_', '-')} to enable" + ) + continue + work = _make_work(ex.steps, ctx.root, ctx.output_dir, ctx.timeout_notebook) + futures[ex.name] = _timed_run(ctx.executor, ctx.results, ex.name, work) + return futures + + +def _await_mobile_price(ctx: _Context, phase2_futures: dict[str, Future]) -> bool: + """Wait for mlflow-mobile-price and poll its KFP run inline. + + Phase 3 ISVCs need the registered model, so this must complete before + Phase 3 starts. Returns True if the model is registered and ready. + """ + name = "mlflow/mobile-price-classification" + if name not in phase2_futures: + return False + + phase2_futures[name].result() + _print_result(ctx.results[name]) + if ctx.results[name].status != "PASS": + return False + + run_ids = ctx.results[name].kfp_run_ids + if not run_ids: + return True + + print( + " Polling mlflow-mobile-price KFP pipeline (model must be registered before ISVCs)..." + ) + ok = True + for run_id in run_ids: + state, err = _poll_kfp_run(run_id, ctx.timeout_pipeline) + ctx.poll_results[run_id] = state + ctx.poll_errors[run_id] = err + print(f" [{state}] {run_id[:8]}...") + if state.upper() != "SUCCEEDED": + ok = False + return ok + + +def _phase4_poll(ctx: _Context) -> None: + all_run_ids = [ + rid + for r in ctx.results.values() + for rid in r.kfp_run_ids + if rid not in ctx.poll_results + ] + if not all_run_ids: + print("\nPhase 4: no KFP run IDs found, skipping poll.") + return + + print(f"\nPhase 4: polling {len(all_run_ids)} KFP run(s)...") + poll_futures = { + run_id: ctx.executor.submit(_poll_kfp_run, run_id, ctx.timeout_pipeline) + for run_id in all_run_ids + } + for run_id, f in poll_futures.items(): + try: + ctx.poll_results[run_id], ctx.poll_errors[run_id] = f.result() + except Exception as exc: # noqa: BLE001 + ctx.poll_results[run_id] = f"POLL_ERROR: {exc}" + ctx.poll_errors[run_id] = "" + print(f" [{ctx.poll_results[run_id]}] {run_id[:8]}...") + + +def _phase5_cleanup(cleanup_scripts: list[Path]) -> None: + print("\nPhase 5: running cleanup scripts...") + with ThreadPoolExecutor(max_workers=4) as ex: + futs = [ex.submit(_run_cleanup, p) for p in cleanup_scripts if p.exists()] + for f in as_completed(futs): + f.result() + + +# ── Main ────────────────────────────────────────────────────────────────────── + + +def run_all( + timeout_notebook: int = 1800, + timeout_pipeline: int = 3600, + include_keda: bool = False, + include_pytorch: bool = False, + include_shadow: bool = False, + dry_run: bool = False, +) -> dict[str, Result]: + root = _REPO_ROOT + results: dict[str, Result] = {} + + if dry_run: + _print_dry_run() + return results + + _preflight(results) + + opts = { + "include_keda": include_keda, + "include_pytorch": include_pytorch, + "include_shadow": include_shadow, + } + cleanup_scripts = [root / ex.cleanup for ex in _EXAMPLES if ex.cleanup] + [ + root / p for p in _EXTRA_CLEANUP_PATHS + ] + poll_results: dict[str, str] = {} + poll_errors: dict[str, str] = {} + + try: + with ThreadPoolExecutor(max_workers=8) as executor: + ctx = _Context( + executor=executor, + root=root, + output_dir=root / "ci" / "output", + timeout_notebook=timeout_notebook, + timeout_pipeline=timeout_pipeline, + results=results, + poll_results=poll_results, + poll_errors=poll_errors, + ) + + print("Phase 1: running independent examples in parallel...") + _drain(ctx, _run_phase(ctx, 1, opts)) + + print("\nPhase 2: submitting pipelines...") + phase2_futures = _run_phase(ctx, 2, opts) + mobile_price_ok = _await_mobile_price(ctx, phase2_futures) + + print("\nPhase 3: deploying MLflow KServe InferenceServices...") + if mobile_price_ok: + phase3_futures = _run_phase(ctx, 3, opts) + else: + skipped = "mlflow/mobile-price-classification" not in phase2_futures + reason = ( + "prerequisite mlflow/mobile-price-classification skipped (MLflow credentials)" + if skipped + else "prerequisite mlflow/mobile-price-classification notebook or KFP pipeline did not succeed" + ) + print(f" [SKIP] {reason}") + for ex in _EXAMPLES: + if ex.phase == 3: + ctx.results[ex.name] = Result( + name=ex.name, status="SKIP", error=reason + ) + phase3_futures = {} + + print("\nPhase 2 (remaining) + Phase 3 running concurrently...") + remaining = { + k: v + for k, v in phase2_futures.items() + if k != "mlflow/mobile-price-classification" + } + remaining.update(phase3_futures) + _drain(ctx, remaining) + + _phase4_poll(ctx) + + finally: + _phase5_cleanup(cleanup_scripts) + + run_id_to_name = { + run_id: r.name for r in results.values() for run_id in r.kfp_run_ids + } + _print_report(results, poll_results, poll_errors, run_id_to_name) + return results + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--timeout-notebook", + type=int, + default=1800, + help="Per-notebook execution timeout in seconds (default: 1800)", + ) + parser.add_argument( + "--timeout-pipeline", + type=int, + default=3600, + help="KFP run poll timeout in seconds (default: 3600)", + ) + parser.add_argument( + "--include-keda", + action="store_true", + help="Include KEDA autoscaling example (opt-in; requires KEDA in the cluster; runs on CPU)", + ) + parser.add_argument( + "--include-shadow", + action="store_true", + help=( + "Include minimal-example-shadow-deployment (opt-in; " + "requires CrunchyData postgres-operator; Istio VirtualService not tested)" + ), + ) + parser.add_argument( + "--include-pytorch", + action="store_true", + help="Include pytorch_lightning examples (opt-in; requires pytorch in the notebook image)", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Print plan without executing anything", + ) + args = parser.parse_args() + + results = run_all( + timeout_notebook=args.timeout_notebook, + timeout_pipeline=args.timeout_pipeline, + include_keda=args.include_keda, + include_shadow=args.include_shadow, + include_pytorch=args.include_pytorch, + dry_run=args.dry_run, + ) + + failed = sum(1 for r in results.values() if r.status == "FAIL") + sys.exit(1 if failed else 0) diff --git a/hparam-tuning/minimal-mnist/cleanup.py b/hparam-tuning/minimal-mnist/cleanup.py new file mode 100644 index 0000000..32d6475 --- /dev/null +++ b/hparam-tuning/minimal-mnist/cleanup.py @@ -0,0 +1,54 @@ +""" +Cleanup script for the hparam-tuning/minimal-mnist example. + +Deletes (idempotently): + - Katib Experiment ``minimal-mnist`` + - Any Trial pods created by the Experiment (Katib garbage-collects these + automatically, but this speeds things up if you need to clean up quickly) + +Usage +----- + python cleanup.py [--dry-run] +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + +_EXPERIMENT_NAME = "minimal-mnist" + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_delete(*args: str, dry_run: bool = False) -> None: + cmd = ["kubectl", "delete", *args, "--ignore-not-found"] + if dry_run: + print(f"[dry-run] {' '.join(cmd)}") + return + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f"WARNING: {result.stderr.strip()}", file=sys.stderr) + else: + print(result.stdout.strip() or f"deleted (or not found): {' '.join(args)}") + + +def cleanup(dry_run: bool = False) -> None: + ns = _namespace() + # Deleting the Experiment cascades to Trials via owner references + _kubectl_delete("experiment", _EXPERIMENT_NAME, "-n", ns, dry_run=dry_run) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--dry-run", action="store_true", help="Print commands without executing them" + ) + args = parser.parse_args() + cleanup(dry_run=args.dry_run) diff --git a/mlflow/mlflow-image-example.ipynb b/mlflow/mlflow-image-example.ipynb index 186cdf6..19a05ab 100644 --- a/mlflow/mlflow-image-example.ipynb +++ b/mlflow/mlflow-image-example.ipynb @@ -9,20 +9,16 @@ "\n", "This notebook demonstrates how to use MLflow in the prokube platform with Personal Access Token (PAT) authentication on a slightly more complex example then the Quick Start.\n", "\n", - "## Prerequisites\n", + "## MLflow Setup\n", "\n", - "- Generate your Personal Access Token:\n", - " - Open [MLflow UI](/mlflow) in your browser\n", - " - Navigate to the [Permissions](/mlflow/oidc/ui/#) page\n", - " - Click on \"Create access key\" button\n", - " - Copy the generated token and store it securely\n", - " - Note: You won't be able to see the token again!\n", + "Run `scripts/setup_mlflow_credentials.py` once to create the required secret:\n", + "```bash\n", + "# From a JupyterLab terminal:\n", + "cd && python examples/scripts/setup_mlflow_credentials.py\n", + "```\n", + "You will be prompted for your MLflow tracking URI, username (e-mail), and Personal Access Token.\n", "\n", - "- Configure the credentials below with your values\n", - "\n", - "## Authentication\n", - "\n", - "The prokube MLflow setup uses OIDC authentication with PAT support for programmatic access.\n" + "To generate a Personal Access Token, open the [MLflow UI](/mlflow) and navigate to [Permissions](/mlflow/oidc/ui/user) and click \"Create Access Token\"." ] }, { @@ -42,12 +38,27 @@ "metadata": {}, "outputs": [], "source": [ - "import os\n", + "import base64, json, os, subprocess\n", "\n", - "# Set your MLflow credentials\n", - "os.environ['MLFLOW_TRACKING_URI'] = 'https:///mlflow'\n", - "os.environ['MLFLOW_TRACKING_USERNAME'] = '' \n", - "os.environ['MLFLOW_TRACKING_PASSWORD'] = ''" + "# Load MLflow credentials from the mlflow-credentials Kubernetes secret.\n", + "# Run scripts/setup_mlflow_credentials.py once to create this secret.\n", + "with open(\"/var/run/secrets/kubernetes.io/serviceaccount/namespace\") as _f:\n", + " _ns = _f.read().strip()\n", + "\n", + "_result = subprocess.run(\n", + " [\"kubectl\", \"get\", \"secret\", \"mlflow-credentials\", \"-n\", _ns, \"-o\", \"json\"],\n", + " capture_output=True, text=True,\n", + ")\n", + "if _result.returncode != 0:\n", + " raise RuntimeError(\n", + " \"mlflow-credentials secret not found.\\n\"\n", + " \"Run scripts/setup_mlflow_credentials.py to create it.\"\n", + " )\n", + "_data = json.loads(_result.stdout)[\"data\"]\n", + "os.environ[\"MLFLOW_TRACKING_URI\"] = base64.b64decode(_data[\"MLFLOW_TRACKING_URI\"]).decode()\n", + "os.environ[\"MLFLOW_TRACKING_USERNAME\"] = base64.b64decode(_data[\"MLFLOW_TRACKING_USERNAME\"]).decode()\n", + "os.environ[\"MLFLOW_TRACKING_PASSWORD\"] = base64.b64decode(_data[\"MLFLOW_TRACKING_PASSWORD\"]).decode()\n", + "os.environ[\"MLFLOW_ENABLE_PROXY_MULTIPART_UPLOAD\"] = \"true\"\n" ] }, { diff --git a/mlflow/mlflow-kfp-example.ipynb b/mlflow/mlflow-kfp-example.ipynb index 7ef127f..4fc26bf 100644 --- a/mlflow/mlflow-kfp-example.ipynb +++ b/mlflow/mlflow-kfp-example.ipynb @@ -7,39 +7,24 @@ "source": [ "# Simple pipeline with MLflow model tracking\n", "\n", - "An experiment with iris dataset. In general, to use MLflow in a Kubeflow Pipeline, the necessary environment variables should be passed to the containers using the MLflow logic. This is implemented in the `add_env_vars_to_tasks` function. \n", + "An experiment with iris dataset. In general, to use MLflow in a Kubeflow Pipeline, the necessary environment variables should be passed to the containers using the MLflow logic. This is implemented in the `add_env_vars_to_tasks` function.\n", "\n", + "## MLflow Setup\n", "\n", - "## Prerequisites\n", - "\n", - "To use MLflow in Kubeflow Pipelines, you need to create a Kubernetes secret with your MLflow credentials. \n", - "\n", - "To generate \n", - "\n", - "- Generate your Personal Access Token (re-use your old one, if you have already generated it):\n", - " - Open [MLflow UI](/mlflow) in your browser\n", - " - Navigate to the [Permissions](/mlflow/oidc/ui/#) page\n", - " - Click on \"Create access key\" button\n", - " - Copy the generated token and store it securely\n", - " - Note: You won't be able to see the token again!\n", - "\n", - "Now change copy below command and change it to your values:\n", + "Run `scripts/setup_mlflow_credentials.py` once to create the required secret:\n", "```bash\n", - "kubectl create secret generic mlflow-credentials \\\n", - "--from-literal=MLFLOW_TRACKING_URI='https:///mlflow' \\\n", - "--from-literal=MLFLOW_TRACKING_USERNAME='your-email@example.com' \\\n", - "--from-literal=MLFLOW_TRACKING_PASSWORD='your-personal-access-token'\n", + "# From a JupyterLab terminal:\n", + "cd && python examples/scripts/setup_mlflow_credentials.py\n", "```\n", + "You will be prompted for your MLflow tracking URI, username (e-mail), and Personal Access Token.\n", "\n", - "You can then paste the above command in the terminal built into jupyter-lab and press enter to run it.\n", + "To generate a Personal Access Token, open the [MLflow UI](/mlflow) and navigate to [Permissions](/mlflow/oidc/ui/user) and click \"Create Access Token\".\n", "\n", "## How it works\n", "\n", - "In this pipeline, the add_mlflow_env_vars_to_tasks function injects the MLflow credentials from the Kubernetes secret into each pipeline component. This allows the components to authenticate with\n", - "MLflow and track experiments.\n", + "In this pipeline, the add_mlflow_env_vars_to_tasks function injects the MLflow credentials from the Kubernetes secret into each pipeline component. This allows the components to authenticate with MLflow and track experiments.\n", "\n", - "This notebook defines a simple pipeline for preprocessing data, training and logging model, and prediction on test data. It also shows one way to handle MLflow experiment info inside the pipeline - by\n", - "passing run information between components." + "This notebook defines a simple pipeline for preprocessing data, training and logging model, and prediction on test data. It also shows one way to handle MLflow experiment info inside the pipeline - by passing run information between components." ] }, { diff --git a/mlflow/mlflow-quickstart-example.ipynb b/mlflow/mlflow-quickstart-example.ipynb index f1927bc..7e3e289 100644 --- a/mlflow/mlflow-quickstart-example.ipynb +++ b/mlflow/mlflow-quickstart-example.ipynb @@ -27,20 +27,16 @@ "\n", "This notebook demonstrates how to use MLflow in the prokube platform with Personal Access Token (PAT) authentication.\n", "\n", - "## Prerequisites\n", + "## MLflow Setup\n", "\n", - "- Generate your Personal Access Token:\n", - " - Open [MLflow UI](/mlflow) in your browser\n", - " - Navigate to the [Permissions](/mlflow/oidc/ui/#) page\n", - " - Click on \"Create access key\" button\n", - " - Copy the generated token and store it securely\n", - " - Note: You won't be able to see the token again!\n", + "Run `scripts/setup_mlflow_credentials.py` once to create the required secret:\n", + "```bash\n", + "# From a JupyterLab terminal:\n", + "cd && python examples/scripts/setup_mlflow_credentials.py\n", + "```\n", + "You will be prompted for your MLflow tracking URI, username (e-mail), and Personal Access Token.\n", "\n", - "- Configure the credentials below with your values\n", - "\n", - "## Authentication\n", - "\n", - "The prokube MLflow setup uses OIDC authentication with PAT support for programmatic access.\n" + "To generate a Personal Access Token, open the [MLflow UI](/mlflow) and navigate to [Permissions](/mlflow/oidc/ui/user) and click \"Create Access Token\"." ] }, { @@ -52,61 +48,27 @@ }, "outputs": [], "source": [ - "import os\n", + "import base64, json, os, subprocess\n", "\n", - "# Set your MLflow credentials\n", - "os.environ['MLFLOW_TRACKING_URI'] = 'https:///mlflow'\n", - "os.environ['MLFLOW_TRACKING_USERNAME'] = '' \n", - "os.environ['MLFLOW_TRACKING_PASSWORD'] = ''\n", + "# Load MLflow credentials from the mlflow-credentials Kubernetes secret.\n", + "# Run scripts/setup_mlflow_credentials.py once to create this secret.\n", + "with open(\"/var/run/secrets/kubernetes.io/serviceaccount/namespace\") as _f:\n", + " _ns = _f.read().strip()\n", "\n", - "# Enable presigned URL uploads to avoid RAM buffering in MLflow server\n", - "os.environ['MLFLOW_ENABLE_PROXY_MULTIPART_UPLOAD'] = 'true'" - ] - }, - { - "cell_type": "code", - "execution_count": null, - "id": "02a920aa-df99-4902-b9ca-b534542882c7", - "metadata": {}, - "outputs": [], - "source": [ - "# Validate MLflow configuration\n", - "import os\n", - "\n", - "REQUIRED_ENV_VARS = [\"MLFLOW_TRACKING_URI\", \"MLFLOW_TRACKING_USERNAME\", \"MLFLOW_TRACKING_PASSWORD\"]\n", - "missing_vars = [var for var in REQUIRED_ENV_VARS if not os.environ.get(var)]\n", - "placeholder_vars = [\n", - " var for var in REQUIRED_ENV_VARS \n", - " if os.environ.get(var, \"\").startswith((\"https://<\", \"/mlflow' \\\n", - " --from-literal=MLFLOW_TRACKING_USERNAME='your-email@example.com' \\\n", - " --from-literal=MLFLOW_TRACKING_PASSWORD='your-personal-access-token'\n", + "# From a JupyterLab terminal:\n", + "cd && python examples/scripts/setup_mlflow_credentials.py\n", "```\n", + "You will be prompted for your MLflow tracking URI, username (e-mail), and Personal Access Token.\n", "\n", - "To generate a Personal Access Token, open the [MLflow UI](/mlflow) and navigate to [Permissions](/mlflow/oidc/ui/#)." + "To generate a Personal Access Token, open the [MLflow UI](/mlflow) and navigate to [Permissions](/mlflow/oidc/ui/user) and click \"Create Access Token\"." ] }, { @@ -171,7 +171,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"pyarrow\", \"s3fs\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def read_data(\n", " train_data_path: str,\n", @@ -206,7 +206,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def split_data(\n", " train_df: Input[Dataset],\n", @@ -254,7 +254,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def fit_scaler(\n", " train_x: Input[Dataset],\n", @@ -295,7 +295,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def tune_hyperparams(\n", " train_x: Input[Dataset],\n", @@ -365,7 +365,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\", \"mlflow==3.10.0\", \"s3fs\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def train_model(\n", " train_x: Input[Dataset],\n", @@ -445,7 +445,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\", \"mlflow==3.10.0\", \"s3fs\", \"matplotlib\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def evaluate_model(\n", " val_x: Input[Dataset],\n", @@ -530,7 +530,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\", \"plotly\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def test_model(\n", " test_x: Input[Dataset],\n", diff --git a/notebooks/dask/cleanup.py b/notebooks/dask/cleanup.py new file mode 100644 index 0000000..0269852 --- /dev/null +++ b/notebooks/dask/cleanup.py @@ -0,0 +1,53 @@ +""" +Cleanup script for the dask example. + +Deletes (idempotently): + - Any DaskCluster CRs left running in the current namespace + (dask-kubernetes names them after the KubeCluster object; the example + uses the default name ``dask-cluster``) + +Usage +----- + python cleanup.py [--dry-run] +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + +_CLUSTER_NAME = "dask-cluster" + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_delete(*args: str, dry_run: bool = False) -> None: + cmd = ["kubectl", "delete", *args, "--ignore-not-found"] + if dry_run: + print(f"[dry-run] {' '.join(cmd)}") + return + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f"WARNING: {result.stderr.strip()}", file=sys.stderr) + else: + print(result.stdout.strip() or f"deleted (or not found): {' '.join(args)}") + + +def cleanup(dry_run: bool = False) -> None: + ns = _namespace() + _kubectl_delete("daskcluster", _CLUSTER_NAME, "-n", ns, dry_run=dry_run) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--dry-run", action="store_true", help="Print commands without executing them" + ) + args = parser.parse_args() + cleanup(dry_run=args.dry_run) diff --git a/notebooks/dask/dask_example.ipynb b/notebooks/dask/dask_example.ipynb index a0686ce..b8ac14b 100644 --- a/notebooks/dask/dask_example.ipynb +++ b/notebooks/dask/dask_example.ipynb @@ -17,7 +17,8 @@ "metadata": {}, "outputs": [], "source": [ - "namespace = 'your-namespace-name' # put the name of your namespace here" + "with open(\"/var/run/secrets/kubernetes.io/serviceaccount/namespace\") as _f:\n", + " namespace = _f.read().strip()\n" ] }, { @@ -27,8 +28,7 @@ "metadata": {}, "outputs": [], "source": [ - "if namespace == 'your-namespace-name':\n", - " raise ValueError('You need to set the `namespace` variable to your personal namespace')" + "print(f\"Using namespace: {namespace}\")\n" ] }, { @@ -40,7 +40,7 @@ }, "outputs": [], "source": [ - "!pip install --upgrade ipywidgets dask-kubernetes==2024.9.0 pandas \"dask[complete]\"==2024.9.1" + "!pip install --upgrade ipywidgets \"dask-kubernetes==2026.3.0\" pandas \"dask[complete]==2026.3.0\" \"anyio>=4.12\"\n" ] }, { @@ -120,7 +120,7 @@ }, "outputs": [], "source": [ - "cluster = KubeCluster(name=\"test-cluster\", namespace=namespace, image=\"daskdev/dask:2024.9.1-py3.11\")\n", + "cluster = KubeCluster(name=\"test-cluster\", namespace=namespace, image=\"daskdev/dask:2026.3.0-py3.11\")\n", "cluster.scale(3)\n", "client = Client(cluster)" ] diff --git a/notebooks/mnist-vae/run_training.py b/notebooks/mnist-vae/run_training.py index e52458f..afa0cdd 100644 --- a/notebooks/mnist-vae/run_training.py +++ b/notebooks/mnist-vae/run_training.py @@ -1,3 +1,43 @@ +import os +import subprocess +import sys + +# Re-exec with conda's libstdc++ prepended to LD_LIBRARY_PATH when needed. +# scipy 1.17+ requires CXXABI_1.3.15 which the system libstdc++ may lack; the +# conda-bundled libstdc++ provides it. LD_LIBRARY_PATH must be set before the +# process starts (the dynamic linker reads it once), so we re-exec if the +# conda lib dir is not already in the path. +_conda_lib = os.path.normpath( + os.path.join(os.path.dirname(sys.executable), "..", "lib") +) +_ld = os.environ.get("LD_LIBRARY_PATH", "") +if os.path.isdir(_conda_lib) and _conda_lib not in _ld.split(":"): + os.execvpe( + sys.executable, + [sys.executable] + sys.argv, + {**os.environ, "LD_LIBRARY_PATH": f"{_conda_lib}:{_ld}"}, + ) + +# pytorch-lightning, torchvision, and tensorboard are not bundled in all notebook images +_missing = [] +try: + import pytorch_lightning # noqa: F401 +except ImportError: + _missing.append("pytorch-lightning") +try: + import torchvision # noqa: F401 +except ImportError: + _missing.append("torchvision") +try: + import tensorboard # noqa: F401 +except ImportError: + _missing.append("tensorboard") +if _missing: + subprocess.run( + [sys.executable, "-m", "pip", "install", "-q"] + _missing, + check=True, + ) + import click import pytorch_lightning as pl from pytorch_lightning.loggers import TensorBoardLogger @@ -5,16 +45,23 @@ from model.datamodule import MNISTDataModule import torch + @click.command() -@click.option('--hidden_dim', default=400, type=int, help='Dimension of the hidden layer.') -@click.option('--latent_dim', default=2, type=int, help='Dimension of the latent space.') -def run(hidden_dim: int, latent_dim: int) -> None: +@click.option( + "--hidden_dim", default=400, type=int, help="Dimension of the hidden layer." +) +@click.option( + "--latent_dim", default=2, type=int, help="Dimension of the latent space." +) +@click.option("--max_epochs", default=50, type=int, help="Number of training epochs.") +def run(hidden_dim: int, latent_dim: int, max_epochs: int) -> None: """ Train a VAE model on the MNIST dataset using PyTorch Lightning. Args: hidden_dim (int): Dimension of the hidden layer. latent_dim (int): Dimension of the latent space. + max_epochs (int): Number of training epochs. """ # Initialize data module @@ -23,8 +70,8 @@ def run(hidden_dim: int, latent_dim: int) -> None: # Initialize model model = VAE( - input_dim=784, # 28x28 pixels - hidden_dim=hidden_dim, # Dimension of the hidden layer + input_dim=784, # 28x28 pixels + hidden_dim=hidden_dim, # Dimension of the hidden layer latent_dim=latent_dim, # Dimension of the latent space ) @@ -33,14 +80,15 @@ def run(hidden_dim: int, latent_dim: int) -> None: # Initialize trainer trainer = pl.Trainer( - max_epochs=50, + max_epochs=max_epochs, accelerator="gpu" if torch.cuda.is_available() else "cpu", devices=1, - logger=logger + logger=logger, ) # Start training trainer.fit(model=model, datamodule=dm) -if __name__ == '__main__': + +if __name__ == "__main__": run() diff --git a/notebooks/mnist-vae/visualizations.ipynb b/notebooks/mnist-vae/visualizations.ipynb index 95a78da..354fa49 100644 --- a/notebooks/mnist-vae/visualizations.ipynb +++ b/notebooks/mnist-vae/visualizations.ipynb @@ -101,7 +101,11 @@ "cell_type": "code", "execution_count": null, "id": "d032c5d3-1476-40b6-b32f-68b462c27890", - "metadata": {}, + "metadata": { + "tags": [ + "ci-skip" + ] + }, "outputs": [], "source": [ "def generate_image(z1=0, z2=0):\n", diff --git a/pipelines/lightweight-components/mobile-price-classifications.ipynb b/pipelines/lightweight-components/mobile-price-classifications.ipynb index 5179316..d4954c4 100644 --- a/pipelines/lightweight-components/mobile-price-classifications.ipynb +++ b/pipelines/lightweight-components/mobile-price-classifications.ipynb @@ -125,7 +125,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"pyarrow\", \"s3fs\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def read_data(\n", " train_data_path: str,\n", @@ -160,7 +160,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def split_data(\n", " train_df: Input[Dataset],\n", @@ -208,7 +208,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def fit_scaler(\n", " train_x: Input[Dataset],\n", @@ -249,7 +249,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def tune_hyperparams(\n", " train_x: Input[Dataset],\n", @@ -319,7 +319,7 @@ "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def train_model(\n", " train_x: Input[Dataset],\n", @@ -357,7 +357,9 @@ { "cell_type": "markdown", "id": "40b9ef30-3352-44ec-a0e4-9e52f032194e", - "metadata": {}, + "metadata": { + "tags": [] + }, "source": [ "### Evaluate model on validation dataset" ] @@ -366,12 +368,14 @@ "cell_type": "code", "execution_count": null, "id": "3f9453f2-0c56-4c25-af05-7dfe42eaf3b1", - "metadata": {}, + "metadata": { + "tags": [] + }, "outputs": [], "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def evaluate_model(\n", " val_x: Input[Dataset],\n", @@ -416,7 +420,9 @@ { "cell_type": "markdown", "id": "836412b8-fe89-48e9-8383-53a5db9761bf", - "metadata": {}, + "metadata": { + "tags": [] + }, "source": [ "### Run predictions on test dataset" ] @@ -425,12 +431,14 @@ "cell_type": "code", "execution_count": null, "id": "12e40c2d-889e-4cc6-bb9d-a722653acb30", - "metadata": {}, + "metadata": { + "tags": [] + }, "outputs": [], "source": [ "@dsl.component(\n", " packages_to_install=[\"pandas\", \"scikit-learn\", \"pyarrow\", \"plotly\"],\n", - " base_image=\"python:3.9\",\n", + " base_image=\"python:3.11\",\n", ")\n", "def test_model(\n", " test_x: Input[Dataset],\n", @@ -478,7 +486,9 @@ { "cell_type": "markdown", "id": "604fab6f-5dc8-47e0-b15f-868032f076b2", - "metadata": {}, + "metadata": { + "tags": [] + }, "source": [ "## Build pipeline" ] @@ -622,7 +632,11 @@ { "cell_type": "markdown", "id": "2e73f8fc", - "metadata": {}, + "metadata": { + "tags": [ + "ci-skip" + ] + }, "source": [ "# Debugging with Lightweight Components\n", "Debugging can be challenging when using lightweight components in Kubeflow Pipelines. A practical approach is to download the artifacts from the steps preceding the failing one from object storage, and then run the functions used in the components locally. You can easily copy the paths to these artifacts from the Kubeflow Pipelines UI. Once you have these paths, you can use them as shown in the example below to download and read in Pandas DataFrames or perform similar operations. Make sure to adjust the paths according to your specific setup requirements." @@ -632,7 +646,11 @@ "cell_type": "code", "execution_count": null, "id": "315f6c39-4341-45a8-8c24-a81e0a92225f", - "metadata": {}, + "metadata": { + "tags": [ + "ci-skip" + ] + }, "outputs": [], "source": [ "storage_options={\n", @@ -646,7 +664,11 @@ "cell_type": "code", "execution_count": null, "id": "09f1655a-2e79-49c7-aae3-d698169cb7ee", - "metadata": {}, + "metadata": { + "tags": [ + "ci-skip" + ] + }, "outputs": [], "source": [ "# reading pandas dataframes (Modify paths!)\n", @@ -664,7 +686,11 @@ "cell_type": "code", "execution_count": null, "id": "f6fc60c0-714e-4623-bc2c-b3366de6f2b1", - "metadata": {}, + "metadata": { + "tags": [ + "ci-skip" + ] + }, "outputs": [], "source": [ "# use the mc tool to download model artifact\n", @@ -675,7 +701,11 @@ "cell_type": "code", "execution_count": null, "id": "e14a9baa-e110-45f1-90e9-c6eb2df71d9e", - "metadata": {}, + "metadata": { + "tags": [ + "ci-skip" + ] + }, "outputs": [], "source": [ "from joblib import load\n", diff --git a/pipelines/lightweight-python-package/Dockerfile b/pipelines/lightweight-python-package/Dockerfile index 071487b..7a50c53 100644 --- a/pipelines/lightweight-python-package/Dockerfile +++ b/pipelines/lightweight-python-package/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.9-slim +FROM python:3.11-slim WORKDIR /app diff --git a/pipelines/lightweight-python-package/pipeline.py b/pipelines/lightweight-python-package/pipeline.py index f1d89b2..f770aaf 100644 --- a/pipelines/lightweight-python-package/pipeline.py +++ b/pipelines/lightweight-python-package/pipeline.py @@ -2,10 +2,22 @@ from typing import Dict, List from kfp import dsl -from kfp.dsl import HTML, Input, Output, Dataset, Artifact, Model, ClassificationMetrics, Markdown +from kfp.dsl import ( + HTML, + Input, + Output, + Dataset, + Artifact, + Model, + ClassificationMetrics, + Markdown, +) -COMPONENTS_IMAGE = "europe-west3-docker.pkg.dev/prokube-internal/prokube-customer/mobile-price-classification:v1" +COMPONENTS_IMAGE = os.environ.get( + "COMPONENTS_IMAGE", + "europe-west3-docker.pkg.dev/prokube-internal/prokube-customer/mobile-price-classification:v2", +) @dsl.component(base_image=COMPONENTS_IMAGE) @@ -187,16 +199,18 @@ def mobile_price_classification_pipeline( test_data_path=test_data_path, ) # Use the cluster internal object storage endpoint - read_data_task.set_env_variable('AWS_ENDPOINT_URL', f'http://{os.environ["S3_ENDPOINT"]}') + read_data_task.set_env_variable( + "AWS_ENDPOINT_URL", f"http://{os.environ['S3_ENDPOINT']}" + ) # Use Kubernetes secrets to provide AWS credentials to the read_data component kubernetes.use_secret_as_env( read_data_task, - secret_name='s3creds', + secret_name="s3creds", secret_key_to_env={ - 'AWS_ACCESS_KEY_ID': 'AWS_ACCESS_KEY_ID', - 'AWS_SECRET_ACCESS_KEY': 'AWS_SECRET_ACCESS_KEY', - } - ) + "AWS_ACCESS_KEY_ID": "AWS_ACCESS_KEY_ID", + "AWS_SECRET_ACCESS_KEY": "AWS_SECRET_ACCESS_KEY", + }, + ) # Step 2: Split the data split_data_task = split_data( diff --git a/pipelines/lightweight-python-package/submit-cluster.py b/pipelines/lightweight-python-package/submit-cluster.py index d9499ae..fc4485b 100644 --- a/pipelines/lightweight-python-package/submit-cluster.py +++ b/pipelines/lightweight-python-package/submit-cluster.py @@ -4,7 +4,9 @@ if __name__ == "__main__": - with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace", "r", encoding="utf-8") as namespace_file: + with open( + "/var/run/secrets/kubernetes.io/serviceaccount/namespace", "r", encoding="utf-8" + ) as namespace_file: namespace = namespace_file.read().strip() s3_bucket = f"{namespace}-data" @@ -30,3 +32,4 @@ experiment_name="mobile-price-classification-containerized", run_name=f"Containerized pipeline {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", ) + print(f"KFP_RUN_ID={run.run_id}") diff --git a/pipelines/minimal-container-components/submit-cluster.py b/pipelines/minimal-container-components/submit-cluster.py index 334d1f8..6b95a77 100644 --- a/pipelines/minimal-container-components/submit-cluster.py +++ b/pipelines/minimal-container-components/submit-cluster.py @@ -7,15 +7,13 @@ if __name__ == "__main__": client = Client() - compiler.Compiler().compile(container_components_pipeline, 'pipeline.yaml') + compiler.Compiler().compile(container_components_pipeline, "pipeline.yaml") run = client.create_run_from_pipeline_package( - 'pipeline.yaml', + "pipeline.yaml", enable_caching=False, - arguments={ - "input1": "Hello", - "input2": "world!" - }, - experiment_name='minimal-container-components-experiment', - run_name=f'Simple pipeline {datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}' + arguments={"input1": "Hello", "input2": "world!"}, + experiment_name="minimal-container-components-experiment", + run_name=f"Simple pipeline {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", ) + print(f"KFP_RUN_ID={run.run_id}") diff --git a/scripts/get_or_create_api_key.py b/scripts/get_or_create_api_key.py new file mode 100644 index 0000000..e2eb076 --- /dev/null +++ b/scripts/get_or_create_api_key.py @@ -0,0 +1,217 @@ +""" +Utility to create or retrieve a prokube API key for use in examples. + +Usage from a notebook +--------------------- + import sys, subprocess, os + _root = subprocess.check_output(["git", "rev-parse", "--show-toplevel"], text=True).strip() + sys.path.insert(0, os.path.join(_root, "scripts")) + from get_or_create_api_key import get_or_create_api_key + + API_KEY = get_or_create_api_key() + +CLI usage +--------- + python get_or_create_api_key.py + # prints the key to stdout + +Resolution order +---------------- +The function returns a usable model-serving API key, choosing the mechanism +that fits the deployment it runs on: + +1. **New deployments (AI Gateway).** If the ``AIGatewayKey`` CRD is present, + the key is stored as an AIGatewayKey CR named ``examples-key`` backed by a + Secret named ``examples-key-secret`` in the notebook's namespace. Both + resources are created if they do not already exist. Re-running is safe. + +2. **Older deployments (hardcoded EnvoyFilter).** Older clusters do not have + the AIGatewayKey CRD; instead an admin pre-provisions a single API key and + injects it into the notebook pod via the ``INFERENCE_SERVICE_API_KEY`` + environment variable. When the CRD is absent, that env var is used. + +3. **Fallback.** If neither is available, the caller is prompted to paste the + key interactively, so the example degrades gracefully instead of crashing. +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import os +import secrets +import subprocess +import sys +from getpass import getpass + +_KEY_NAME = "examples-key" +_SECRET_NAME = "examples-key-secret" + +# Env var an admin injects into the notebook pod on older (pre-AI-Gateway) +# deployments that rely on a hardcoded EnvoyFilter for auth. +_API_KEY_ENV_VAR = "INFERENCE_SERVICE_API_KEY" + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl(*args: str, input: str | None = None) -> subprocess.CompletedProcess: + return subprocess.run( + ["kubectl", *args], + input=input, + capture_output=True, + text=True, + ) + + +def _aigatewaykey_crd_available() -> bool: + """Return True if the AIGatewayKey CRD is registered in the cluster. + + Uses the API discovery endpoint (kubectl api-resources) rather than + directly reading the CRD object, because notebook pod service accounts + typically lack RBAC permission to get cluster-scoped CRD resources. + API discovery is allowed for all authenticated Kubernetes users. + """ + result = _kubectl( + "api-resources", + "--api-group=prokube.ai", + "-o", + "name", + "--no-headers", + ) + return result.returncode == 0 and "aigatewaykeys" in result.stdout + + +def _key_exists(namespace: str) -> bool: + result = _kubectl( + "get", "aigatewaykey", _KEY_NAME, "-n", namespace, "--ignore-not-found" + ) + return bool(result.stdout.strip()) + + +def _read_key_from_secret(namespace: str) -> str: + result = _kubectl("get", "secret", _SECRET_NAME, "-n", namespace, "-o", "json") + if result.returncode != 0: + raise RuntimeError(f"Could not read secret {_SECRET_NAME}:\n{result.stderr}") + data = json.loads(result.stdout)["data"] + return base64.b64decode(data["token"]).decode() + + +def _apply(manifest: str, namespace: str) -> None: + result = _kubectl("apply", "-n", namespace, "-f", "-", input=manifest) + if result.returncode != 0: + raise RuntimeError(f"kubectl apply failed:\n{result.stderr}") + + +def _owner(namespace: str) -> str: + """Determine the key owner. + + Tries to read MLFLOW_TRACKING_USERNAME from the mlflow-credentials secret + (the canonical identity in the prokube platform). Falls back to a + namespace-scoped placeholder if the secret is absent. + """ + result = _kubectl( + "get", + "secret", + "mlflow-credentials", + "-n", + namespace, + "-o", + "jsonpath={.data.MLFLOW_TRACKING_USERNAME}", + ) + if result.returncode == 0 and result.stdout.strip(): + return base64.b64decode(result.stdout.strip()).decode() + return f"notebook-examples@{namespace}" + + +def _create_key(namespace: str) -> str: + key_value = f"pk_live_{secrets.token_hex(20)}" + owner = _owner(namespace) + + _apply( + f"apiVersion: v1\n" + f"kind: Secret\n" + f"metadata:\n" + f" name: {_SECRET_NAME}\n" + f" namespace: {namespace}\n" + f"stringData:\n" + f" token: {key_value}\n", + namespace, + ) + + _apply( + f"apiVersion: prokube.ai/v1alpha1\n" + f"kind: AIGatewayKey\n" + f"metadata:\n" + f" name: {_KEY_NAME}\n" + f" namespace: {namespace}\n" + f"spec:\n" + f" displayName: Examples API key\n" + f" owner: {owner}\n" + f" secretRef:\n" + f" name: {_SECRET_NAME}\n" + f" key: token\n" + f" scopes:\n" + f" - type: workspace\n", + namespace, + ) + + return key_value + + +def _key_from_env() -> str | None: + """Return the admin-provisioned API key from the environment, if set.""" + value = os.environ.get(_API_KEY_ENV_VAR, "").strip() + return value or None + + +def get_or_create_api_key(ns=None) -> str: + """Return a model-serving API key for the current namespace. + + See the module docstring for the resolution order: AIGatewayKey CR on new + deployments, the ``INFERENCE_SERVICE_API_KEY`` env var on older ones, and an + interactive prompt as a last resort. + """ + # New deployments: manage an AIGatewayKey CR (fully automated). + if _aigatewaykey_crd_available(): + ns = ns or _namespace() + if _key_exists(ns): + return _read_key_from_secret(ns) + return _create_key(ns) + + # Older deployments: admin injects the key via an env var. + env_key = _key_from_env() + if env_key: + return env_key + + # Fallback: prompt instead of crashing. + key = getpass( + f"AIGatewayKey CRD not found and ${_API_KEY_ENV_VAR} is unset. " + "Please enter the model-serving API key (ask your cluster admin): " + ).strip() + if not key: + raise RuntimeError( + "No API key available: the AIGatewayKey CRD is absent, " + f"${_API_KEY_ENV_VAR} is unset, and no key was entered." + ) + return key + + +if __name__ == "__main__": + try: + argparser = argparse.ArgumentParser( + description="Get or create a prokube API key" + ) + argparser.add_argument( + "--namespace", + "-n", + help="The Kubernetes namespace to use (defaults to the pods namespace)", + ) + args = argparser.parse_args() + print(get_or_create_api_key(args.namespace)) + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + sys.exit(1) diff --git a/scripts/setup_mlflow_credentials.py b/scripts/setup_mlflow_credentials.py new file mode 100644 index 0000000..b1b3011 --- /dev/null +++ b/scripts/setup_mlflow_credentials.py @@ -0,0 +1,125 @@ +""" +One-time setup: store MLflow credentials in a Kubernetes secret so that +example notebooks can read them without hardcoded values. + +This is the only step in the automation plan that requires human input — +the MLflow Personal Access Token (PAT) must be created via the MLflow UI +(Permissions → Create access key) and cannot be obtained programmatically. + +Usage from a notebook +--------------------- + import sys, subprocess, os + _root = subprocess.check_output(["git", "rev-parse", "--show-toplevel"], text=True).strip() + sys.path.insert(0, os.path.join(_root, "scripts")) + from setup_mlflow_credentials import setup_mlflow_credentials + + setup_mlflow_credentials( + uri="https://my-cluster.example.com/mlflow/", + username="alice@example.com", + password="", + ) + +CLI usage (any argument can be omitted; you will be prompted for it) +--------------------------------------------------------------------- + python setup_mlflow_credentials.py \\ + --uri https://my-cluster.example.com/mlflow/ \\ + --username alice@example.com \\ + --password + +The credentials are stored in a Secret named ``mlflow-credentials`` in the +notebook's namespace. Re-running updates the secret in place. +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + +_SECRET_NAME = "mlflow-credentials" + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _prompt(label: str, secret: bool = False) -> str: + if secret: + import getpass + + return getpass.getpass(f"{label}: ") + return input(f"{label}: ").strip() + + +def setup_mlflow_credentials( + uri: str | None = None, + username: str | None = None, + password: str | None = None, +) -> None: + """Create or update the ``mlflow-credentials`` secret. + + Any parameter left as ``None`` will be requested interactively. + """ + if uri is None: + print("MLflow tracking URI — typically https:///mlflow/") + uri = _prompt("MLFLOW_TRACKING_URI") + if username is None: + username = _prompt("MLFLOW_TRACKING_USERNAME (your login e-mail)") + if password is None: + password = _prompt( + "MLFLOW_TRACKING_PASSWORD (Personal Access Token)", secret=True + ) + + ns = _namespace() + + result = subprocess.run( + [ + "kubectl", + "create", + "secret", + "generic", + _SECRET_NAME, + "-n", + ns, + f"--from-literal=MLFLOW_TRACKING_URI={uri}", + f"--from-literal=MLFLOW_TRACKING_USERNAME={username}", + f"--from-literal=MLFLOW_TRACKING_PASSWORD={password}", + "--save-config", + "--dry-run=client", + "-o", + "yaml", + ], + capture_output=True, + text=True, + check=True, + ) + + apply = subprocess.run( + ["kubectl", "apply", "-n", ns, "-f", "-"], + input=result.stdout, + capture_output=True, + text=True, + ) + if apply.returncode != 0: + raise RuntimeError(f"kubectl apply failed:\n{apply.stderr}") + + print(f"Secret '{_SECRET_NAME}' created/updated in namespace '{ns}'.") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--uri", default=None, help="MLFLOW_TRACKING_URI") + parser.add_argument("--username", default=None, help="MLFLOW_TRACKING_USERNAME") + parser.add_argument( + "--password", default=None, help="MLFLOW_TRACKING_PASSWORD (PAT)" + ) + args = parser.parse_args() + + try: + setup_mlflow_credentials(args.uri, args.username, args.password) + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + sys.exit(1) diff --git a/serving/hf-vllm-completion/apply.py b/serving/hf-vllm-completion/apply.py new file mode 100644 index 0000000..8b13032 --- /dev/null +++ b/serving/hf-vllm-completion/apply.py @@ -0,0 +1,130 @@ +""" +Deploy the DistilBERT CPU InferenceService (HuggingFace backend) and smoke-test it. + +The model (distilbert-base-uncased-finetuned-sst-2-english, ~250 MB) is +downloaded from the HuggingFace Hub on first start, so the ISVC may take a +few minutes to become ready. + +Usage +----- + python apply.py +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +import urllib.error +import urllib.request + +_ISVC_NAME = "distilbert-inf-serv" +_MODEL_NAME = "distilbert" +_YAML = os.path.join(os.path.dirname(__file__), "inference-service-cpu.yaml") + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_apply(manifest: str, namespace: str) -> None: + result = subprocess.run( + ["kubectl", "apply", "-f", "-", "-n", namespace], + input=manifest, + text=True, + capture_output=True, + ) + if result.returncode != 0: + raise RuntimeError(result.stderr or result.stdout) + print(result.stdout.strip()) + + +def _wait_isvc_ready(name: str, namespace: str, timeout: int) -> None: + print( + f"Waiting for InferenceService '{name}' to become ready (timeout {timeout}s)..." + ) + result = subprocess.run( + [ + "kubectl", + "wait", + "inferenceservice", + name, + "--for=condition=Ready", + f"--timeout={timeout}s", + "-n", + namespace, + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + # Dump ISVC status to help diagnose + subprocess.run( + ["kubectl", "describe", "inferenceservice", name, "-n", namespace], + check=False, + ) + raise RuntimeError( + f"InferenceService '{name}' did not become ready within {timeout}s:\n" + + (result.stderr or result.stdout) + ) + + +def _smoke_test(namespace: str, timeout: int = 120) -> None: + """POST to the internal cluster URL; retries until the model responds.""" + url = ( + f"http://{_ISVC_NAME}.{namespace}.svc.cluster.local" + f"/v1/models/{_MODEL_NAME}:predict" + ) + payload = json.dumps({"instances": ["KServe is wonderful!"]}).encode() + deadline = time.time() + timeout + last_err: Exception | None = None + while time.time() < deadline: + try: + req = urllib.request.Request( + url, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=15) as resp: + body = json.loads(resp.read()) + if "predictions" not in body: + raise RuntimeError(f"unexpected response body: {body}") + print(f"Smoke test passed: {body}") + return + except Exception as exc: + last_err = exc + time.sleep(5) + raise RuntimeError(f"Smoke test failed after {timeout}s: {last_err}") + + +# ── Entry point ─────────────────────────────────────────────────────────────── + + +def deploy(timeout: int = 900) -> None: + ns = _namespace() + + with open(_YAML) as fh: + manifest = fh.read() + + _kubectl_apply(manifest, ns) + print(f"Applied InferenceService '{_ISVC_NAME}' in namespace '{ns}'.") + + _wait_isvc_ready(_ISVC_NAME, ns, timeout) + print(f"InferenceService '{_ISVC_NAME}' is ready.") + + _smoke_test(ns) + + +if __name__ == "__main__": + try: + deploy() + except Exception as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) diff --git a/serving/hf-vllm-completion/cleanup.py b/serving/hf-vllm-completion/cleanup.py new file mode 100644 index 0000000..a874d37 --- /dev/null +++ b/serving/hf-vllm-completion/cleanup.py @@ -0,0 +1,51 @@ +""" +Cleanup script for the hf-vllm-completion example. + +Deletes (idempotently): + - InferenceService ``distilbert-inf-serv`` + +Usage +----- + python cleanup.py [--dry-run] +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_delete(*args: str, dry_run: bool = False) -> None: + cmd = ["kubectl", "delete", *args, "--ignore-not-found"] + if dry_run: + print(f"[dry-run] {' '.join(cmd)}") + return + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f"WARNING: {result.stderr.strip()}", file=sys.stderr) + else: + print(result.stdout.strip() or f"deleted (or not found): {' '.join(args)}") + + +def cleanup(dry_run: bool = False) -> None: + ns = _namespace() + _kubectl_delete( + "inferenceservice", "distilbert-inf-serv", "-n", ns, dry_run=dry_run + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--dry-run", action="store_true", help="Print commands without executing" + ) + args = parser.parse_args() + cleanup(dry_run=args.dry_run) diff --git a/serving/kserve-keda-autoscaling/apply.py b/serving/kserve-keda-autoscaling/apply.py new file mode 100644 index 0000000..5f928f0 --- /dev/null +++ b/serving/kserve-keda-autoscaling/apply.py @@ -0,0 +1,205 @@ +""" +Deploy the KEDA autoscaling example (OPT-125M, vLLM backend, CPU) and verify +that both the InferenceService and the ScaledObject become active. + +KEDA CRD check +-------------- +If the ``scaledobjects.keda.sh`` CRD is absent the script exits with 0 and a +SKIP message rather than failing — the example is opt-in because KEDA is not +installed in every prokube cluster. + +Usage +----- + python apply.py +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +import urllib.error +import urllib.request + +_ISVC_NAME = "opt-125m" +_SCALED_OBJECT_NAME = "opt-125m-scaledobject" +_ISVC_YAML = os.path.join(os.path.dirname(__file__), "inference-service.yaml") +_SO_YAML = os.path.join(os.path.dirname(__file__), "scaled-object.yaml") + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_apply(manifest: str, namespace: str) -> None: + result = subprocess.run( + ["kubectl", "apply", "-f", "-", "-n", namespace], + input=manifest, + text=True, + capture_output=True, + ) + if result.returncode != 0: + raise RuntimeError(result.stderr or result.stdout) + print(result.stdout.strip()) + + +def _try_apply_scaledobject(manifest: str, namespace: str) -> str | None: + """Apply the ScaledObject manifest; return an error string if KEDA is absent. + + Returns None on success, or a non-empty string if kubectl rejected the + manifest because the ScaledObject CRD is not installed. Any other error + is raised so CI marks the example as FAIL instead of silently skipping. + """ + result = subprocess.run( + ["kubectl", "apply", "-f", "-", "-n", namespace], + input=manifest, + text=True, + capture_output=True, + ) + if result.returncode == 0: + print(result.stdout.strip()) + return None + stderr = result.stderr or result.stdout + if ( + "no matches for kind" in stderr + or "the server doesn't have a resource type" in stderr + ): + return stderr.strip() + raise RuntimeError(stderr) + + +def _wait_isvc_ready(name: str, namespace: str, timeout: int) -> None: + print( + f"Waiting for InferenceService '{name}' to become ready (timeout {timeout}s)..." + ) + result = subprocess.run( + [ + "kubectl", + "wait", + "inferenceservice", + name, + "--for=condition=Ready", + f"--timeout={timeout}s", + "-n", + namespace, + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + subprocess.run( + ["kubectl", "describe", "inferenceservice", name, "-n", namespace], + check=False, + ) + raise RuntimeError( + f"InferenceService '{name}' did not become ready within {timeout}s:\n" + + (result.stderr or result.stdout) + ) + + +def _wait_scaledobject_active(name: str, namespace: str, timeout: int = 120) -> None: + """Poll until KEDA reports the ScaledObject as Active=True.""" + print(f"Waiting for ScaledObject '{name}' to become active...") + deadline = time.time() + timeout + while time.time() < deadline: + r = subprocess.run( + [ + "kubectl", + "get", + "scaledobject", + name, + "-n", + namespace, + "-o", + "jsonpath={.status.conditions[?(@.type=='Active')].status}", + ], + capture_output=True, + text=True, + ) + if r.stdout.strip() == "True": + print(f"ScaledObject '{name}' is active.") + return + time.sleep(10) + raise RuntimeError( + f"ScaledObject '{name}' did not become active within {timeout}s. " + "Check 'kubectl describe scaledobject' for trigger errors." + ) + + +def _smoke_test(namespace: str, timeout: int = 60) -> None: + """Send a single completion request via the cluster-internal predictor Service.""" + # RawDeployment mode exposes the predictor as a plain Kubernetes Service named + # -predictor — no external gateway or API key required. + url = ( + f"http://opt-125m-predictor.{namespace}.svc.cluster.local/openai/v1/completions" + ) + payload = json.dumps( + {"model": "opt-125m", "prompt": "KServe is", "max_tokens": 8} + ).encode() + deadline = time.time() + timeout + last_err: Exception | None = None + while time.time() < deadline: + try: + req = urllib.request.Request( + url, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=20) as resp: + body = json.loads(resp.read()) + if "choices" not in body: + raise RuntimeError(f"unexpected response body: {body}") + text = body["choices"][0].get("text", "") + print(f"Smoke test passed: {text!r}") + return + except Exception as exc: + last_err = exc + time.sleep(5) + raise RuntimeError(f"Smoke test failed after {timeout}s: {last_err}") + + +# ── Entry point ─────────────────────────────────────────────────────────────── + + +def deploy(timeout: int = 900) -> None: + ns = _namespace() + + with open(_ISVC_YAML) as fh: + isvc_manifest = fh.read() + _kubectl_apply(isvc_manifest, ns) + print(f"Applied InferenceService '{_ISVC_NAME}' in namespace '{ns}'.") + + _wait_isvc_ready(_ISVC_NAME, ns, timeout) + print(f"InferenceService '{_ISVC_NAME}' is ready.") + + # Apply ScaledObject — detect KEDA availability via the apply itself rather + # than 'kubectl get crd' which requires cluster-level RBAC notebook SAs lack. + with open(_SO_YAML) as fh: + so_manifest = fh.read() + skip_reason = _try_apply_scaledobject(so_manifest, ns) + if skip_reason: + print( + f"SKIP: KEDA is not installed in this cluster " + f"(kubectl apply returned: {skip_reason}).\n" + "The ScaledObject was not created; the ISVC is running without autoscaling." + ) + sys.exit(0) + print(f"Applied ScaledObject '{_SCALED_OBJECT_NAME}' in namespace '{ns}'.") + + _wait_scaledobject_active(_SCALED_OBJECT_NAME, ns) + _smoke_test(ns) + + +if __name__ == "__main__": + try: + deploy() + except Exception as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) diff --git a/serving/kserve-keda-autoscaling/cleanup.py b/serving/kserve-keda-autoscaling/cleanup.py new file mode 100644 index 0000000..d0c5946 --- /dev/null +++ b/serving/kserve-keda-autoscaling/cleanup.py @@ -0,0 +1,53 @@ +""" +Cleanup script for the kserve-keda-autoscaling example. + +Deletes (idempotently, ScaledObject first to stop KEDA activity): + - ScaledObject ``opt-125m-scaledobject`` + - InferenceService ``opt-125m`` + +Usage +----- + python cleanup.py [--dry-run] +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_delete(*args: str, dry_run: bool = False) -> None: + cmd = ["kubectl", "delete", *args, "--ignore-not-found"] + if dry_run: + print(f"[dry-run] {' '.join(cmd)}") + return + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f"WARNING: {result.stderr.strip()}", file=sys.stderr) + else: + print(result.stdout.strip() or f"deleted (or not found): {' '.join(args)}") + + +def cleanup(dry_run: bool = False) -> None: + ns = _namespace() + # Delete the ScaledObject before the ISVC so KEDA stops managing the HPA + # before the underlying Deployment is gone. + _kubectl_delete("scaledobject", "opt-125m-scaledobject", "-n", ns, dry_run=dry_run) + _kubectl_delete("inferenceservice", "opt-125m", "-n", ns, dry_run=dry_run) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--dry-run", action="store_true", help="Print commands without executing" + ) + args = parser.parse_args() + cleanup(dry_run=args.dry_run) diff --git a/serving/minimal-example-shadow-deployment/apply.py b/serving/minimal-example-shadow-deployment/apply.py new file mode 100644 index 0000000..1bbe915 --- /dev/null +++ b/serving/minimal-example-shadow-deployment/apply.py @@ -0,0 +1,322 @@ +""" +Deploy the shadow deployment example and smoke-test the primary InferenceService. + +What this script does +--------------------- +1. Checks for the CrunchyData postgres-operator CRD — exits 0 with a SKIP + message if not found (it is installed by default in prokube clusters). +2. Deploys the PostgresCluster (postgres-cluster.yaml) and waits for the + primary pod to become ready. +3. Creates the two tables the transformer needs (inference_requests and + inference_response) via a short-lived psql pod. +4. Deploys the primary (doubler) and shadow (tripler) InferenceServices and + waits for both to become ready. +5. Sends a smoke-test request to the primary via its cluster-internal URL + and verifies the response contains predictions. + +What this script does NOT do +----------------------------- +- Apply the Istio VirtualService manifests — those embed namespace and domain + name and must be configured manually for each environment. +- Verify the shadow mirroring behaviour (that requires traffic via the + VirtualService, which is out of scope for automated CI). + +Usage +----- + python apply.py +""" + +from __future__ import annotations + +import base64 +import json +import os +import subprocess +import sys +import time +import urllib.error +import urllib.request + +_ROOT = os.path.dirname(__file__) + +_PG_CLUSTER_NAME = "inferencing-postgres" +_PG_SECRET_NAME = "inferencing-postgres-pguser-transformer-admin" +_PG_HOST_TEMPLATE = "inferencing-postgres-primary.{ns}.svc" +_PG_USER = "transformer-admin" +_PG_DB = "scale-inference" + +_DOUBLER_ISVC = "double-minimal-custom-inference" +_TRIPLER_ISVC = "triple-minimal-custom-inference" + +_SCHEMA_SQL = """\ +CREATE TABLE IF NOT EXISTS public.inference_requests ( + request_id uuid NOT NULL, + request_time timestamp with time zone NULL, + request_data json NULL, + predict_url text NULL, + created_at timestamp NULL, + PRIMARY KEY (request_id) +); +CREATE TABLE IF NOT EXISTS public.inference_response ( + request_id uuid NOT NULL, + request_data json NULL, + created_at timestamp NULL, + PRIMARY KEY (request_id) +); +""" + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_apply(manifest: str, namespace: str) -> None: + result = subprocess.run( + ["kubectl", "apply", "-f", "-", "-n", namespace], + input=manifest, + text=True, + capture_output=True, + ) + if result.returncode == 0: + print(result.stdout.strip()) + return + stderr = result.stderr or result.stdout + # Translate common failure modes into actionable messages + if ( + "no matches for kind" in stderr + or "the server doesn't have a resource type" in stderr + ): + raise RuntimeError( + "postgres-operator is not installed in this cluster. " + "The shadow deployment example requires the CrunchyData postgres-operator." + ) + if "Forbidden" in stderr and "postgres-operator.crunchydata.com" in stderr: + raise RuntimeError( + "The notebook ServiceAccount lacks RBAC permission to manage PostgresCluster " + "resources.\n" + "Ask your cluster admin to grant get/create/update/patch on " + "postgresclusters.postgres-operator.crunchydata.com in this namespace." + ) + raise RuntimeError(stderr) + + +def _wait_for_secret(name: str, namespace: str, timeout: int = 300) -> None: + """Block until the CrunchyData operator creates the user secret.""" + print( + f"Waiting for secret '{name}' to appear (postgres-operator creates it after cluster init)..." + ) + deadline = time.time() + timeout + while time.time() < deadline: + r = subprocess.run( + ["kubectl", "get", "secret", name, "-n", namespace], + capture_output=True, + text=True, + ) + if r.returncode == 0: + return + time.sleep(10) + raise RuntimeError( + f"Secret '{name}' did not appear within {timeout}s. " + "Check the PostgresCluster status." + ) + + +def _wait_pg_primary_ready(namespace: str, timeout: int = 300) -> None: + """Wait for the postgres primary pod to be ready.""" + print("Waiting for PostgresCluster primary pod to be ready...") + result = subprocess.run( + [ + "kubectl", + "wait", + "pod", + "-l", + f"postgres-operator.crunchydata.com/cluster={_PG_CLUSTER_NAME}," + "postgres-operator.crunchydata.com/role=master", + "--for=condition=Ready", + f"--timeout={timeout}s", + "-n", + namespace, + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + f"PostgresCluster primary did not become ready within {timeout}s:\n" + + (result.stderr or result.stdout) + ) + print("PostgresCluster primary is ready.") + + +def _get_secret_value(name: str, key: str, namespace: str) -> str: + result = subprocess.run( + [ + "kubectl", + "get", + "secret", + name, + "-n", + namespace, + "-o", + f"jsonpath={{.data.{key}}}", + ], + capture_output=True, + text=True, + ) + if result.returncode != 0 or not result.stdout.strip(): + raise RuntimeError( + f"Could not read key '{key}' from secret '{name}': {result.stderr}" + ) + return base64.b64decode(result.stdout.strip()).decode() + + +def _create_schema(namespace: str, password: str) -> None: + """Run a one-shot psql pod to create the required tables.""" + print("Creating database schema via temporary psql pod...") + host = _PG_HOST_TEMPLATE.format(ns=namespace) + result = subprocess.run( + [ + "kubectl", + "run", + "pg-schema-init", + "--rm", + "-i", + "--restart=Never", + f"-n={namespace}", + "--image=postgres:17", + f"--env=PGPASSWORD={password}", + "--", + "psql", + f"-h={host}", + f"-U={_PG_USER}", + f"-d={_PG_DB}", + ], + input=_SCHEMA_SQL, + text=True, + capture_output=True, + timeout=120, + ) + if result.returncode != 0: + raise RuntimeError(f"Schema creation failed:\n{result.stderr or result.stdout}") + print("Schema created (or already existed).") + + +def _wait_isvc_ready(name: str, namespace: str, timeout: int) -> None: + print(f"Waiting for InferenceService '{name}' (timeout {timeout}s)...") + result = subprocess.run( + [ + "kubectl", + "wait", + "inferenceservice", + name, + "--for=condition=Ready", + f"--timeout={timeout}s", + "-n", + namespace, + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + subprocess.run( + ["kubectl", "describe", "inferenceservice", name, "-n", namespace], + check=False, + ) + raise RuntimeError( + f"InferenceService '{name}' did not become ready within {timeout}s:\n" + + (result.stderr or result.stdout) + ) + + +def _smoke_test(namespace: str, timeout: int = 120) -> None: + """POST numeric values to the primary (doubler) ISVC and verify predictions. + + The doubler predictor multiplies each input value by FACTOR=2, so + [1.0, 2.0, 3.0] must produce predictions [2.0, 4.0, 6.0]. + """ + url = ( + f"http://{_DOUBLER_ISVC}.{namespace}.svc.cluster.local/v1/models/model:predict" + ) + inputs = [1.0, 2.0, 3.0] + expected = [2.0, 4.0, 6.0] + payload = json.dumps({"values": inputs}).encode() + deadline = time.time() + timeout + last_err: Exception | None = None + while time.time() < deadline: + try: + req = urllib.request.Request( + url, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=15) as resp: + body = json.loads(resp.read()) + predictions = body.get("predictions") + if predictions is None: + raise RuntimeError(f"no 'predictions' key in response: {body}") + if predictions != expected: + raise RuntimeError( + f"doubler (FACTOR=2) returned wrong predictions: " + f"got {predictions}, expected {expected}" + ) + print(f"Smoke test passed: {inputs} → {predictions} (FACTOR=2 verified)") + return + except Exception as exc: + last_err = exc + time.sleep(5) + raise RuntimeError(f"Smoke test failed after {timeout}s: {last_err}") + + +# ── Entry point ─────────────────────────────────────────────────────────────── + + +def deploy(timeout: int = 600) -> None: + ns = _namespace() + + # 1. Postgres cluster — _kubectl_apply raises with an actionable message on + # "no matches for kind" (operator not installed) or Forbidden (RBAC missing). + with open(os.path.join(_ROOT, "postgres-cluster.yaml")) as fh: + pg_manifest = fh.read() + _kubectl_apply(pg_manifest, ns) + print(f"Applied PostgresCluster '{_PG_CLUSTER_NAME}' in namespace '{ns}'.") + + _wait_pg_primary_ready(ns, timeout=300) + _wait_for_secret(_PG_SECRET_NAME, ns, timeout=120) + + password = _get_secret_value(_PG_SECRET_NAME, "password", ns) + _create_schema(ns, password) + + # 2. Both InferenceServices + for yaml_file in ( + "doubler-inference-service.yaml", + "tripler-inference-service.yaml", + ): + with open(os.path.join(_ROOT, yaml_file)) as fh: + manifest = fh.read() + _kubectl_apply(manifest, ns) + + print("Applied doubler and tripler InferenceServices.") + + for isvc_name in (_DOUBLER_ISVC, _TRIPLER_ISVC): + _wait_isvc_ready(isvc_name, ns, timeout) + + print("Both InferenceServices are ready.") + print( + "NOTE: Istio VirtualService manifests (istio/) are not applied in CI — " + "they require namespace and domain substitution and must be configured manually." + ) + + _smoke_test(ns) + + +if __name__ == "__main__": + try: + deploy() + except Exception as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) diff --git a/serving/minimal-example-shadow-deployment/cleanup.py b/serving/minimal-example-shadow-deployment/cleanup.py new file mode 100644 index 0000000..eafd42b --- /dev/null +++ b/serving/minimal-example-shadow-deployment/cleanup.py @@ -0,0 +1,59 @@ +""" +Cleanup script for the minimal-example-shadow-deployment example. + +Deletes (idempotently): + - InferenceService ``double-minimal-custom-inference`` + - InferenceService ``triple-minimal-custom-inference`` + - PostgresCluster ``inferencing-postgres`` + +Usage +----- + python cleanup.py [--dry-run] +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_delete(*args: str, dry_run: bool = False) -> None: + cmd = ["kubectl", "delete", *args, "--ignore-not-found"] + if dry_run: + print(f"[dry-run] {' '.join(cmd)}") + return + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f"WARNING: {result.stderr.strip()}", file=sys.stderr) + else: + print(result.stdout.strip() or f"deleted (or not found): {' '.join(args)}") + + +def cleanup(dry_run: bool = False) -> None: + ns = _namespace() + _kubectl_delete( + "inferenceservice", "double-minimal-custom-inference", "-n", ns, dry_run=dry_run + ) + _kubectl_delete( + "inferenceservice", "triple-minimal-custom-inference", "-n", ns, dry_run=dry_run + ) + _kubectl_delete( + "postgrescluster", "inferencing-postgres", "-n", ns, dry_run=dry_run + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--dry-run", action="store_true", help="Print commands without executing" + ) + args = parser.parse_args() + cleanup(dry_run=args.dry_run) diff --git a/serving/minimal-s3-model/cleanup.py b/serving/minimal-s3-model/cleanup.py new file mode 100644 index 0000000..9059dd3 --- /dev/null +++ b/serving/minimal-s3-model/cleanup.py @@ -0,0 +1,63 @@ +""" +Cleanup script for the minimal-s3-model example. + +Deletes (idempotently): + - InferenceService ``kserve-object-storage-test`` + - AIGatewayKey ``examples-key`` and its backing Secret ``examples-key-secret`` + (only when --include-api-key is passed, since the key is shared across examples) + +Usage +----- + python cleanup.py [--include-api-key] [--dry-run] +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_delete(*args: str, dry_run: bool = False) -> None: + cmd = ["kubectl", "delete", *args, "--ignore-not-found"] + if dry_run: + print(f"[dry-run] {' '.join(cmd)}") + return + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f"WARNING: {result.stderr.strip()}", file=sys.stderr) + else: + print(result.stdout.strip() or f"deleted (or not found): {' '.join(args)}") + + +def cleanup(include_api_key: bool = False, dry_run: bool = False) -> None: + ns = _namespace() + + _kubectl_delete( + "inferenceservice", "kserve-object-storage-test", "-n", ns, dry_run=dry_run + ) + + if include_api_key: + _kubectl_delete("aigatewaykey", "examples-key", "-n", ns, dry_run=dry_run) + _kubectl_delete("secret", "examples-key-secret", "-n", ns, dry_run=dry_run) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--include-api-key", + action="store_true", + help="Also delete the shared AIGatewayKey", + ) + parser.add_argument( + "--dry-run", action="store_true", help="Print commands without executing them" + ) + args = parser.parse_args() + cleanup(include_api_key=args.include_api_key, dry_run=args.dry_run) diff --git a/serving/minimal-s3-model/minimal-s3-model.ipynb b/serving/minimal-s3-model/minimal-s3-model.ipynb index 6d1b063..df52e86 100644 --- a/serving/minimal-s3-model/minimal-s3-model.ipynb +++ b/serving/minimal-s3-model/minimal-s3-model.ipynb @@ -297,7 +297,21 @@ "id": "2df5cbae-f406-4c33-be90-dbb0b61cf81d", "metadata": {}, "source": [ - "Enter the model serving API key in the cell below. If not known, ask the cluster administrator for it." + "
\n", + " 💡 Tip:\n", + "
\n", + " The cell below automatically creates a dedicated examples-key AIGatewayKey and its backing Secret in your namespace — no manual steps required.\n", + "

\n", + " If you'd rather use your own key, replace the cell contents with:\n", + " 
INFERENCE_SERVICE_API_KEY = \"your-key-here\"
\n", + "
\n", + "
" ] }, { @@ -307,9 +321,12 @@ "metadata": {}, "outputs": [], "source": [ - "INFERENCE_SERVICE_API_KEY = \"\"\n", - "if not INFERENCE_SERVICE_API_KEY:\n", - " raise RuntimeError(\"Please provide the API Key that will be used to test the deployed InferenceService\")" + "import sys, subprocess, os\n", + "_root = subprocess.check_output([\"git\", \"rev-parse\", \"--show-toplevel\"], text=True).strip()\n", + "sys.path.insert(0, os.path.join(_root, \"scripts\"))\n", + "from get_or_create_api_key import get_or_create_api_key\n", + "\n", + "INFERENCE_SERVICE_API_KEY = get_or_create_api_key()\n" ] }, { diff --git a/serving/mlflow-kserve-inference-protocols/README.md b/serving/mlflow-kserve-inference-protocols/README.md new file mode 100644 index 0000000..ddea2de --- /dev/null +++ b/serving/mlflow-kserve-inference-protocols/README.md @@ -0,0 +1,118 @@ +# MLflow KServe Inference Protocols + +This directory demonstrates both the v1 and v2 inference protocols for +MLflow-tracked models deployed as KServe InferenceServices on prokube. + +It uses two ISVCs side-by-side: + +| ISVC | Model format | Protocol | YAML | +|---|---|---|---| +| `v1-mobile-price-classification-inference` | `sklearn` | v1 | `v1-InferenceService.yaml` | +| `v2-mobile-price-classification-inference` | `mlflow` | v2 | `v2-InferenceService.yaml` | + +The notebook `inference_protocol_version_example.ipynb` walks through deploying +both and comparing request/response shapes for each protocol. + +## Prerequisites + +- A model trained and registered in MLflow. See the + [mobile price classification MLflow example](../../mlflow/mobile-price-classification/) + for how to train and register the SVM model used here. +- `kubectl` access to your prokube namespace. (already installed in a pk-notebook) +- Python with the `requests` package installed (for testing, already installed in a pk-notebook) + +## Why a dedicated ServiceAccount is required + +> [!IMPORTANT] +> MLflow ISVCs **must** use the dedicated `mlflow-isvc-sa` ServiceAccount +> defined in `ServiceAccount.yaml`. Do not use the namespace `default` SA. + +### Root cause: KServe S3 credential injection conflict + +KServe v0.18.0 automatically injects S3 credentials into the +`storage-initializer` init container for every ISVC whose ServiceAccount +references a secret annotated as an S3 credential source (e.g. the +`s3creds` secret present in every prokube namespace). + +The injection produces two kinds of env entries for the same variable names +(`S3_ENDPOINT`, `AWS_ENDPOINT_URL`, `S3_USE_HTTPS`, etc.): + +| Source | Env form | +|---|---| +| Secret annotations (`serving.kserve.io/s3-endpoint`, `s3-usehttps`, …) | `value: ` | +| Secret keys (mounted via `valueFrom.secretKeyRef`) | `valueFrom: {secretKeyRef: …}` | + +Kubernetes rejects a container spec that sets **both** `value` and +`valueFrom` on the same env var. The admission webhook blocks the pod, the +storage-initializer never runs, and the ISVC times out with +`timed out waiting for condition`. + +This only affects `mlflow://` ISVCs: S3-backed ISVCs (`s3://`) rely on +exactly that injection and work correctly with the `default` SA. + +### The fix + +`ServiceAccount.yaml` defines `mlflow-isvc-sa`, a dedicated SA that: + +- Holds the `regcred-prokube` and `regcred-dev` imagePullSecrets so that + the private prokube storage-initializer image can be pulled. +- Does **not** reference `s3creds`, so KServe skips S3 credential injection + entirely for any ISVC that uses it. + +Both `v1-InferenceService.yaml` and `v2-InferenceService.yaml` already +reference this SA. + +## API key + +> [!TIP] +> `apply.py` (and the notebook's `deploy()` call) automatically creates a +> dedicated `examples-key` AIGatewayKey and its backing Secret +> (`examples-key-secret`) in your namespace. If you'd rather use your own key, +> pass it to the smoke-test step directly or set `API_KEY` in the environment +> before calling `apply.py`. + +## Deploy + +1. Apply the ServiceAccount (once per namespace): + + ```sh + kubectl apply -f ServiceAccount.yaml + ``` + +2. Replace the placeholder values in both ISVC YAMLs: + + | Placeholder | Description | + |---|---| + | `` | Your Kubeflow namespace / workspace | + | `` | Your username, matching the model registered in MLflow | + +3. Apply both ISVCs: + + ```sh + kubectl apply -f v1-InferenceService.yaml -n + kubectl apply -f v2-InferenceService.yaml -n + ``` + +4. Wait for both to become ready: + + ```sh + kubectl get inferenceservice -n + ``` + + > [!WARNING] + > You need a MLFlow ClusterStorageContainer in order to use the + > `mlflow://` scheme (prokube platform versions >= 1.7.0) + +## Cleanup + +To delete both ISVCs: + +```sh +python cleanup.py +``` + +Or with `--dry-run` to preview the commands: + +```sh +python cleanup.py --dry-run +``` diff --git a/serving/mlflow-kserve-inference-protocols/ServiceAccount.yaml b/serving/mlflow-kserve-inference-protocols/ServiceAccount.yaml new file mode 100644 index 0000000..1f2b812 --- /dev/null +++ b/serving/mlflow-kserve-inference-protocols/ServiceAccount.yaml @@ -0,0 +1,19 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: mlflow-isvc-sa +# imagePullSecrets grant access to the private prokube container registry +# where the custom MLflow storage-initializer image lives. +# These secrets are provisioned automatically by the prokube platform. +imagePullSecrets: + - name: regcred-prokube + - name: regcred-dev +# NOTE: intentionally does not reference the 's3creds' Secret. +# KServe v0.18.0 injects S3 env vars into the storage-initializer init +# container for every ISVC whose ServiceAccount references 's3creds'. When +# the storageUri uses the mlflow:// scheme the injected env contains both a +# plain 'value' field (from the secret's annotations) and a 'valueFrom' +# field on the same variable names (S3_ENDPOINT, AWS_ENDPOINT_URL, +# S3_USE_HTTPS). Kubernetes rejects that combination at admission, so pods +# are never created and the ISVC times out. By using a dedicated SA without +# 's3creds' the injection is skipped entirely. diff --git a/serving/mlflow-kserve-inference-protocols/apply.py b/serving/mlflow-kserve-inference-protocols/apply.py new file mode 100644 index 0000000..eb83970 --- /dev/null +++ b/serving/mlflow-kserve-inference-protocols/apply.py @@ -0,0 +1,223 @@ +""" +Deploy the v1 and v2 MLflow KServe InferenceServices and smoke-test both. + +Substitutes ```` and ```` in the YAML manifests from +the cluster environment, applies them, waits for readiness, and verifies that +both endpoints return identical predictions for the same input. + +Usage from a notebook +--------------------- + import sys + sys.path.insert(0, ".") # run from the serving/mlflow-kserve-inference-protocols/ dir + from apply import deploy + + v1_uri, v2_uri, api_key = deploy() + +CLI usage +--------- + python apply.py + +Prerequisites +------------- +- ``mlflow-credentials`` secret must exist (run scripts/setup_mlflow_credentials.py). +- The MLflow model ``mobile-price-svm-`` version 1 must be registered + (run mlflow/mobile-price-classification/mlflow-mobile-price-classification.ipynb first). +""" + +from __future__ import annotations + +import base64 +import json +import os +import subprocess +import sys +import urllib.error +import urllib.request + +_HERE = os.path.dirname(__file__) + +_SA_YAML = os.path.join(_HERE, "ServiceAccount.yaml") +_ISVC_NAMES = { + "v1": "v1-mobile-price-classification-inference", + "v2": "v2-mobile-price-classification-inference", +} +_YAML_FILES = { + "v1": os.path.join(_HERE, "v1-InferenceService.yaml"), + "v2": os.path.join(_HERE, "v2-InferenceService.yaml"), +} +_BODY_FILES = { + "v1": os.path.join(_HERE, "v1-mlflow-inference-body.json"), + "v2": os.path.join(_HERE, "v2-mlflow-inference-body.json"), +} + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _mlflow_username(namespace: str) -> str: + result = subprocess.run( + [ + "kubectl", + "get", + "secret", + "mlflow-credentials", + "-n", + namespace, + "-o", + "json", + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + "mlflow-credentials secret not found.\n" + "Run scripts/setup_mlflow_credentials.py to create it." + ) + data = json.loads(result.stdout)["data"] + return base64.b64decode(data["MLFLOW_TRACKING_USERNAME"]).decode().split("@")[0] + + +def _apply_yaml(yaml_file: str, namespace: str, username: str) -> None: + manifest = open(yaml_file).read() + manifest = manifest.replace("", namespace).replace( + "", username + ) + result = subprocess.run( + ["kubectl", "apply", "-n", namespace, "-f", "-"], + input=manifest, + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + f"kubectl apply failed for {os.path.basename(yaml_file)}:\n{result.stderr}" + ) + print(result.stdout.strip()) + + +def _wait_ready(name: str, namespace: str, timeout: int = 600) -> None: + print(f" Waiting for {name} (timeout {timeout}s)...") + result = subprocess.run( + [ + "kubectl", + "wait", + "inferenceservice", + name, + "-n", + namespace, + "--for=condition=Ready", + f"--timeout={timeout}s", + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + subprocess.run( + ["kubectl", "describe", "inferenceservice", name, "-n", namespace] + ) + raise RuntimeError(f"{name} did not become ready:\n{result.stderr}") + + +def _isvc_url(name: str, namespace: str) -> str: + return subprocess.check_output( + [ + "kubectl", + "get", + "inferenceservice", + name, + "-n", + namespace, + "-o", + "jsonpath={.status.url}", + ], + text=True, + ).strip() + + +def _get_api_key() -> str: + root = subprocess.check_output( + ["git", "rev-parse", "--show-toplevel"], text=True + ).strip() + sys.path.insert(0, os.path.join(root, "scripts")) + from get_or_create_api_key import get_or_create_api_key + + return get_or_create_api_key() + + +def _smoke_test(uri: str, name: str, protocol: str, api_key: str) -> None: + """POST one inference request; raise on non-2xx or missing predictions.""" + body = json.load(open(_BODY_FILES[protocol])) + if protocol == "v1": + url = f"{uri}/v1/models/{name}:predict" + pred_key = "predictions" + else: + url = f"{uri}/v2/models/{name}/infer" + pred_key = "outputs" + + data = json.dumps(body).encode() + req = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/json", "X-Api-Key": api_key}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + result = json.loads(resp.read()) + except urllib.error.HTTPError as exc: + raise RuntimeError( + f"Smoke test {protocol} returned HTTP {exc.code}: {exc.read().decode()}" + ) from exc + + if pred_key not in result: + raise RuntimeError(f"Smoke test {protocol}: unexpected response: {result}") + print(f" [{protocol}] smoke test passed — {len(result[pred_key])} prediction(s)") + + +# ── Entry point ─────────────────────────────────────────────────────────────── + + +def deploy(timeout: int = 600) -> tuple[str, str, str]: + """Deploy both ISVCs, wait for readiness, and return (v1_uri, v2_uri, api_key).""" + ns = _namespace() + username = _mlflow_username(ns) + + print("Applying ServiceAccount...") + _apply_yaml(_SA_YAML, ns, username) + + print("Applying InferenceService manifests...") + for proto, yaml_file in _YAML_FILES.items(): + _apply_yaml(yaml_file, ns, username) + + print("Waiting for InferenceServices to become ready...") + for proto, name in _ISVC_NAMES.items(): + _wait_ready(name, ns, timeout) + + v1_uri = _isvc_url(_ISVC_NAMES["v1"], ns) + v2_uri = _isvc_url(_ISVC_NAMES["v2"], ns) + api_key = _get_api_key() + + print(f"v1 URI: {v1_uri}") + print(f"v2 URI: {v2_uri}") + + print("Running smoke tests...") + _smoke_test(v1_uri, _ISVC_NAMES["v1"], "v1", api_key) + _smoke_test(v2_uri, _ISVC_NAMES["v2"], "v2", api_key) + + return v1_uri, v2_uri, api_key + + +if __name__ == "__main__": + try: + v1_uri, v2_uri, api_key = deploy() + print(f"ISVC_URI_V1={v1_uri}") + print(f"ISVC_URI_V2={v2_uri}") + except Exception as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) diff --git a/serving/mlflow-kserve-inference-protocols/cleanup.py b/serving/mlflow-kserve-inference-protocols/cleanup.py new file mode 100644 index 0000000..f4fb07d --- /dev/null +++ b/serving/mlflow-kserve-inference-protocols/cleanup.py @@ -0,0 +1,56 @@ +""" +Cleanup script for the mlflow-kserve-inference-protocols example. + +Deletes (idempotently): + - InferenceService ``v1-mobile-price-classification-inference`` + - InferenceService ``v2-mobile-price-classification-inference`` + +Usage +----- + python cleanup.py [--dry-run] +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + +_ISVC_NAMES = [ + "v1-mobile-price-classification-inference", + "v2-mobile-price-classification-inference", +] + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_delete(*args: str, dry_run: bool = False) -> None: + cmd = ["kubectl", "delete", *args, "--ignore-not-found"] + if dry_run: + print(f"[dry-run] {' '.join(cmd)}") + return + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f"WARNING: {result.stderr.strip()}", file=sys.stderr) + else: + print(result.stdout.strip() or f"deleted (or not found): {' '.join(args)}") + + +def cleanup(dry_run: bool = False) -> None: + ns = _namespace() + for name in _ISVC_NAMES: + _kubectl_delete("inferenceservice", name, "-n", ns, dry_run=dry_run) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--dry-run", action="store_true", help="Print commands without executing them" + ) + args = parser.parse_args() + cleanup(dry_run=args.dry_run) diff --git a/serving/mlflow-kserve-inference-protocols/inference_protocol_version_example.ipynb b/serving/mlflow-kserve-inference-protocols/inference_protocol_version_example.ipynb index d6971c5..710c07b 100644 --- a/serving/mlflow-kserve-inference-protocols/inference_protocol_version_example.ipynb +++ b/serving/mlflow-kserve-inference-protocols/inference_protocol_version_example.ipynb @@ -53,8 +53,8 @@ "metadata": {}, "outputs": [], "source": [ - "v1_yaml = Path(\"v1_InferenceService.yaml\").read_text()\n", - "v2_yaml = Path(\"v2_InferenceService.yaml\").read_text()\n", + "v1_yaml = Path(\"v1-InferenceService.yaml\").read_text()\n", + "v2_yaml = Path(\"v2-InferenceService.yaml\").read_text()\n", "\n", "display(Markdown(\n", " \"| v1 InferenceService | v2 InferenceService |\\n\"\n", @@ -340,10 +340,27 @@ "source": [ "---\n", "\n", - "## 6. Live Testing (Optional)\n", + "## 6. Live Testing\n", "\n", - "If you have deployed both InferenceServices, you can test them here. \n", - "Fill in your connection details and run the cells below." + "The cells below deploy both InferenceServices, wait for them to become ready,\n", + "and run inference requests automatically.\n", + "\n", + "**Prerequisite:** the `mlflow-credentials` secret must exist in your namespace\n", + "(run `scripts/setup_mlflow_credentials.py` once if you have not done so).\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "auto-setup-isvc", + "metadata": {}, + "outputs": [], + "source": [ + "import sys\n", + "sys.path.insert(0, \".\")\n", + "from apply import deploy\n", + "\n", + "_uri_v1, _uri_v2, _api_key = deploy()\n" ] }, { @@ -356,27 +373,24 @@ "import requests\n", "\n", "# ---------------------------------------------------------------------------\n", - "# Connection settings (fill these in)\n", + "# Connection settings (auto-populated by the setup cell above)\n", "# ---------------------------------------------------------------------------\n", - "INFERENCE_SERVICE_URI_V1 = \"\" # external URL, no trailing slash (copied from KServe InferenceService status.url)\n", - "INFERENCE_SERVICE_URI_V2 = \"\" # external URL, no trailing slash (copied from KServe InferenceService status.url)\n", - "API_KEY = \"\" # API key, provided by your administrator \n", + "INFERENCE_SERVICE_URI_V1 = _uri_v1\n", + "INFERENCE_SERVICE_URI_V2 = _uri_v2\n", + "API_KEY = _api_key\n", "\n", "# ---------------------------------------------------------------------------\n", "# Model names (must match metadata.name in the InferenceService YAMLs)\n", "# ---------------------------------------------------------------------------\n", - "MODEL_NAME_V1 = \"v1-mobile-price-classification-inference\" # from v1_InferenceService.yaml\n", - "MODEL_NAME_V2 = \"v2-mobile-price-classification-inference\" # from v2_InferenceService.yaml\n", + "MODEL_NAME_V1 = \"v1-mobile-price-classification-inference\"\n", + "MODEL_NAME_V2 = \"v2-mobile-price-classification-inference\"\n", "\n", "HEADERS = {\"X-Api-Key\": API_KEY, \"Content-Type\": \"application/json\"}\n", "\n", - "if not INFERENCE_SERVICE_URI_V1 or not INFERENCE_SERVICE_URI_V2 or not API_KEY:\n", - " raise ValueError(\"Please fill in the connection settings (URIs and API key) before running the tests.\")\n", - "\n", - "print(f\"v1 URI : {INFERENCE_SERVICE_URI_V1}\")\n", - "print(f\"v2 URI : {INFERENCE_SERVICE_URI_V2}\")\n", - "print(f\"v1 model : {MODEL_NAME_V1}\")\n", - "print(f\"v2 model : {MODEL_NAME_V2}\")" + "print(f\"v1 URI : {INFERENCE_SERVICE_URI_V1}\")\n", + "print(f\"v2 URI : {INFERENCE_SERVICE_URI_V2}\")\n", + "print(f\"v1 model: {MODEL_NAME_V1}\")\n", + "print(f\"v2 model: {MODEL_NAME_V2}\")\n" ] }, { diff --git a/serving/mlflow-kserve-inference-protocols/v1-InferenceService.yaml b/serving/mlflow-kserve-inference-protocols/v1-InferenceService.yaml index c2b2f1f..bb11b3e 100644 --- a/serving/mlflow-kserve-inference-protocols/v1-InferenceService.yaml +++ b/serving/mlflow-kserve-inference-protocols/v1-InferenceService.yaml @@ -5,6 +5,7 @@ metadata: namespace: "" spec: predictor: + serviceAccountName: mlflow-isvc-sa model: modelFormat: name: sklearn diff --git a/serving/mlflow-kserve-inference-protocols/v2-InferenceService.yaml b/serving/mlflow-kserve-inference-protocols/v2-InferenceService.yaml index 1f4db07..c933783 100644 --- a/serving/mlflow-kserve-inference-protocols/v2-InferenceService.yaml +++ b/serving/mlflow-kserve-inference-protocols/v2-InferenceService.yaml @@ -5,6 +5,7 @@ metadata: namespace: "" spec: predictor: + serviceAccountName: mlflow-isvc-sa model: modelFormat: name: mlflow diff --git a/serving/mlflow-kserve-minimal/InferenceService.yaml b/serving/mlflow-kserve-minimal/InferenceService.yaml index ecbdc2d..6000116 100644 --- a/serving/mlflow-kserve-minimal/InferenceService.yaml +++ b/serving/mlflow-kserve-minimal/InferenceService.yaml @@ -5,6 +5,7 @@ metadata: namespace: "" spec: predictor: + serviceAccountName: mlflow-isvc-sa model: modelFormat: name: mlflow diff --git a/serving/mlflow-kserve-minimal/README.md b/serving/mlflow-kserve-minimal/README.md index b3c7461..2a8ac2a 100644 --- a/serving/mlflow-kserve-minimal/README.md +++ b/serving/mlflow-kserve-minimal/README.md @@ -12,9 +12,55 @@ InferenceService using the v2 inference protocol. It uses the built-in - `kubectl` access to your prokube namespace. (already installed in a pk-notebook) - Python with the `requests` package installed (for testing, already installed in a pk-notebook) +## Why a dedicated ServiceAccount is required + +> [!IMPORTANT] +> MLflow ISVCs **must** use the dedicated `mlflow-isvc-sa` ServiceAccount +> defined in `ServiceAccount.yaml`. Do not use the namespace `default` SA. + +### Root cause: KServe S3 credential injection conflict + +KServe v0.18.0 automatically injects S3 credentials into the +`storage-initializer` init container for every ISVC whose ServiceAccount +references a secret annotated as an S3 credential source (e.g. the +`s3creds` secret present in every prokube namespace). + +The injection produces two kinds of env entries for the same variable names +(`S3_ENDPOINT`, `AWS_ENDPOINT_URL`, `S3_USE_HTTPS`, etc.): + +| Source | Env form | +|---|---| +| Secret annotations (`serving.kserve.io/s3-endpoint`, `s3-usehttps`, …) | `value: ` | +| Secret keys (mounted via `valueFrom.secretKeyRef`) | `valueFrom: {secretKeyRef: …}` | + +Kubernetes rejects a container spec that sets **both** `value` and +`valueFrom` on the same env var. The admission webhook blocks the pod, the +storage-initializer never runs, and the ISVC times out with +`timed out waiting for condition`. + +This only affects `mlflow://` ISVCs: S3-backed ISVCs (`s3://`) rely on +exactly that injection and work correctly with the `default` SA. + +### The fix + +`ServiceAccount.yaml` defines `mlflow-isvc-sa`, a dedicated SA that: + +- Holds the `regcred-prokube` and `regcred-dev` imagePullSecrets so that + the private prokube storage-initializer image can be pulled. +- Does **not** reference `s3creds`, so KServe skips S3 credential injection + entirely for any ISVC that uses it. + +Both `InferenceService.yaml` and `apply.py` already reference this SA. + ## Deploy the InferenceService -1. Open `InferenceService.yaml` and replace the placeholder values: +1. Apply the ServiceAccount (once per namespace): + + ```sh + kubectl apply -f ServiceAccount.yaml + ``` + +2. Open `InferenceService.yaml` and replace the placeholder values: | Placeholder | Description | |---|---| @@ -32,29 +78,39 @@ InferenceService using the v2 inference protocol. It uses the built-in > You need to the a MLFlow ClusterStorageContainer in order to use the > `mlflow://` scheme (prokube platform versions >= 1.7.0) -2. Apply the manifest: +3. Apply the manifest: ```sh kubectl apply -f InferenceService.yaml -n ``` -3. Wait for the InferenceService to become ready. You can check the status in +4. Wait for the InferenceService to become ready. You can check the status in the Kubeflow Endpoints UI or via: ```sh kubectl get inferenceservice -n ``` +Alternatively, `apply.py` handles steps 1–4 automatically (see below). + ## Test the Deployment A test script and sample request body are provided to verify the deployment. +> [!TIP] +> `apply.py` automatically creates a dedicated `examples-key` AIGatewayKey and +> its backing Secret (`examples-key-secret`) in your namespace — no manual key +> management required. If you'd rather supply your own key, set `API_KEY` +> before running the test script: +> ```sh +> export API_KEY=your-key-here +> ``` + 1. Set the required environment variables (optional): ```sh - export API_KEY= + export API_KEY= # auto-created by apply.py if omitted export INFERENCE_SERVICE_URI= export PROTOCOL_VERSION=v2 ``` - You can find the inference service URL in the Kubeflow Endpoints UI. If you - don't know your API key, reach out to your prokube admin. + You can find the inference service URL in the Kubeflow Endpoints UI. 2. Run the test script: ```sh diff --git a/serving/mlflow-kserve-minimal/ServiceAccount.yaml b/serving/mlflow-kserve-minimal/ServiceAccount.yaml new file mode 100644 index 0000000..1f2b812 --- /dev/null +++ b/serving/mlflow-kserve-minimal/ServiceAccount.yaml @@ -0,0 +1,19 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: mlflow-isvc-sa +# imagePullSecrets grant access to the private prokube container registry +# where the custom MLflow storage-initializer image lives. +# These secrets are provisioned automatically by the prokube platform. +imagePullSecrets: + - name: regcred-prokube + - name: regcred-dev +# NOTE: intentionally does not reference the 's3creds' Secret. +# KServe v0.18.0 injects S3 env vars into the storage-initializer init +# container for every ISVC whose ServiceAccount references 's3creds'. When +# the storageUri uses the mlflow:// scheme the injected env contains both a +# plain 'value' field (from the secret's annotations) and a 'valueFrom' +# field on the same variable names (S3_ENDPOINT, AWS_ENDPOINT_URL, +# S3_USE_HTTPS). Kubernetes rejects that combination at admission, so pods +# are never created and the ISVC times out. By using a dedicated SA without +# 's3creds' the injection is skipped entirely. diff --git a/serving/mlflow-kserve-minimal/apply.py b/serving/mlflow-kserve-minimal/apply.py new file mode 100644 index 0000000..339185f --- /dev/null +++ b/serving/mlflow-kserve-minimal/apply.py @@ -0,0 +1,197 @@ +""" +Deploy the mlflow-kserve-minimal InferenceService and run an inference test. + +Substitutes ```` and ```` in InferenceService.yaml +from the cluster environment, applies it, waits for readiness, runs a smoke +test via ``test_inference_service.py``, and prints the external URL. + +Usage from a notebook +--------------------- + import sys, subprocess, os + _root = subprocess.check_output(["git", "rev-parse", "--show-toplevel"], text=True).strip() + sys.path.insert(0, os.path.join(_root, "serving", "mlflow-kserve-minimal")) + from apply import deploy + + isvc_uri = deploy() + +CLI usage +--------- + python apply.py + # prints ISVC_URI= and runs inference smoke test + +Prerequisites +------------- +- ``mlflow-credentials`` secret must exist (run scripts/setup_mlflow_credentials.py). +- The MLflow model ``mobile-price-svm-`` version 1 must be registered + (run mlflow/mobile-price-classification/mlflow-mobile-price-classification.ipynb first). +""" + +from __future__ import annotations + +import base64 +import json +import os +import subprocess +import sys + +_ISVC_NAME = "mobile-price-svm" +_YAML_TEMPLATE = os.path.join(os.path.dirname(__file__), "InferenceService.yaml") +_SA_YAML_TEMPLATE = os.path.join(os.path.dirname(__file__), "ServiceAccount.yaml") + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _mlflow_username(namespace: str) -> str: + result = subprocess.run( + [ + "kubectl", + "get", + "secret", + "mlflow-credentials", + "-n", + namespace, + "-o", + "json", + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + "mlflow-credentials secret not found.\n" + "Run scripts/setup_mlflow_credentials.py to create it." + ) + data = json.loads(result.stdout)["data"] + full = base64.b64decode(data["MLFLOW_TRACKING_USERNAME"]).decode() + # Pipeline registers models as "mobile-price-svm-{local_user}" where + # local_user = MLFLOW_TRACKING_USERNAME.split('@')[0], e.g. + # "developer1@prokube.cloud" → "developer1" + return full.split("@")[0] + + +def _kubectl_apply(manifest: str, namespace: str) -> None: + result = subprocess.run( + ["kubectl", "apply", "-n", namespace, "-f", "-"], + input=manifest, + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError(f"kubectl apply failed:\n{result.stderr}") + + +def _wait_ready(name: str, namespace: str, timeout: int = 600) -> None: + print( + f"Waiting for InferenceService '{name}' to become ready (timeout {timeout}s)..." + ) + result = subprocess.run( + [ + "kubectl", + "wait", + "inferenceservice", + name, + "-n", + namespace, + "--for=condition=Ready", + f"--timeout={timeout}s", + ], + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + f"InferenceService '{name}' did not become ready:\n{result.stderr}" + ) + print(f" '{name}' is Ready.") + + +def _isvc_url(name: str, namespace: str) -> str: + return subprocess.check_output( + [ + "kubectl", + "get", + "inferenceservice", + name, + "-n", + namespace, + "-o", + "jsonpath={.status.url}", + ], + text=True, + ).strip() + + +def deploy(timeout: int = 600) -> str: + """Apply ServiceAccount.yaml and InferenceService.yaml, wait for readiness, return the external URL.""" + ns = _namespace() + username = _mlflow_username(ns) + + with open(_SA_YAML_TEMPLATE) as fh: + sa_manifest = fh.read() + + _kubectl_apply(sa_manifest, ns) + print(f"Applied ServiceAccount 'mlflow-isvc-sa' in namespace '{ns}'.") + + with open(_YAML_TEMPLATE) as fh: + manifest = fh.read() + + manifest = ( + manifest.replace("", _ISVC_NAME) + .replace("", ns) + .replace("", username) + ) + + _kubectl_apply(manifest, ns) + print(f"Applied InferenceService '{_ISVC_NAME}' in namespace '{ns}'.") + + _wait_ready(_ISVC_NAME, ns, timeout=timeout) + url = _isvc_url(_ISVC_NAME, ns) + print(f"ISVC_URI={url}") + return url + + +_HERE = os.path.dirname(os.path.abspath(__file__)) +_TEST_SCRIPT = os.path.join(_HERE, "test_inference_service.py") +_TEST_JSON = os.path.join(_HERE, "v2-mlflow-inference-body.json") + + +def test(uri: str) -> None: + """Run the inference smoke test against the deployed ISVC.""" + # Get / create the API key + _root = subprocess.check_output( + ["git", "rev-parse", "--show-toplevel"], text=True + ).strip() + sys.path.insert(0, os.path.join(_root, "scripts")) + from get_or_create_api_key import get_or_create_api_key # noqa: PLC0415 + + api_key = get_or_create_api_key() + + print(f"Running inference smoke test against {uri} ...") + result = subprocess.run( + [sys.executable, _TEST_SCRIPT, "--json", _TEST_JSON, "--model", _ISVC_NAME], + capture_output=True, + text=True, + env={**os.environ, "API_KEY": api_key, "INFERENCE_SERVICE_URI": uri}, + ) + if result.returncode != 0: + raise RuntimeError(f"Inference test failed:\n{result.stdout}\n{result.stderr}") + print("Inference test passed.") + print(result.stdout.strip()) + + +def deploy_and_test(timeout: int = 600) -> str: + """Deploy the ISVC, wait for readiness, run smoke test, return the URL.""" + uri = deploy(timeout=timeout) + test(uri) + return uri + + +if __name__ == "__main__": + try: + deploy_and_test() + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + sys.exit(1) diff --git a/serving/mlflow-kserve-minimal/cleanup.py b/serving/mlflow-kserve-minimal/cleanup.py new file mode 100644 index 0000000..96c341c --- /dev/null +++ b/serving/mlflow-kserve-minimal/cleanup.py @@ -0,0 +1,49 @@ +""" +Cleanup script for the mlflow-kserve-minimal example. + +Deletes (idempotently): + - InferenceService ``mobile-price-svm`` + +Usage +----- + python cleanup.py [--dry-run] +""" + +from __future__ import annotations + +import argparse +import subprocess +import sys + + +def _namespace() -> str: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as fh: + return fh.read().strip() + + +def _kubectl_delete(*args: str, dry_run: bool = False) -> None: + cmd = ["kubectl", "delete", *args, "--ignore-not-found"] + if dry_run: + print(f"[dry-run] {' '.join(cmd)}") + return + result = subprocess.run(cmd, capture_output=True, text=True) + if result.returncode != 0: + print(f"WARNING: {result.stderr.strip()}", file=sys.stderr) + else: + print(result.stdout.strip() or f"deleted (or not found): {' '.join(args)}") + + +def cleanup(dry_run: bool = False) -> None: + ns = _namespace() + _kubectl_delete("inferenceservice", "mobile-price-svm", "-n", ns, dry_run=dry_run) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--dry-run", action="store_true", help="Print commands without executing them" + ) + args = parser.parse_args() + cleanup(dry_run=args.dry_run)