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..4d54463 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ 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 @@ -14,8 +15,17 @@ For full platform documentation, see [docs.prokube.ai](https://docs.prokube.ai/) ├── pipelines # Kubeflow Pipelines examples ├── rstudio # RStudio examples ├── serving # model serving examples (KServe, vLLM, shadow deployments) +└── src/pk_helpers # installable prokube helpers (credential setup, API key, KServe URLs) ``` +The shared `pk_helpers` package is installable from the repo root +(`pip install -e .`); notebooks install it via a `%pip install -q -e ~/kubeflow-examples` +setup cell and import helpers with `from pk_helpers import ...`. + +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 +53,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..03f438e --- /dev/null +++ b/ci/README.md @@ -0,0 +1,233 @@ +# 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 +and API-key skipping, env-mutating ordering — 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 + api_key_dependent=False, # True = skip when INFERENCE_SERVICE_API_KEY is unset + env_mutating=False, # True = pip installs/upgrades packages; + # runs before the rest of its phase (see below) +) +``` + +### env_mutating flag + +Every notebook in a phase is executed via papermill against the **same** +`python3` kernel/site-packages — there is no per-notebook virtualenv. If a +notebook runs `pip install --upgrade ` while another notebook in the +same phase is concurrently importing that package, the concurrent +reinstall can corrupt the import (e.g. `ModuleNotFoundError: +No module named 'pandas._libs.internals'` from a partially-replaced +compiled extension). + +Set `env_mutating=True` on any example whose steps run `pip install` / +`%pip install` (uncommented, not `-q`-only-metadata, actually mutating +installed packages) against packages that other examples in the same +phase might import. `run_all.py` runs all `env_mutating` examples in a +phase to completion **before** starting the rest of that phase, instead of +throwing everything into the same parallel batch. Prefer avoiding +`pip install --upgrade` in new examples entirely (pin/bake deps into the +notebook image) — reach for `env_mutating=True` only when that isn't +possible. + +### 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. + +--- + +## pk_helpers package + +`pk_helpers` (source in `src/pk_helpers/`) bundles the prokube-platform +utilities importable from notebooks and apply scripts. It is a regular, +installable Python package — install it once, editable, from the repo root: + +```bash +pip install -e . +``` + +In notebooks, do this from a setup cell near the top: + +```python +%pip install -q -e ~/kubeflow-examples +``` + +CI installs it automatically in the preflight step (`_ensure_pk_helpers`), +so `apply.py` scripts can `from pk_helpers import ...` without any path +wiring. Every helper works identically whether called from a notebook cell +or from an `apply.py`. + +### setup_mlflow_credentials + +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 via the installed console script: + +```bash +pk-setup-mlflow-credentials +``` + +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 + +Returns a model-serving API key: the `INFERENCE_SERVICE_API_KEY` env var if +an admin has injected one into the pod, otherwise an interactive prompt +(ask your cluster admin, or use pkui if available on your platform). Call it +from any notebook cell or script that needs an inference API key: + +```python +from pk_helpers import get_or_create_api_key + +API_KEY = get_or_create_api_key() +``` + +CI runs headlessly and cannot answer the interactive prompt, so +`INFERENCE_SERVICE_API_KEY` **must** be exported before running +`ci/run_all.py`: + +```bash +export INFERENCE_SERVICE_API_KEY= # ask your admin, or use pkui if available +``` + +CI validates this in the preflight check and skips `api_key_dependent` +examples if it is unset — mark any new example that calls +`get_or_create_api_key()` with `api_key_dependent=True`. + +### internal_predict_url + +Returns the correct **internal, in-cluster** predict URL for a KServe +InferenceService, compatible with both prokube generations: + +- Currently released: hits the predictor Service directly — + `http://-predictor..svc.cluster.local/v1/models/:predict`. +- Upcoming (agentgateway-based, not yet released): routes through the + shared `agentgateway-proxy` Service instead — + `http://agentgateway-proxy.agentgateway-system.svc.cluster.local/_platform/serving///v2/models//infer`. + +It picks the URL via a plain DNS lookup for the `agentgateway-proxy` Service +(cached for the process) — no config flag needed, and no RBAC required +(unlike `kubectl get service`, which notebook pod service accounts +typically can't do cross-namespace; DNS resolution needs no permissions, +and a non-existent Service just fails to resolve). The request/response +payload format is unchanged either way (plain KServe V1 JSON, e.g. +`{"instances": [...]}`) — only the URL differs. Any example that predicts +against an InferenceService via its internal cluster URL (not the external +gateway URL) should use this instead of hardcoding the `-predictor` +pattern: + +```python +from pk_helpers import internal_predict_url + +url = internal_predict_url(isvc_name, namespace, model_name) +``` + +--- + +## 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..8a818c9 --- /dev/null +++ b/ci/run_all.py @@ -0,0 +1,1019 @@ +"""Run registered examples by phase, poll KFP runs, and always clean up.""" + +from __future__ import annotations + +import argparse +import json +import os +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.") + + +def _ensure_pk_helpers() -> None: + """Editable-install the pk_helpers package if it is not importable. + + Notebooks and serving apply.py scripts import shared prokube helpers via + ``from pk_helpers import ...``. Installing the repo editable makes the same + interpreter used by the CI subprocesses able to resolve those imports. + """ + try: + import pk_helpers # noqa: F401 + except ImportError: + print("pk_helpers not found — installing repo editable...") + subprocess.run( + [sys.executable, "-m", "pip", "install", "-e", str(_REPO_ROOT)], + check=True, + ) + print("pk_helpers 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 + ) + api_key_dependent: bool = ( + False # skip automatically when INFERENCE_SERVICE_API_KEY is unset + ) + env_mutating: bool = False # run first; modifies the shared Python environment + + +# ───────────────────────────────────────────────────────────────────────────── +# 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", + env_mutating=True, + ), + 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", + api_key_dependent=True, + ), + 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, + api_key_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, + api_key_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)" + if ex.api_key_dependent: + label += " (api-key)" + if ex.env_mutating: + label += " (env-mutating, runs first in its phase)" + 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 `pk-setup-mlflow-credentials` 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 `pk-setup-mlflow-credentials`", + ) + + 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 `pk-setup-mlflow-credentials`" + ) + 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 _check_kserve_api_key() -> tuple[bool, str]: + """Check that headless CI has an inference API key.""" + if os.environ.get("INFERENCE_SERVICE_API_KEY", "").strip(): + return True, "OK" + return False, ( + "INFERENCE_SERVICE_API_KEY is unset — export a model-serving API key " + "(ask your cluster admin, or use pkui if available) before running CI" + ) + + +def _preflight(results: dict[str, Result]) -> bool: + """Install papermill and validate MLflow credentials and the kserve API key. + + MLflow-dependent and API-key-dependent examples are recorded as SKIP in + `results` on failure. Returns True if MLflow credentials are valid. + """ + _ensure_papermill() + _ensure_pk_helpers() + + print("Pre-flight: checking MLflow credentials...") + mlflow_ok, mlflow_reason = _check_mlflow_credentials() + if mlflow_ok: + print(" [OK] MLflow credentials valid") + else: + print(f" [SKIP] {mlflow_reason}") + for ex in _EXAMPLES: + if ex.mlflow_dependent: + results[ex.name] = Result( + name=ex.name, status="SKIP", error=mlflow_reason + ) + + print("Pre-flight: checking kserve API key...") + api_key_ok, api_key_reason = _check_kserve_api_key() + if api_key_ok: + print(" [OK] INFERENCE_SERVICE_API_KEY is set") + else: + print(f" [SKIP] {api_key_reason}") + for ex in _EXAMPLES: + if ex.api_key_dependent and ex.name not in results: + results[ex.name] = Result( + name=ex.name, status="SKIP", error=api_key_reason + ) + + return mlflow_ok + + +# ── Phases ──────────────────────────────────────────────────────────────────── + + +def _run_phase( + ctx: _Context, + phase: int, + opts: dict[str, bool], + predicate: Callable[[Example], bool] | None = None, +) -> dict[str, Future]: + """Submit all examples for a given phase; return {name: Future}. + + If `predicate` is given, only examples for which it returns True are + submitted (used to run env_mutating examples in their own sub-step). + """ + futures: dict[str, Future] = {} + for ex in _EXAMPLES: + if ex.phase != phase: + continue + if predicate is not None and not predicate(ex): + 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 1a: running environment-mutating examples " + "(pip install into the shared kernel env) sequentially first..." + ) + _drain(ctx, _run_phase(ctx, 1, opts, predicate=lambda ex: ex.env_mutating)) + + print("Phase 1: running independent examples in parallel...") + _drain( + ctx, + _run_phase(ctx, 1, opts, predicate=lambda ex: not ex.env_mutating), + ) + + 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..9e8aab2 --- /dev/null +++ b/hparam-tuning/minimal-mnist/cleanup.py @@ -0,0 +1,43 @@ +"""Delete the minimal MNIST Katib experiment and its trial pods.""" + +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..058a001 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 `pk-setup-mlflow-credentials` once to create the required secret:\n", + "```bash\n", + "# From a JupyterLab terminal:\n", + "pk-setup-mlflow-credentials\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 `pk-setup-mlflow-credentials` 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 `pk-setup-mlflow-credentials` 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..58d43b1 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 `pk-setup-mlflow-credentials` 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", + "pk-setup-mlflow-credentials\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..6b56716 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 `pk-setup-mlflow-credentials` once to create the required secret:\n", + "```bash\n", + "# From a JupyterLab terminal:\n", + "pk-setup-mlflow-credentials\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 `pk-setup-mlflow-credentials` 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", + "pk-setup-mlflow-credentials\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..9bff01a --- /dev/null +++ b/notebooks/dask/cleanup.py @@ -0,0 +1,42 @@ +"""Delete the Dask cluster created by the example.""" + +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..6aec986 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,10 @@ }, "outputs": [], "source": [ - "!pip install --upgrade ipywidgets dask-kubernetes==2024.9.0 pandas \"dask[complete]\"==2024.9.1" + "# pandas is a transitive dependency of dask[complete] already, no need to\n", + "# force-upgrade it separately (that was needlessly bumping pandas to the\n", + "# latest release and mutating the shared notebook kernel env).\n", + "!pip install --upgrade ipywidgets \"dask-kubernetes==2026.3.0\" \"dask[complete]==2026.3.0\" \"anyio>=4.12\"\n" ] }, { @@ -120,7 +123,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..27a0664 100644 --- a/notebooks/mnist-vae/run_training.py +++ b/notebooks/mnist-vae/run_training.py @@ -1,3 +1,39 @@ +import os +import subprocess +import sys + +# Re-exec with Conda's libstdc++ so SciPy can resolve its CXXABI symbols. +_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,17 +41,17 @@ 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: - """ - 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. - """ +@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." +) +@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 on MNIST.""" # Initialize data module dm = MNISTDataModule(data_path="./data", num_workers=0, batch_size=32) @@ -23,8 +59,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 +69,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..f0acbb5 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) @@ -167,18 +179,7 @@ def mobile_price_classification_pipeline( scatter_plot_column_y: str = "battery_power", seed: int = 42, ): - """ - Mobile price classification pipeline using containerized components. - - This pipeline covers the following steps: - 1. Read data from the specified paths. - 2. Split the data into training and validation sets. - 3. Fit the MinMax scaler. - 4. Tune hyperparameters for the SVM model. - 5. Train the SVM model with the best hyperparameters. - 6. Evaluate the trained model. - 7. Test the model and visualize the results with a scatter plot. - """ + """Train, tune, evaluate, and visualize a mobile-price classifier.""" from kfp import kubernetes # Step 1: Read the data @@ -187,16 +188,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/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..7d07dd1 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,18 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "pk-helpers" +version = "0.1.0" +description = "Prokube platform helpers shared across the Kubeflow examples." +readme = "README.md" +requires-python = ">=3.10" +dependencies = [] + +[project.scripts] +pk-get-api-key = "pk_helpers.api_key:main" +pk-setup-mlflow-credentials = "pk_helpers.mlflow_credentials:main" + +[tool.hatch.build.targets.wheel] +packages = ["src/pk_helpers"] diff --git a/serving/hf-vllm-completion/apply.py b/serving/hf-vllm-completion/apply.py new file mode 100644 index 0000000..f12b4d4 --- /dev/null +++ b/serving/hf-vllm-completion/apply.py @@ -0,0 +1,119 @@ +"""Deploy and smoke-test the DistilBERT InferenceService.""" + +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.""" + from pk_helpers import internal_predict_url + + url = internal_predict_url(_ISVC_NAME, namespace, _MODEL_NAME) + 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..4760584 --- /dev/null +++ b/serving/hf-vllm-completion/cleanup.py @@ -0,0 +1,42 @@ +"""Delete the DistilBERT InferenceService.""" + +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..30070e0 --- /dev/null +++ b/serving/kserve-keda-autoscaling/apply.py @@ -0,0 +1,187 @@ +"""Deploy and verify KEDA autoscaling, skipping when KEDA is unavailable.""" + +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, returning an error only when KEDA is absent.""" + 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..ef45978 --- /dev/null +++ b/serving/kserve-keda-autoscaling/cleanup.py @@ -0,0 +1,43 @@ +"""Delete the KEDA ScaledObject and InferenceService.""" + +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..8cc28f2 --- /dev/null +++ b/serving/minimal-example-shadow-deployment/apply.py @@ -0,0 +1,296 @@ +"""Deploy Postgres and the doubler/tripler services, then test the primary.""" + +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]. + """ + from pk_helpers import internal_predict_url + + url = internal_predict_url(_DOUBLER_ISVC, namespace, "model") + 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..c276ff5 --- /dev/null +++ b/serving/minimal-example-shadow-deployment/cleanup.py @@ -0,0 +1,48 @@ +"""Delete the shadow example's InferenceServices and Postgres cluster.""" + +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..52177f6 --- /dev/null +++ b/serving/minimal-s3-model/cleanup.py @@ -0,0 +1,43 @@ +"""Delete the object-storage InferenceService.""" + +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", "kserve-object-storage-test", "-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/minimal-s3-model/minimal-s3-model.ipynb b/serving/minimal-s3-model/minimal-s3-model.ipynb index 6d1b063..64ea38f 100644 --- a/serving/minimal-s3-model/minimal-s3-model.ipynb +++ b/serving/minimal-s3-model/minimal-s3-model.ipynb @@ -16,7 +16,11 @@ "outputs": [], "source": [ "# sklearn, joblib, s3fs already come with the notebook image\n", - "# %pip install sklearn joblib s3fs" + "# %pip install sklearn joblib s3fs\n", + "\n", + "# Install the prokube platform helpers (pk_helpers) shared by these examples.\n", + "# The kubeflow-examples repo is cloned into $HOME at notebook startup.\n", + "%pip install -q -e ~/kubeflow-examples\n" ] }, { @@ -274,14 +278,32 @@ "metadata": {}, "outputs": [], "source": [ + "import time\n", "import requests\n", - "isvc_internal_url = f\"{inference_service_name}-predictor.{namespace}\" # KServe internal endpoint is always built like this\n", - "response = requests.post(\n", - " f\"http://{isvc_internal_url}/v1/models/{inference_service_name}:predict\",\n", - " # an iris instance is [sepal_length, sepal_width, petal_length, petal_width]\n", - " json={\"instances\": [[6.8, 2.8, 4.8, 1.4], [5.1, 3.5, 1.4, 0.2]]}\n", - ")\n", - "print(response.json())" + "from pk_helpers import internal_predict_url\n", + "\n", + "# Resolves to the plain predictor Service URL on the currently-released\n", + "# prokube platform, or to the agentgateway-proxy URL on newer platforms\n", + "# where all serving traffic (internal and external) is routed through it.\n", + "predict_url = internal_predict_url(inference_service_name, namespace, inference_service_name)\n", + "# an iris instance is [sepal_length, sepal_width, petal_length, petal_width]\n", + "payload = {\"instances\": [[6.8, 2.8, 4.8, 1.4], [5.1, 3.5, 1.4, 0.2]]}\n", + "\n", + "# `kubectl wait --for=condition=ready` above only checks the\n", + "# InferenceService CR; the predictor Pod can take a few more seconds to\n", + "# register as a ready Service endpoint, so retry briefly instead of\n", + "# failing on the very first request.\n", + "for attempt in range(6):\n", + " try:\n", + " response = requests.post(predict_url, json=payload, timeout=10)\n", + " response.raise_for_status()\n", + " print(response.json())\n", + " break\n", + " except Exception as exc:\n", + " if attempt == 5:\n", + " raise\n", + " print(f\" attempt {attempt + 1}/6 failed ({exc}); retrying in 5s...\")\n", + " time.sleep(5)" ] }, { @@ -297,7 +319,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 prompts for your model-serving API key (or reads it from the INFERENCE_SERVICE_API_KEY env var if your admin already injected one). Ask your cluster administrator for a key, or use pkui if it is available on your platform.\n", + "

\n", + " If you'd rather set it directly, replace the cell contents with:\n", + " 
INFERENCE_SERVICE_API_KEY = \"your-key-here\"
\n", + "
\n", + "
" ] }, { @@ -307,9 +343,9 @@ "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\")" + "from pk_helpers import get_or_create_api_key\n", + "\n", + "INFERENCE_SERVICE_API_KEY = get_or_create_api_key()" ] }, { diff --git a/serving/mlflow-kserve-inference-protocols/README.md b/serving/mlflow-kserve-inference-protocols/README.md new file mode 100644 index 0000000..e1ccb14 --- /dev/null +++ b/serving/mlflow-kserve-inference-protocols/README.md @@ -0,0 +1,117 @@ +# 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) need a model-serving API +> key. Ask your cluster administrator for one, or use pkui if it is +> available on your platform. Set `API_KEY` in the environment before +> calling `apply.py`, or you'll be prompted for it interactively. + +## 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..0a069f4 --- /dev/null +++ b/serving/mlflow-kserve-inference-protocols/apply.py @@ -0,0 +1,195 @@ +"""Deploy and smoke-test the v1 and v2 MLflow KServe services.""" + +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 `pk-setup-mlflow-credentials` 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: + from pk_helpers 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..9bf653b --- /dev/null +++ b/serving/mlflow-kserve-inference-protocols/cleanup.py @@ -0,0 +1,46 @@ +"""Delete the v1 and v2 MLflow InferenceServices.""" + +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..6217534 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 `pk-setup-mlflow-credentials` 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..3889eaf 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` needs a model-serving API key. Ask your cluster administrator +> for one, or use pkui if it is available on your platform, then set it +> before running the test script: +> ```sh +> export API_KEY=your-key-here +> ``` +> If `API_KEY` is not set, `apply.py` will prompt for it interactively. + 1. Set the required environment variables (optional): ```sh - export API_KEY= + export API_KEY= # ask your admin, or use pkui if available 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..e832445 --- /dev/null +++ b/serving/mlflow-kserve-minimal/apply.py @@ -0,0 +1,165 @@ +"""Deploy and smoke-test the MLflow-backed InferenceService.""" + +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 `pk-setup-mlflow-credentials` to create it." + ) + data = json.loads(result.stdout)["data"] + full = base64.b64decode(data["MLFLOW_TRACKING_USERNAME"]).decode() + # Model names use the username portion before "@". + 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 + from pk_helpers import get_or_create_api_key + + 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..3f8ea59 --- /dev/null +++ b/serving/mlflow-kserve-minimal/cleanup.py @@ -0,0 +1,40 @@ +"""Delete the minimal MLflow InferenceService.""" + +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) diff --git a/src/pk_helpers/__init__.py b/src/pk_helpers/__init__.py new file mode 100644 index 0000000..43379ea --- /dev/null +++ b/src/pk_helpers/__init__.py @@ -0,0 +1,26 @@ +"""Prokube platform helpers shared across the Kubeflow examples. + +These utilities abstract away logic specific to the prokube platform so that +notebooks and ``apply.py`` scripts can import them directly instead of wiring +up ``sys.path`` by hand. + +Install once (editable) from the repo root:: + + pip install -e . + +Then import what you need:: + + from pk_helpers import get_or_create_api_key, internal_predict_url +""" + +from __future__ import annotations + +from pk_helpers.api_key import get_or_create_api_key +from pk_helpers.kserve_url import internal_predict_url +from pk_helpers.mlflow_credentials import setup_mlflow_credentials + +__all__ = [ + "get_or_create_api_key", + "internal_predict_url", + "setup_mlflow_credentials", +] diff --git a/src/pk_helpers/api_key.py b/src/pk_helpers/api_key.py new file mode 100644 index 0000000..e819e27 --- /dev/null +++ b/src/pk_helpers/api_key.py @@ -0,0 +1,52 @@ +"""Read an inference API key from the environment or prompt for one.""" + +from __future__ import annotations + +import argparse +import os +import sys +from getpass import getpass + +_API_KEY_ENV_VAR = "INFERENCE_SERVICE_API_KEY" + + +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() -> str: + """Return the configured API key, prompting if necessary.""" + env_key = _key_from_env() + if env_key: + return env_key + + key = getpass( + f"${_API_KEY_ENV_VAR} is unset. Please enter your model-serving API " + "key (ask your cluster admin, or use pkui if available on your " + "platform): " + ).strip() + if not key: + raise RuntimeError( + f"No API key available: ${_API_KEY_ENV_VAR} is unset and no key " + "was entered." + ) + return key + + +def main() -> None: + """Console-script entry point: print the resolved API key.""" + try: + argparser = argparse.ArgumentParser( + description="Get a prokube model-serving API key" + ) + argparser.parse_args() + print(get_or_create_api_key()) + except Exception as exc: # noqa: BLE001 + print(f"ERROR: {exc}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/src/pk_helpers/kserve_url.py b/src/pk_helpers/kserve_url.py new file mode 100644 index 0000000..eb21e7f --- /dev/null +++ b/src/pk_helpers/kserve_url.py @@ -0,0 +1,33 @@ +"""Build an internal KServe prediction URL for the available routing mode.""" + +from __future__ import annotations + +import socket +from functools import lru_cache + +_AGENTGATEWAY_NAMESPACE = "agentgateway-system" +_AGENTGATEWAY_SERVICE = "agentgateway-proxy" +_AGENTGATEWAY_HOST = f"{_AGENTGATEWAY_SERVICE}.{_AGENTGATEWAY_NAMESPACE}.svc.cluster.local" + + +@lru_cache(maxsize=1) +def _agentgateway_available() -> bool: + """Return whether agentgateway resolves through cluster DNS.""" + try: + socket.gethostbyname(_AGENTGATEWAY_HOST) + return True + except socket.gaierror: + return False + + +def internal_predict_url(isvc_name: str, namespace: str, model_name: str) -> str: + """Return the prediction URL for the detected in-cluster route.""" + if _agentgateway_available(): + return ( + f"http://{_AGENTGATEWAY_HOST}" + f"/_platform/serving/{namespace}/{isvc_name}/v2/models/{model_name}/infer" + ) + return ( + f"http://{isvc_name}-predictor.{namespace}.svc.cluster.local" + f"/v1/models/{model_name}:predict" + ) diff --git a/src/pk_helpers/mlflow_credentials.py b/src/pk_helpers/mlflow_credentials.py new file mode 100644 index 0000000..a6c1a10 --- /dev/null +++ b/src/pk_helpers/mlflow_credentials.py @@ -0,0 +1,100 @@ +"""Create or update the namespace's MLflow credentials Secret.""" + +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}'.") + + +def main() -> None: + """Console-script entry point: create/update the MLflow credentials secret.""" + 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) + + +if __name__ == "__main__": + main()