diff --git a/apis/modelcaches/definition.yaml b/apis/modelcaches/definition.yaml index 5ba469954..24637a625 100644 --- a/apis/modelcaches/definition.yaml +++ b/apis/modelcaches/definition.yaml @@ -117,3 +117,61 @@ spec: enum: [Pending, Hydrating, Ready, Failed] message: type: string + description: >- + Why the model isn't readable on this cluster yet, + while phase isn't Ready. + mount: + type: object + description: >- + How an engine pod reads the model on this cluster. + Set only while phase is Ready. A ModelReplica adds + these to each of its engine pods. + properties: + volumes: + type: array + description: >- + Pod volumes, each a subset of a Kubernetes + Volume. + maxItems: 8 + items: + type: object + required: [name] + properties: + name: + type: string + persistentVolumeClaim: + type: object + required: [claimName] + properties: + claimName: + type: string + volumeMounts: + type: array + description: >- + Container volume mounts, each a subset of a + Kubernetes VolumeMount. + maxItems: 8 + items: + type: object + required: [name, mountPath] + properties: + name: + type: string + mountPath: + type: string + readOnly: + type: boolean + env: + type: array + description: >- + Container environment variables. A variable the + engine sets itself takes precedence. + maxItems: 8 + items: + type: object + required: [name, value] + properties: + name: + type: string + value: + type: string diff --git a/design/modelcache-mount-contract.md b/design/modelcache-mount-contract.md new file mode 100644 index 000000000..d40c62c08 --- /dev/null +++ b/design/modelcache-mount-contract.md @@ -0,0 +1,142 @@ +# ModelCache mount contract + +**Status:** Draft +**Date:** October 2026 +**Author:** Dennis Ramdass + +This document proposes that each `ModelCache` say how to read its model on each +cluster, and that a `ModelReplica` wait until it does. It builds on +[modelcache.md](./modelcache.md). + +## Background + +A `ModelCache` stages a HuggingFace model onto a ReadWriteMany claim on each +cluster it matches, with a hydration Job. A `ModelDeployment` that references +the cache mounts that claim into its engine pods at `/mnt/models`. + +`compose-model-replica` works out the claim's name itself, with a copy of the +code `compose-model-cache` uses to name it. It mounts the claim as soon as the +scheduler places the replica, and it can't tell whether the model is there yet. + +This caused these faults: + +- An engine started while its 464 GB cache was still staging. It found no + model on the claim, and downloaded the whole model again in parallel with the + Job ([#495](https://github.com/modelplaneai/modelplane/issues/495)). +- A claim waited on a node the cloud had replaced. The cache said `Pending` + for 40 minutes, and nothing on the control plane said why + ([#494](https://github.com/modelplaneai/modelplane/issues/494)). + +## Goals + +- Give each cluster one answer about how to read the model there, which a + replica copies instead of computing. +- Don't start an engine on a cluster until the model is readable there. +- Say why a cluster can't read the model yet. +- Keep a serving replica's engines running when its cache changes. + +It's not a goal to add sources other than HuggingFace. That's +[modelcache-sources.md](https://github.com/modelplaneai/modelplane/pull/505). + +## Proposal + +### The mount + +Each entry in `status.clusters[]` gets a `mount` field, set while the entry's +phase is `Ready`. Each item uses some of the fields of the Kubernetes type in +its comment: + +```yaml +status: + clusters: + - name: cluster-a + phase: Ready + mount: + volumes: # corev1.Volume: name, persistentVolumeClaim + - name: model-cache + persistentVolumeClaim: + claimName: modelcache-ml-team-qwen-17db2 + volumeMounts: # corev1.VolumeMount: name, mountPath, readOnly + - name: model-cache + mountPath: /mnt/models + env: # corev1.EnvVar: name, value + - name: HF_HUB_CACHE + value: /mnt/models +``` + +`compose-model-cache` knows the source and the claim, so it writes the mount. +For a `HuggingFace` source it's the claim at `/mnt/models`, and `HF_HUB_CACHE` +pointing there. An engine then finds the staged snapshot by its repo ID. These +match the volume, mount and variable that `compose-model-replica` composes +today, so running engines don't restart when you upgrade Modelplane. + +### Waiting for the mount + +`compose-model-replica` requires the `ModelCache` and reads the entry for its +own cluster. It adds the mount's volumes to each engine pod, and its volume +mounts and variables to each engine container. A variable the engine sets +itself comes later in the list, so it takes effect. + +Until the entry has a mount, a new replica composes nothing. Its +`ModelAccepted` condition is `False`, with reason `WaitingForCache` and a +message that says what it waits for. The message comes from the cache's entry, +or says the cache doesn't stage to the replica's cluster, or that the cache +doesn't exist. + +A replica whose engines already run doesn't wait. Until its entry has a mount, +it keeps the cache volume, volume mount and `HF_HUB_CACHE` its workload has +now, or no cache if it has none, and warns. It reads them from the workload it +composed last time, and its engines don't restart when the claim disappears, +when someone deletes the cache, or when someone adds or switches to a cache +that isn't ready. Stopping a serving model because its cache changed would +turn a storage problem into an outage, and moving it onto a cache that's still +downloading would start it without its model. + +### Messages + +`compose-model-cache` writes a `message` for each entry that isn't `Ready`: + +| Phase | Message gives | +|---|---| +| `Pending` | The claim it waits on, and the node the scheduler chose for it, from the claim's `volume.kubernetes.io/selected-node` annotation. | +| Any phase but `Ready`, with no token | The Secret and key the hydration Job needs. | +| `Hydrating` | The repository and the claim it downloads into. | +| `Failed` | The Job, how many attempts failed, and the Job's own message. | + +## Future improvements + +**Other sources.** A source that reads the model from a registry or from a +claim the user filled writes its own mount. Replicas copy whatever the cache +publishes, so they don't need to change for it. + +**Several caches per deployment.** +[#281](https://github.com/modelplaneai/modelplane/issues/281) asks for a draft +model or tokenizer from a second cache. Each cache would then need its own +volume name and its own path. The mount's shape already allows that, because +replicas copy whatever the cache publishes. + +## Alternatives considered + +### Compute the mount in the replica + +The replica could keep naming the claim and check the cache's phase. That keeps +two copies of the naming code in step, and every new source would add a branch +to both functions. + +### A child resource for each cluster + +[#210](https://github.com/modelplaneai/modelplane/issues/210) asks for a +`ModelCacheHydration` child for each cluster. The mount needs one place for each +cluster, and `status.clusters[]` already gives it. + +### Hold the replica's last mount in its own status + +A running replica could keep the mount it last used in its own status, and +reuse it when the cache stops publishing one. The workload the replica composed +already holds that mount, and a second copy in status would have to match it. + +### Derive the mount from the replica's reference + +A running replica could derive the mount a `HuggingFace` cache would publish +from its own `modelCacheRef`. That restarts the engines onto a cache that isn't +ready when someone adds a cache or switches to another one. diff --git a/docs/content/models/model-cache.md b/docs/content/models/model-cache.md index 4b6d5e4d3..c7a2a7d30 100644 --- a/docs/content/models/model-cache.md +++ b/docs/content/models/model-cache.md @@ -38,6 +38,29 @@ default branch, which finds a cache staged without a `revision` or with revision too (`--revision` for vLLM). If the engine uses the default branch instead, it finds nothing staged there and downloads the model a second time. +## Waiting for the cache + +A replica doesn't start its engines until the cache is ready on its cluster. +Until then the `ModelReplica` reports `ModelAccepted` as `False` with reason +`WaitingForCache`, and a message from the cache that says what it's waiting +for. The `ModelCache` gives the same message for each cluster: + +```bash +kubectl get modelcache qwen -n ml-team -o jsonpath='{.status.clusters}' +``` + +The message gives the claim that hasn't bound and the node it's waiting on, +the repository the hydration Job is downloading (`Downloading into +PersistentVolumeClaim `), or the Job that failed and why. A replica checks +its cache about once a minute. Its engines start up to about a minute after the +cache turns `Ready`. + +A replica whose engines already run doesn't wait. Until its cache is ready on +its cluster, the engines keep the cache they mount now, or run without one if +they mount none. This holds when the cache stops being ready, when someone +deletes it, and when you add a cache to the deployment or switch it to another +one. The engines restart onto the new cache once it's ready. + ## Authenticating A gated or private model needs a credential to fetch. When a cache stages the diff --git a/e2e/tests/test_model_cache.py b/e2e/tests/test_model_cache.py new file mode 100644 index 000000000..c51bfb7b7 --- /dev/null +++ b/e2e/tests/test_model_cache.py @@ -0,0 +1,301 @@ +# Copyright 2026 The Modelplane Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""A replica waits for its ModelCache, then mounts it the way the cache says to. + +The workload cluster has no ReadWriteMany storage, so the test brings a +StorageClass with no provisioner, and binds the cache's claim to a hostPath +volume it creates later. Until then the claim is Pending, and the replica +must wait. The cache stages a tiny public repository. + +The local pool has one fake GPU, which mock-demo holds, so the test scales +mock-demo to zero while it runs and back afterwards. Run it on its own, after +the rest of the suite: E2E_MODEL_CACHE=1 pytest e2e/tests/test_model_cache.py. +""" + +import collections.abc +import os +import typing +import uuid + +import pytest +from kubernetes import client + +from e2e import kube, wait + +pytestmark = pytest.mark.skipif( + os.environ.get("E2E_MODEL_CACHE") != "1", reason="scales mock-demo to zero; set E2E_MODEL_CACHE=1 to run it" +) + +NAMESPACE = "ml-team" +CLUSTER = "local" +CACHE = "tiny-gpt2" +DEPLOYMENT = "cached-mock" +REPO = "hf-internal-testing/tiny-random-gpt2" +STORAGE_CLASS = "e2e-model-cache" +VOLUME = "e2e-model-cache" +# Deleting the volume leaves its hostPath data on the node, and a cache that +# finds a previous run's completion marker skips its download. A path of its +# own per run makes every run download the repository again. +HOST_PATH = f"/tmp/e2e-model-cache-{uuid.uuid4().hex[:8]}" +# Where the serving pods run on the workload cluster: the namespace mirroring +# ml-team. +REMOTE_NAMESPACE = "mp-ml-team-51733" +GROUP = "modelplane.ai" +# python:3.12-alpine pinned by digest, as the mock-demo engine. +IMAGE = "python@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" + +# The mock engine reports healthy only once the cache's snapshot of the repo +# is readable through HF_HUB_CACHE, so the replica can't go Ready on an empty +# or missing mount. +ENGINE = """ +import glob, http.server, json, os +repo = "models--hf-internal-testing--tiny-random-gpt2" +snap = os.path.join(os.environ["HF_HUB_CACHE"], repo, "snapshots", "*", "config.json") +class H(http.server.BaseHTTPRequestHandler): + def do_GET(self): + ok = self.path == "/health" and glob.glob(snap) + b = json.dumps({"status": "ok" if ok else "no model"}).encode() + self.send_response(200 if ok else 503) + self.send_header("content-length", str(len(b))) + self.end_headers() + self.wfile.write(b) +http.server.ThreadingHTTPServer(("", 8000), H).serve_forever() +""" + + +def _model_cache() -> dict[str, typing.Any]: + return { + "apiVersion": "modelplane.ai/v1alpha1", + "kind": "ModelCache", + "metadata": {"name": CACHE, "namespace": NAMESPACE}, + "spec": { + "source": "HuggingFace", + "huggingFace": {"repo": REPO, "sizeGiB": 1}, + }, + } + + +def _model_deployment() -> dict[str, typing.Any]: + return { + "apiVersion": "modelplane.ai/v1alpha1", + "kind": "ModelDeployment", + "metadata": {"name": DEPLOYMENT, "namespace": NAMESPACE}, + "spec": { + "replicas": 1, + "template": { + "spec": { + "modelCacheRef": {"name": CACHE}, + "engines": [ + { + "name": "mock", + "members": [ + { + "role": "Standalone", + "nodeSelector": { + "devices": [ + { + "name": "gpu", + "count": 1, + "selectors": [ + { + "cel": 'device.capacity["gpu.example.com"].memory.' + 'compareTo(quantity("20Gi")) >= 0' + } + ], + } + ] + }, + "template": { + "spec": { + "containers": [ + { + "name": "engine", + "image": IMAGE, + "command": ["python", "-u", "-c", ENGINE], + } + ] + } + }, + } + ], + } + ], + } + }, + }, + } + + +def _scale_mock_demo(control_plane: kube.Cluster, replicas: int) -> None: + control_plane.custom.patch_namespaced_custom_object( + GROUP, + "v1alpha1", + NAMESPACE, + "modeldeployments", + "mock-demo", + {"spec": {"replicas": replicas}}, + _request_timeout=kube.TIMEOUT_SECONDS, + ) + + +def _set_cache_storage(control_plane: kube.Cluster, storage_class: str | None) -> None: + control_plane.custom.patch_cluster_custom_object( + GROUP, + "v1alpha1", + "inferenceclusters", + CLUSTER, + {"spec": {"cluster": {"existing": {"cache": {"storageClassName": storage_class}}}}}, + _request_timeout=kube.TIMEOUT_SECONDS, + ) + + +def _delete(fn: collections.abc.Callable[[], object]) -> None: + try: + fn() + except client.ApiException as e: + if e.status != 404: + raise + + +@pytest.fixture(scope="module") +def cached(control_plane: kube.Cluster, workload: kube.Cluster) -> collections.abc.Iterator[None]: + """A ModelCache whose claim can't bind yet, and a ModelDeployment that reads it.""" + storage = client.StorageV1Api(workload.core.api_client) + storage.create_storage_class( + client.V1StorageClass( + metadata=client.V1ObjectMeta(name=STORAGE_CLASS), + provisioner="kubernetes.io/no-provisioner", + volume_binding_mode="Immediate", + ), + _request_timeout=kube.TIMEOUT_SECONDS, + ) + _scale_mock_demo(control_plane, 0) + _set_cache_storage(control_plane, STORAGE_CLASS) + for plural, obj in (("modelcaches", _model_cache()), ("modeldeployments", _model_deployment())): + control_plane.custom.create_namespaced_custom_object( + GROUP, "v1alpha1", NAMESPACE, plural, obj, _request_timeout=kube.TIMEOUT_SECONDS + ) + try: + yield + finally: + for plural, name in (("modeldeployments", DEPLOYMENT), ("modelcaches", CACHE)): + _delete( + lambda plural=plural, name=name: control_plane.custom.delete_namespaced_custom_object( + GROUP, "v1alpha1", NAMESPACE, plural, name, _request_timeout=kube.TIMEOUT_SECONDS + ) + ) + _delete(lambda: workload.core.delete_persistent_volume(VOLUME, _request_timeout=kube.TIMEOUT_SECONDS)) + _delete(lambda: storage.delete_storage_class(STORAGE_CLASS, _request_timeout=kube.TIMEOUT_SECONDS)) + _set_cache_storage(control_plane, None) + _scale_mock_demo(control_plane, 1) + + +def _replica(control_plane: kube.Cluster) -> dict[str, typing.Any]: + replicas = control_plane.custom.list_namespaced_custom_object( + GROUP, + "v1alpha1", + NAMESPACE, + "modelreplicas", + label_selector=f"modelplane.ai/deployment={DEPLOYMENT}", + _request_timeout=kube.TIMEOUT_SECONDS, + )["items"] + assert replicas, f"ModelDeployment {DEPLOYMENT} has no replica yet" + return replicas[0] + + +def _condition(obj: dict[str, typing.Any], typ: str) -> dict[str, typing.Any]: + return next((c for c in obj.get("status", {}).get("conditions", []) if c["type"] == typ), {}) + + +def _cache_entry(control_plane: kube.Cluster) -> dict[str, typing.Any]: + mc = control_plane.modelplane("modelcaches", CACHE, NAMESPACE) + assert mc is not None, f"ModelCache {CACHE} doesn't exist" + entry = next((c for c in mc.get("status", {}).get("clusters", []) if c["name"] == CLUSTER), None) + assert entry is not None, f"ModelCache {CACHE} reports nothing for cluster {CLUSTER}" + return entry + + +def _engine_pods(workload: kube.Cluster) -> list[client.V1Pod]: + # The endpoint picker's pods carry the deployment label too. + return workload.core.list_namespaced_pod( + REMOTE_NAMESPACE, + label_selector=f"modelplane.ai/deployment={DEPLOYMENT},modelplane.ai/engine=mock", + _request_timeout=kube.TIMEOUT_SECONDS, + ).items + + +@pytest.mark.usefixtures("cached") +def test_replica_waits_for_cache(control_plane: kube.Cluster, workload: kube.Cluster) -> None: + """While the cache's claim can't bind, the cache says so and the replica starts no engine.""" + + def waiting() -> str: + entry = _cache_entry(control_plane) + assert entry.get("phase") == "Pending", f"cache is {entry.get('phase')}" + assert "to bind" in entry.get("message", ""), f"cache message is {entry.get('message')!r}" + accepted = _condition(_replica(control_plane), "ModelAccepted") + assert accepted.get("reason") == "WaitingForCache", f"replica ModelAccepted is {accepted}" + return accepted["message"] + + message = wait.until(waiting, timeout=10 * 60, what="the replica to wait for its cache", retry=kube.RETRY) + assert f"ModelCache {CACHE} is Pending on cluster {CLUSTER}" in message + assert not _engine_pods(workload), "the replica started an engine before its cache was ready" + + +@pytest.mark.usefixtures("cached") +def test_replica_mounts_cache(control_plane: kube.Cluster, workload: kube.Cluster) -> None: + """Once the claim binds and the cache hydrates, the replica mounts what the cache publishes.""" + workload.core.create_persistent_volume( + client.V1PersistentVolume( + metadata=client.V1ObjectMeta(name=VOLUME), + spec=client.V1PersistentVolumeSpec( + capacity={"storage": "1Gi"}, + access_modes=["ReadWriteMany"], + storage_class_name=STORAGE_CLASS, + persistent_volume_reclaim_policy="Delete", + host_path=client.V1HostPathVolumeSource(path=HOST_PATH, type="DirectoryOrCreate"), + ), + ), + _request_timeout=kube.TIMEOUT_SECONDS, + ) + + def mounted() -> dict[str, typing.Any]: + entry = _cache_entry(control_plane) + assert entry.get("phase") == "Ready", f"cache is {entry.get('phase')}: {entry.get('message')}" + assert "mount" in entry, "a Ready cache publishes no mount" + return entry["mount"] + + mount = wait.until(mounted, timeout=10 * 60, what="the cache to hydrate", retry=kube.RETRY) + claim = mount["volumes"][0]["persistentVolumeClaim"]["claimName"] + + def serving() -> client.V1Pod: + ready = _condition(_replica(control_plane), "ModelReady") + assert ready.get("status") == "True", f"replica ModelReady is {ready}" + pods = _engine_pods(workload) + assert pods, "no engine pod" + return pods[0] + + pod = wait.until(serving, timeout=10 * 60, what="the cached replica to be Ready", retry=kube.RETRY) + claims = [v.persistent_volume_claim.claim_name for v in pod.spec.volumes if v.persistent_volume_claim] + assert claims == [claim], f"engine pod mounts {claims}, the cache publishes {claim}" + engine = next(c for c in pod.spec.containers if c.name == "engine") + assert {"name": "HF_HUB_CACHE", "value": "/mnt/models"} in [{"name": e.name, "value": e.value} for e in engine.env] + + staged = workload.exec( + pod.metadata.name, + REMOTE_NAMESPACE, + ["sh", "-c", "ls /mnt/models/models--hf-internal-testing--tiny-random-gpt2/snapshots/*/"], + ) + assert staged.code == 0, staged.stderr + assert "config.json" in staged.stdout.split(), staged.stdout diff --git a/functions/compose-model-cache/function/fn.py b/functions/compose-model-cache/function/fn.py index e81b370e2..fa2ea3cbe 100644 --- a/functions/compose-model-cache/function/fn.py +++ b/functions/compose-model-cache/function/fn.py @@ -89,6 +89,17 @@ def _namespace(meta: metav1.ObjectMeta | None) -> str: HYDRATION_IMAGE = "python:3.13-slim@sha256:57b29a7c050c4e679e4bea572e14b3965cf4b5839de2da49d55e49ffc4024f39" HYDRATION_MOUNT = "/mnt/artifact" +# Where an engine pod reads the cache, and the name of the volume it reads it +# through. A ModelReplica copies the mount status publishes into its pods. +# compose-model-replica's cache.mounted finds a running replica's cache by +# this volume name and HF_HUB_CACHE, so the three must change together. +MOUNT_PATH = "/mnt/models" +_MOUNT_VOLUME = "model-cache" + +# The annotation the scheduler sets on a claim it's waiting to provision for a +# node. +_SELECTED_NODE_ANNOTATION = "volume.kubernetes.io/selected-node" + # ttlSecondsAfterFinished governs how long the completed Job (and its # PVC-pinning pod) lingers before its TTL controller cascade-deletes both. Keep # it short so a just-deleted cache's PVC isn't pinned for long - but above @@ -117,7 +128,6 @@ def _storage_class(cluster: icv1alpha1.InferenceCluster) -> str | None: return None -# A completion marker written only after a fully successful download. The Job # A completion marker, checked instead of directory emptiness. A re-run # (eviction, replay, backoff) skips when the marker is present. Checking the # marker — not a non-empty dir — keeps re-runs safe: an interrupted download @@ -407,10 +417,11 @@ def _wrap_remote( spec.managementPolicies = management_policies return k8sobjv1alpha1.Object(spec=spec) - # --- naming (must stay in sync with backends/base.cache_pvc_name) --- - # Both sides share resource.child_name("modelcache", namespace, name), so the - # serving pods mount the PVC this composes by the same name in the same - # mirrored namespace. + # --- naming --- + # A replica reads the claim's name from the mount status publishes, so + # nothing else derives it. Keep it stable: running engines mount the claim + # by this name, and ModelExpress identifies the cache by it, so a new name + # rolls every cached engine onto a claim the cache must download again. def _pvc_name(self) -> str: return resource.child_name("modelcache", _namespace(self.xr.metadata), _name(self.xr.metadata)) @@ -496,12 +507,15 @@ def _was_ready(self, cluster_name: str) -> bool: def _observed_status(self, key: str) -> dict: """Remote resource status echoed back under Object.status.atProvider.manifest.status.""" + return self._observed_manifest(key).get("status", {}) or {} + + def _observed_manifest(self, key: str) -> dict: + """The remote resource as provider-kubernetes last observed it, under Object.status.atProvider.manifest.""" observed = self.req.observed.resources.get(key) if not observed: return {} obj = k8sobjv1alpha1.Object.model_validate(resource.struct_to_dict(observed.resource)) - manifest = (obj.status.atProvider.manifest if obj.status and obj.status.atProvider else None) or {} - return manifest.get("status", {}) or {} + return (obj.status.atProvider.manifest if obj.status and obj.status.atProvider else None) or {} def mark_ready_resources(self, per_cluster_phase: list[tuple[str, _Phase]]) -> None: """Mark composed Objects ready from each Object's own Ready condition. @@ -532,9 +546,63 @@ def write_status( ready_count = sum(1 for _, p in per_cluster_phase if p == PHASE_READY) status = v1alpha1.Status( summary=v1alpha1.Summary(ready=f"{ready_count}/{len(matched)}"), - clusters=[v1alpha1.Cluster(name=n, phase=p) for n, p in per_cluster_phase], + clusters=[ + v1alpha1.Cluster( + name=n, + phase=p, + message=self._message(n, p), + mount=self._mount() if p == PHASE_READY else None, + ) + for n, p in per_cluster_phase + ], + ) + resource.update_status(self.rsp.desired.composite, status.model_dump(exclude_none=True, by_alias=True)) + + def _mount(self) -> v1alpha1.Mount: + """How an engine pod reads the staged model: the claim at MOUNT_PATH, found by repo ID. + + HF_HUB_CACHE points HuggingFace's resolver at the hub cache the + hydration Job staged, so an engine loads the model by its repo ID. + """ + return v1alpha1.Mount( + volumes=[ + v1alpha1.Volume( + name=_MOUNT_VOLUME, + persistentVolumeClaim=v1alpha1.PersistentVolumeClaim(claimName=self._pvc_name()), + ) + ], + volumeMounts=[v1alpha1.VolumeMount(name=_MOUNT_VOLUME, mountPath=MOUNT_PATH)], + env=[v1alpha1.EnvItem(name="HF_HUB_CACHE", value=MOUNT_PATH)], + ) + + def _message(self, cluster_name: str, phase: _Phase) -> str | None: + """Why the model isn't readable on a cluster yet, or None once it is.""" + if phase == PHASE_READY: + return None + if self._auth_missing(): + auth = self.xr.spec.huggingFace.authSecret # ty: ignore[unresolved-attribute] # XRD guarantees huggingFace is set + assert auth is not None + return f"Waiting for key {auth.key or 'HF_TOKEN'} in Secret {_namespace(self.xr.metadata)}/{auth.name}" + repo = self.xr.spec.huggingFace.repo # ty: ignore[unresolved-attribute] # XRD guarantees huggingFace is set + pvc = f"{self.namespace}/{self._pvc_name()}" + if phase == PHASE_PENDING: + annotations = (self._observed_manifest(self._pvc_key(cluster_name)).get("metadata") or {}).get( + "annotations" + ) or {} + node = annotations.get(_SELECTED_NODE_ANNOTATION) + return f"Waiting for PersistentVolumeClaim {pvc} to bind" + (f" on node {node}" if node else "") + if phase == PHASE_HYDRATING: + return f"Downloading {repo} into PersistentVolumeClaim {pvc}" + job = self._observed_status(self._job_key(cluster_name)) + failed = int(job.get("failed") or 0) + reason = next( + (c.get("message") for c in job.get("conditions", []) if c.get("type") == "Failed" and c.get("message")), + None, + ) + times = f" {failed} times" if failed else "" + return f"Job {self.namespace}/{self._job_name()} failed{times} downloading {repo}" + ( + f": {reason}" if reason else "" ) - resource.update_status(self.rsp.desired.composite, status) def derive_conditions( self, matched: list[icv1alpha1.InferenceCluster], per_cluster_phase: list[tuple[str, _Phase]] diff --git a/functions/compose-model-cache/tests/test_fn.py b/functions/compose-model-cache/tests/test_fn.py index f364bdb16..eac55ceee 100644 --- a/functions/compose-model-cache/tests/test_fn.py +++ b/functions/compose-model-cache/tests/test_fn.py @@ -244,6 +244,62 @@ def _observed_job_object(*, condition: str, ready: bool) -> fnv1.Resource: ) +def _observed_pending_pvc_object(*, selected_node: str | None) -> fnv1.Resource: + """The cache PVC's Object as observed with the PVC Pending, annotated with the node the scheduler chose for it, if any.""" + manifest: dict[str, Any] = {"status": {"phase": "Pending"}} + if selected_node is not None: + manifest["metadata"] = {"annotations": {"volume.kubernetes.io/selected-node": selected_node}} + return fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "kubernetes.m.crossplane.io/v1alpha1", + "kind": "Object", + "spec": {"forProvider": {"manifest": {}}}, + "status": {"atProvider": {"manifest": manifest}}, + } + ), + ) + + +def _observed_failed_job_object(*, failed: int, message: str) -> fnv1.Resource: + """The hydration Job's Object as observed, the Job Failed with message after failed attempts.""" + return fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "kubernetes.m.crossplane.io/v1alpha1", + "kind": "Object", + "spec": {"forProvider": {"manifest": {}}}, + "status": { + "atProvider": { + "manifest": { + "status": { + "failed": failed, + "conditions": [ + { + "type": "Failed", + "status": "True", + "reason": "BackoffLimitExceeded", + "message": message, + } + ], + } + } + } + }, + } + ), + ) + + +def _mount() -> dict: + """The qwen ModelCache's mount on a cluster where it's Ready.""" + return { + "volumes": [{"name": "model-cache", "persistentVolumeClaim": {"claimName": "modelcache-ml-team-qwen-17db2"}}], + "volumeMounts": [{"name": "model-cache", "mountPath": "/mnt/models"}], + "env": [{"name": "HF_HUB_CACHE", "value": "/mnt/models"}], + } + + def _to_dict(msg: message.Message) -> dict: """msg as a dict with sorted keys, so pytest's diff of two lines them up.""" return json.loads(json_format.MessageToJson(msg, sort_keys=True)) @@ -281,7 +337,13 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="0/1", - clusters=[{"name": "cluster-a", "phase": "Pending"}], + clusters=[ + { + "name": "cluster-a", + "phase": "Pending", + "message": "Waiting for PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2 to bind", + } + ], ready=fnv1.READY_UNSPECIFIED, ), resources={ @@ -353,7 +415,13 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="0/1", - clusters=[{"name": "cluster-a", "phase": "Pending"}], + clusters=[ + { + "name": "cluster-a", + "phase": "Pending", + "message": "Waiting for PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2 to bind", + } + ], ready=fnv1.READY_UNSPECIFIED, ), resources={ @@ -453,7 +521,13 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="0/1", - clusters=[{"name": "eks-a", "phase": "Pending"}], + clusters=[ + { + "name": "eks-a", + "phase": "Pending", + "message": "Waiting for PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2 to bind", + } + ], ready=fnv1.READY_UNSPECIFIED, ), resources={ @@ -522,7 +596,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="1/1", - clusters=[{"name": "cluster-a", "phase": "Ready"}], + clusters=[{"name": "cluster-a", "phase": "Ready", "mount": _mount()}], ready=fnv1.READY_TRUE, ), resources={ @@ -577,7 +651,13 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="0/1", - clusters=[{"name": "cluster-a", "phase": "Hydrating"}], + clusters=[ + { + "name": "cluster-a", + "phase": "Hydrating", + "message": "Downloading Qwen/Qwen3-0.6B into PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2", + } + ], ready=fnv1.READY_UNSPECIFIED, ), resources={ @@ -638,7 +718,13 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="0/1", - clusters=[{"name": "cluster-a", "phase": "Failed"}], + clusters=[ + { + "name": "cluster-a", + "phase": "Failed", + "message": "Job mp-ml-team-51733/modelcache-ml-team-qwen-hydrate-256ec failed downloading Qwen/Qwen3-0.6B", + } + ], ready=fnv1.READY_UNSPECIFIED, ), resources={ @@ -707,7 +793,14 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="1/2", - clusters=[{"name": "a", "phase": "Ready"}, {"name": "b", "phase": "Hydrating"}], + clusters=[ + {"name": "a", "phase": "Ready", "mount": _mount()}, + { + "name": "b", + "phase": "Hydrating", + "message": "Downloading Qwen/Qwen3-0.6B into PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2", + }, + ], ready=fnv1.READY_UNSPECIFIED, ), resources={ @@ -782,7 +875,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="1/1", - clusters=[{"name": "cluster-a", "phase": "Ready"}], + clusters=[{"name": "cluster-a", "phase": "Ready", "mount": _mount()}], ready=fnv1.READY_TRUE, ), resources={ @@ -880,7 +973,13 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="0/1", - clusters=[{"name": "cluster-a", "phase": "Pending"}], + clusters=[ + { + "name": "cluster-a", + "phase": "Pending", + "message": "Waiting for key HF_TOKEN in Secret ml-team/hf-token", + } + ], ready=fnv1.READY_FALSE, ), resources={ @@ -951,7 +1050,13 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="0/1", - clusters=[{"name": "cluster-a", "phase": "Pending"}], + clusters=[ + { + "name": "cluster-a", + "phase": "Pending", + "message": "Waiting for key HF_TOKEN in Secret ml-team/hf-token", + } + ], ready=fnv1.READY_FALSE, ), resources={ @@ -1047,7 +1152,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="1/1", - clusters=[{"name": "cluster-a", "phase": "Ready"}], + clusters=[{"name": "cluster-a", "phase": "Ready", "mount": _mount()}], ready=fnv1.READY_TRUE, ), resources={ @@ -1125,7 +1230,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( composite=_desired_model_cache( summary="1/1", - clusters=[{"name": "cluster-a", "phase": "Ready"}], + clusters=[{"name": "cluster-a", "phase": "Ready", "mount": _mount()}], ready=fnv1.READY_TRUE, ), resources={ @@ -1197,6 +1302,141 @@ def _to_dict(msg: message.Message) -> dict: ], ), ), + Case( + name="PendingClaimNamesNode", + reason="While its PVC waits to bind, a ModelCache says which claim it waits on, and the node the scheduler chose for it.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_cache(revision=None, auth_secret=None, status=None), + resources={ + "pvc-cluster-a": _observed_pending_pvc_object(selected_node="computeinstance-e05mkw6q6qs5hcmaa7"), + }, + ), + required_resources={ + "clusters": fnv1.Resources( + items=[ + _cluster( + name="cluster-a", + provider_config="cluster-a-pc", + cluster={"source": "GKE", "gke": {"project": "my-project", "region": "us-central1"}}, + storage_class="modelplane-rwx", + ), + ], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + composite=_desired_model_cache( + summary="0/1", + clusters=[ + { + "name": "cluster-a", + "phase": "Pending", + "message": "Waiting for PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2 to bind on node computeinstance-e05mkw6q6qs5hcmaa7", + } + ], + ready=fnv1.READY_UNSPECIFIED, + ), + resources={ + "pvc-cluster-a": _pvc_object( + provider_config="cluster-a-pc", + storage_class="modelplane-rwx", + ready=fnv1.READY_UNSPECIFIED, + ), + "hydrate-cluster-a": _job_object( + provider_config="cluster-a-pc", + command=( + "set -e; if [ -f /mnt/artifact/.modelplane-hydrated ]; then echo 'already hydrated, skipping'; exit 0; fi; " + "pip install --quiet huggingface_hub; hf download Qwen/Qwen3-0.6B; " + "touch /mnt/artifact/.modelplane-hydrated" + ), + env=[{"name": "HF_HUB_CACHE", "value": "/mnt/artifact"}], + ), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "clusters": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceCluster"), + }, + ), + conditions=[ + fnv1.Condition(type="ClustersMatched", status=fnv1.STATUS_CONDITION_TRUE, reason="Matched"), + fnv1.Condition(type="ArtifactReady", status=fnv1.STATUS_CONDITION_FALSE, reason="Hydrating"), + ], + ), + ), + Case( + name="FailedJobMessage", + reason="A ModelCache whose hydration Job failed says which Job, how many attempts failed, and why.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_cache(revision=None, auth_secret=None, status=None), + resources={ + "pvc-cluster-a": _observed_pvc_object(), + "hydrate-cluster-a": _observed_failed_job_object( + failed=4, message="Job has reached the specified backoff limit" + ), + }, + ), + required_resources={ + "clusters": fnv1.Resources( + items=[ + _cluster( + name="cluster-a", + provider_config="cluster-a-pc", + cluster={"source": "GKE", "gke": {"project": "my-project", "region": "us-central1"}}, + storage_class="modelplane-rwx", + ), + ], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + composite=_desired_model_cache( + summary="0/1", + clusters=[ + { + "name": "cluster-a", + "phase": "Failed", + "message": "Job mp-ml-team-51733/modelcache-ml-team-qwen-hydrate-256ec failed 4 times downloading Qwen/Qwen3-0.6B: Job has reached the specified backoff limit", + } + ], + ready=fnv1.READY_UNSPECIFIED, + ), + resources={ + "pvc-cluster-a": _pvc_object( + provider_config="cluster-a-pc", + storage_class="modelplane-rwx", + ready=fnv1.READY_TRUE, + ), + "hydrate-cluster-a": _job_object( + provider_config="cluster-a-pc", + command=( + "set -e; if [ -f /mnt/artifact/.modelplane-hydrated ]; then echo 'already hydrated, skipping'; exit 0; fi; " + "pip install --quiet huggingface_hub; hf download Qwen/Qwen3-0.6B; " + "touch /mnt/artifact/.modelplane-hydrated" + ), + env=[{"name": "HF_HUB_CACHE", "value": "/mnt/artifact"}], + ), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "clusters": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceCluster"), + }, + ), + conditions=[ + fnv1.Condition(type="ClustersMatched", status=fnv1.STATUS_CONDITION_TRUE, reason="Matched"), + fnv1.Condition(type="ArtifactReady", status=fnv1.STATUS_CONDITION_FALSE, reason="Failed"), + ], + ), + ), ] diff --git a/functions/compose-model-replica/function/backends/base.py b/functions/compose-model-replica/function/backends/base.py index f962652b5..5f64e93ae 100644 --- a/functions/compose-model-replica/function/backends/base.py +++ b/functions/compose-model-replica/function/backends/base.py @@ -26,6 +26,7 @@ from crossplane.function import resource from crossplane.function.proto.v1 import run_function_pb2 as fnv1 +from models.ai.modelplane.modelcache import v1alpha1 as mcv1alpha1 from models.ai.modelplane.modelreplica import v1alpha1 from models.io.crossplane.m.kubernetes.object import v1alpha1 as k8sobjv1alpha1 from models.io.k8s.apimachinery.pkg.apis.meta import v1 as metav1 @@ -55,6 +56,7 @@ def build( provider_config: str, serving_label: str, stack: str, + cache: mcv1alpha1.Mount | None, ) -> dict[str, k8sobjv1alpha1.Object]: ... def ready_copies(self, engine: v1alpha1.Engine, observed: Mapping[str, fnv1.Resource]) -> int: @@ -72,67 +74,47 @@ def ready_copies(self, engine: v1alpha1.Engine, observed: Mapping[str, fnv1.Reso ROLE_LEADER = "Leader" ROLE_WORKER = "Worker" -# Mount path the cache PVC is exposed at inside every engine pod. Intrinsic -# to the cache contract; the deployment points the engine here. -CACHE_MOUNT_PATH = "/mnt/models" -# Volume name shared by the PVC volume and its mount. -_CACHE_VOLUME = "model-cache" +def cache_mounts(cache: mcv1alpha1.Mount | None) -> tuple[list[dict], list[dict]]: + """Return (volumes, volumeMounts) that read the replica's cache on its cluster, or ([], []). - -def cache_pvc_name(namespace: str, cache_name: str) -> str: - # MUST stay in sync with compose-model-cache's _pvc_name() - # (functions/compose-model-cache/function/fn.py) — both sides share - # resource.child_name("modelcache", namespace, name), so a replica mounts the - # PVC the cache composes by the same name in the same mirrored namespace. - return resource.child_name("modelcache", namespace, cache_name) - - -def cache_mounts(replica: v1alpha1.ModelReplica) -> tuple[list[dict], list[dict]]: - """Return (volumes, volumeMounts) for the replica's cache, or ([], []). - - The cache is a per-cache PVC qualified by the replica's namespace - (modelCacheRef carries only a name, and the ModelCache is in the replica's - own namespace). The PVC is shared across every engine and member of the - replica. + The ModelCache publishes them for each cluster it has staged the model on, + so every engine and member of the replica reads the cache the same way, + whatever its source. """ - ref = replica.spec.modelCacheRef - if not ref: + if cache is None: return [], [] - pvc = cache_pvc_name(_namespace(replica.metadata), ref.name) - # Mounted read-write (NOT readOnly): engines write into the model dir - # (tokenizer/compile/lock artifacts), and a readOnly mount hard-fails them. - # The PVC is ReadWriteMany, so every pod in the gang shares one read-write - # mount; the hydration Job populates it once and serving pods read N times. return ( - [{"name": _CACHE_VOLUME, "persistentVolumeClaim": {"claimName": pvc}}], - [{"name": _CACHE_VOLUME, "mountPath": CACHE_MOUNT_PATH}], + [v.model_dump(exclude_none=True) for v in cache.volumes or []], + [m.model_dump(exclude_none=True) for m in cache.volumeMounts or []], ) -def cache_env(replica: v1alpha1.ModelReplica) -> list[dict]: - """Env that makes the cache mount resolvable by the model's own name, or []. +def cache_env(cache: mcv1alpha1.Mount | None) -> list[dict]: + """Env that reads the replica's cache, as its ModelCache publishes it, or []. - compose-model-cache stages a HuggingFace source in HuggingFace's own cache - layout, so pointing HF_HUB_CACHE at the mount lets an engine load by repo id - against the staged snapshot. The command is then the same cached or not, and - Modelplane injects no --model of its own (#407). No HF_HUB_OFFLINE: it isn't - needed to resolve the snapshot, and it breaks an engine that fetches a - second repo at startup, like kimi-k2's gated tokenizer. + A HuggingFace cache sets HF_HUB_CACHE to the mount, so an engine loads the + model by its repo ID and its command is the same cached or not (#407). + Modelplane sets no HF_HUB_OFFLINE: it isn't needed to find the staged + model, and it breaks an engine that fetches a second repo at startup, like + kimi-k2's gated tokenizer. An engine that must not reach HuggingFace sets + it itself. An engine must name the revision its cache staged. A bare repo id resolves at the default branch, so a cache pinned to a commit or tag misses unless the engine passes that revision too. - - HF_HUB_CACHE is a cache root, not a pin, so an engine fetching some other - repo by id writes it to the shared PVC rather than its own filesystem. - - HuggingFace-specific because the source is. A second ModelCache source would - stage its own layout and gate this on the source. """ - if not replica.spec.modelCacheRef: + if cache is None: return [] - return [{"name": "HF_HUB_CACHE", "value": CACHE_MOUNT_PATH}] + return [e.model_dump(exclude_none=True) for e in cache.env or []] + + +def _cache_claim(cache: mcv1alpha1.Mount) -> str | None: + """The claim the cache's first claim-backed volume reads, or None.""" + for v in cache.volumes or []: + if v.persistentVolumeClaim: + return v.persistentVolumeClaim.claimName + return None # Well-known name and namespace of the per-cluster shared ModelExpress server @@ -151,7 +133,7 @@ def cache_env(replica: v1alpha1.ModelReplica) -> list[dict]: _MODELEXPRESS_PORT = 8001 -def modelexpress_env(replica: v1alpha1.ModelReplica, stack: str) -> list[dict]: +def modelexpress_env(replica: v1alpha1.ModelReplica, stack: str, cache: mcv1alpha1.Mount | None) -> list[dict]: """ModelExpress P2P env for an engine pod that references a cache on a Dynamo cluster, or [] otherwise. @@ -170,24 +152,24 @@ def modelexpress_env(replica: v1alpha1.ModelReplica, stack: str) -> list[dict]: ModelExpress content-addresses from the model name and revision. Two caches of the same repo, neither pinning a revision, can hold different commits while both engines report none, so a replica could load a peer's other - weights. The PVC name is stable across a cache's replicas and distinct - across caches and namespaces; the resolved commit would be better, but only - the hydration Job sees it. The cost is that identical caches can't share a - source. + weights. It's the name of the claim the engine mounts, which is stable + across a cache's replicas and distinct across caches and namespaces; the + resolved commit would be better, but only the hydration Job sees it. The + cost is that identical caches can't share a source. POD_* identify the publishing pod. The pinned 0.4.1 server ignores them. A 0.5.0 server owns a ModelMetadata CR by its publishing pod, so Kubernetes garbage-collects it, but only for a pod in the server's own namespace; owner references can't cross namespaces, and engines run in their team's. """ - ref = replica.spec.modelCacheRef - if not ref or stack != "Dynamo": + claim = _cache_claim(cache) if cache is not None else None + if not replica.spec.modelCacheRef or claim is None or stack != "Dynamo": return [] address = f"{_MODELEXPRESS_SERVER_SERVICE}.{_MODELEXPRESS_NAMESPACE}.svc:{_MODELEXPRESS_PORT}" return [ {"name": "MX_SERVER_ADDRESS", "value": address}, {"name": "MODEL_EXPRESS_URL", "value": address}, - {"name": "MX_MODEL_REVISION", "value": cache_pvc_name(_namespace(replica.metadata), ref.name)}, + {"name": "MX_MODEL_REVISION", "value": claim}, {"name": "MX_P2P_METADATA", "value": "1"}, {"name": "POD_NAME", "valueFrom": {"fieldRef": {"fieldPath": "metadata.name"}}}, {"name": "POD_UID", "valueFrom": {"fieldRef": {"fieldPath": "metadata.uid"}}}, @@ -195,11 +177,15 @@ def modelexpress_env(replica: v1alpha1.ModelReplica, stack: str) -> list[dict]: ] -def modelexpress_security_context(replica: v1alpha1.ModelReplica, stack: str) -> dict | None: +def modelexpress_security_context( + replica: v1alpha1.ModelReplica, stack: str, cache: mcv1alpha1.Mount | None +) -> dict | None: """IPC_LOCK, which GPUDirect RDMA needs to pin memory, or None unless the - replica references a cache on a Dynamo cluster. + replica mounts a cache on a Dynamo cluster. - Gated like modelexpress_env, but unlike that env this isn't inert. The + Gated like modelexpress_env, on the mount rather than the reference, so a + running replica that gains a cache reference keeps its pods as they are + until it mounts the cache. Unlike that env this isn't inert. The restricted Pod Security Standard permits no added capability except NET_BIND_SERVICE, so a workload cluster that labels the serving namespace pod-security.kubernetes.io/enforce=restricted rejects every @@ -211,8 +197,8 @@ def modelexpress_security_context(replica: v1alpha1.ModelReplica, stack: str) -> declares a fabric? An engine with no fabric can't do RDMA, so it has no use for the capability. """ - ref = replica.spec.modelCacheRef - if not ref or stack != "Dynamo": + claim = _cache_claim(cache) if cache is not None else None + if not replica.spec.modelCacheRef or claim is None or stack != "Dynamo": return None return {"capabilities": {"add": ["IPC_LOCK"]}} diff --git a/functions/compose-model-replica/function/backends/grove.py b/functions/compose-model-replica/function/backends/grove.py index 72d80d75b..dd6b2d7a8 100644 --- a/functions/compose-model-replica/function/backends/grove.py +++ b/functions/compose-model-replica/function/backends/grove.py @@ -40,6 +40,7 @@ from crossplane.function import resource from crossplane.function.proto.v1 import run_function_pb2 as fnv1 +from models.ai.modelplane.modelcache import v1alpha1 as mcv1alpha1 from models.ai.modelplane.modelreplica import v1alpha1 from models.io.crossplane.m.kubernetes.object import v1alpha1 as k8sobjv1alpha1 @@ -61,6 +62,7 @@ def build( provider_config: str, serving_label: str, stack: str, + cache: mcv1alpha1.Mount | None, ) -> dict[str, k8sobjv1alpha1.Object]: leader = base.engine_member(engine, base.ROLE_LEADER) worker = base.engine_member(engine, base.ROLE_WORKER) @@ -74,7 +76,7 @@ def build( # The worker clique's pod count: one follower pod per node. worker_replicas = int(worker.worker.nodes) if worker.worker else 1 - cache_volumes, cache_volume_mounts = base.cache_mounts(replica) + cache_volumes, cache_volume_mounts = base.cache_mounts(cache) def container(member: v1alpha1.Member, *, serving: bool) -> dict: engine_container = base.engine_container(member) @@ -100,14 +102,14 @@ def container(member: v1alpha1.Member, *, serving: bool) -> dict: # too, since each pod publishes itself as a source independently. env = [ base.grove_leader_address_env(), - *base.cache_env(replica), - *base.modelexpress_env(replica, stack), + *base.cache_env(cache), + *base.modelexpress_env(replica, stack, cache), ] if engine_container.env: env.extend(e.model_dump(exclude_none=True) for e in engine_container.env) if env: c["env"] = env - security_context = base.modelexpress_security_context(replica, stack) + security_context = base.modelexpress_security_context(replica, stack, cache) if security_context: c["securityContext"] = security_context if serving: diff --git a/functions/compose-model-replica/function/backends/llmd.py b/functions/compose-model-replica/function/backends/llmd.py index d40ef0669..e5639f7af 100644 --- a/functions/compose-model-replica/function/backends/llmd.py +++ b/functions/compose-model-replica/function/backends/llmd.py @@ -39,6 +39,7 @@ from collections.abc import Mapping from crossplane.function.proto.v1 import run_function_pb2 as fnv1 +from models.ai.modelplane.modelcache import v1alpha1 as mcv1alpha1 from models.ai.modelplane.modelreplica import v1alpha1 from models.io.crossplane.m.kubernetes.object import v1alpha1 as k8sobjv1alpha1 @@ -60,6 +61,7 @@ def build( # backend. Unused here: this backend is Standard-only, so there's no # stack to switch on. stack: str, # noqa: ARG002 + cache: mcv1alpha1.Mount | None, ) -> dict[str, k8sobjv1alpha1.Object]: leader = base.engine_member(engine, base.ROLE_LEADER) worker = base.engine_member(engine, base.ROLE_WORKER) @@ -73,7 +75,7 @@ def build( # Gang size: the leader plus the worker's nodes (one follower pod each). size = 1 + (int(worker.worker.nodes) if worker.worker else 1) - cache_volumes, cache_volume_mounts = base.cache_mounts(replica) + cache_volumes, cache_volume_mounts = base.cache_mounts(cache) def container(member: v1alpha1.Member, *, serving: bool) -> dict: engine_container = base.engine_container(member) @@ -99,7 +101,7 @@ def container(member: v1alpha1.Member, *, serving: bool) -> dict: # LWS prepends its own LWS_* vars ahead of all of these in the # running pod. cache_env points HuggingFace at the mount so every # member's --model= resolves against the pre-staged snapshot. - env = [base.leader_address_env(), base.rank_env(), *base.cache_env(replica)] + env = [base.leader_address_env(), base.rank_env(), *base.cache_env(cache)] if engine_container.env: env.extend(e.model_dump(exclude_none=True) for e in engine_container.env) c["env"] = env diff --git a/functions/compose-model-replica/function/backends/native.py b/functions/compose-model-replica/function/backends/native.py index 5c8a3133a..0229b0013 100644 --- a/functions/compose-model-replica/function/backends/native.py +++ b/functions/compose-model-replica/function/backends/native.py @@ -28,6 +28,7 @@ from collections.abc import Mapping from crossplane.function.proto.v1 import run_function_pb2 as fnv1 +from models.ai.modelplane.modelcache import v1alpha1 as mcv1alpha1 from models.ai.modelplane.modelreplica import v1alpha1 from models.io.crossplane.m.kubernetes.object import v1alpha1 as k8sobjv1alpha1 @@ -42,6 +43,7 @@ def build( provider_config: str, serving_label: str, stack: str, + cache: mcv1alpha1.Mount | None, ) -> dict[str, k8sobjv1alpha1.Object]: member = base.engine_member(engine, base.ROLE_STANDALONE) # select_backend dispatches the native backend only for an engine with a @@ -58,7 +60,7 @@ def build( pod_labels = {base.LABEL_SERVING: serving_label, base.LABEL_WORKLOAD: name} selector = {base.LABEL_WORKLOAD: name} - cache_volumes, cache_volume_mounts = base.cache_mounts(replica) + cache_volumes, cache_volume_mounts = base.cache_mounts(cache) container = { "name": "engine", @@ -81,12 +83,12 @@ def build( # Modelplane's entries lead so the user's can reference them. A scaled # Standalone deployment is as valid a ModelExpress peer set as a gang, # so it gets that env too. - env = [*base.cache_env(replica), *base.modelexpress_env(replica, stack)] + env = [*base.cache_env(cache), *base.modelexpress_env(replica, stack, cache)] if engine_container.env: env.extend(e.model_dump(exclude_none=True) for e in engine_container.env) if env: container["env"] = env - security_context = base.modelexpress_security_context(replica, stack) + security_context = base.modelexpress_security_context(replica, stack, cache) if security_context: container["securityContext"] = security_context diff --git a/functions/compose-model-replica/function/cache.py b/functions/compose-model-replica/function/cache.py new file mode 100644 index 000000000..46b02c6bf --- /dev/null +++ b/functions/compose-model-replica/function/cache.py @@ -0,0 +1,181 @@ +# Copyright 2026 The Modelplane Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Read a replica's ModelCache: how to mount it on the replica's cluster, or why it can't yet. + +compose-model-cache publishes, for each cluster it has staged the model on, the +volumes, mounts and environment that read it there. A replica copies them into +its engine pods instead of working out where the model is. +""" + +import dataclasses +from collections.abc import Iterator, Mapping + +from crossplane.function import request, resource, response +from crossplane.function.proto.v1 import run_function_pb2 as fnv1 +from models.ai.modelplane.modelcache import v1alpha1 as mcv1alpha1 +from models.ai.modelplane.modelreplica import v1alpha1 + +# The name the replica requires its ModelCache under. +_REQUIREMENT = "cache" + +# The volume a HuggingFace cache's mount names, and the variable it points at +# the mount. Must match compose-model-cache's _mount. Used only to read what an +# already running replica's engines mount, so they keep it while their cache +# publishes no mount. +_VOLUME = "model-cache" +_ENV = "HF_HUB_CACHE" + + +@dataclasses.dataclass +class Resolution: + """How the replica reads its cache, or why it can't yet. + + mount is what the replica's engines mount, or None if they mount nothing + yet. waiting is empty once the ModelCache publishes a mount for the + replica's cluster, and otherwise says why it hasn't. + """ + + mount: mcv1alpha1.Mount | None + waiting: str + + +def require(rsp: fnv1.RunFunctionResponse, replica: v1alpha1.ModelReplica) -> None: + """Require the replica's ModelCache, which is in the replica's own namespace, if it reads one.""" + ref = replica.spec.modelCacheRef + if not ref: + return + response.require_resources( + rsp, + name=_REQUIREMENT, + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name=ref.name, + namespace=_namespace(replica), + ) + + +def resolve(req: fnv1.RunFunctionRequest, replica: v1alpha1.ModelReplica) -> Resolution | None: + """How the replica reads its ModelCache on its cluster, or None if it reads no cache.""" + ref = replica.spec.modelCacheRef + if not ref: + return None + # A requirement Crossplane hasn't resolved yet is absent from the request, + # where one that matched nothing is present and empty. + if _REQUIREMENT not in req.required_resources: + return Resolution(mount=None, waiting=f"Waiting for ModelCache {ref.name}") + found = request.get_required_resource(req, _REQUIREMENT) + if found is None: + return Resolution(mount=None, waiting=f"ModelCache {ref.name} not found in namespace {_namespace(replica)}") + mc = mcv1alpha1.ModelCache.model_validate(found) + cluster = replica.spec.clusterName + entry = next((c for c in (mc.status.clusters if mc.status else None) or [] if c.name == cluster), None) + if entry is None: + return Resolution(mount=None, waiting=f"ModelCache {ref.name} isn't staged on cluster {cluster}") + if entry.mount is not None: + return Resolution(mount=entry.mount, waiting="") + if entry.phase == "Ready": + # Status written before the mount existed, until the cache's next + # reconcile publishes one. + waiting = f"Waiting for ModelCache {ref.name} to publish how to mount it on cluster {cluster}" + else: + waiting = f"ModelCache {ref.name} is {entry.phase or 'Pending'} on cluster {cluster}" + if entry.message: + waiting += f": {entry.message}" + return Resolution(mount=None, waiting=waiting) + + +def kept(replica: v1alpha1.ModelReplica, observed: Mapping[str, fnv1.Resource], keys: list[str]) -> Resolution: + """What an already running replica keeps mounting while its cache publishes no mount. + + The engines keep the cache volume, mount and env their workload has now, + or no cache if it has none, so they don't roll onto a cache that isn't + ready. waiting says so, the same way every reconcile. + """ + ref = replica.spec.modelCacheRef + assert ref is not None + for key in keys: + if key not in observed: + continue + obj = resource.struct_to_dict(observed[key].resource) + mount = mounted(obj.get("spec", {}).get("forProvider", {}).get("manifest", {})) + if mount is not None: + return Resolution( + mount=mount, + waiting=f"ModelCache {ref.name} publishes no mount for cluster {replica.spec.clusterName}. " + "The replica's engines keep the cache they mount now.", + ) + return Resolution( + mount=None, + waiting=f"ModelCache {ref.name} publishes no mount for cluster {replica.spec.clusterName}. " + "The replica's engines keep running without a cache.", + ) + + +def mounted(manifest: dict) -> mcv1alpha1.Mount | None: + """The cache volume, mount and env a workload's pods have, or None if they mount no cache. + + A Deployment, a LeaderWorkerSet's leader and worker templates, and a + PodCliqueSet's cliques hold their pod specs at different paths, and the + replica mounts the same cache in each, so the first pod spec with the + cache volume says what they all mount. + """ + for pod in _pod_specs(manifest): + volume = next((v for v in pod.get("volumes") or [] if v.get("name") == _VOLUME), None) + if volume is None: + continue + # The engine container mounts the cache. Its sidecars don't. + engine = next( + ( + c + for c in pod.get("containers") or [] + if any(m.get("name") == _VOLUME for m in c.get("volumeMounts") or []) + ), + {}, + ) + mounts = [ + mcv1alpha1.VolumeMount.model_validate(m) + for m in engine.get("volumeMounts") or [] + if m.get("name") == _VOLUME + ] + # The cache's variables lead the engine's env, so the first HF_HUB_CACHE + # is the cache's. One the engine sets itself comes later, and keeping it + # too would add another copy on every reconcile. + env = next((e for e in engine.get("env") or [] if e.get("name") == _ENV), None) + return mcv1alpha1.Mount( + volumes=[mcv1alpha1.Volume.model_validate(volume)], + volumeMounts=mounts or None, + env=[mcv1alpha1.EnvItem.model_validate(env)] if env else None, + ) + return None + + +def _pod_specs(value: object) -> Iterator[dict]: + """Every pod spec in a workload manifest, wherever its kind keeps it.""" + if isinstance(value, dict): + if "containers" in value: + yield value + return + for v in value.values(): + yield from _pod_specs(v) + elif isinstance(value, list): + for v in value: + yield from _pod_specs(v) + + +def _namespace(replica: v1alpha1.ModelReplica) -> str: + """The replica's namespace, always set on a namespaced resource read from the API server.""" + if replica.metadata is None or replica.metadata.namespace is None: + raise ValueError("metadata.namespace is unexpectedly absent") + return replica.metadata.namespace diff --git a/functions/compose-model-replica/function/fn.py b/functions/compose-model-replica/function/fn.py index 6ed0814c1..d6249560e 100644 --- a/functions/compose-model-replica/function/fn.py +++ b/functions/compose-model-replica/function/fn.py @@ -32,10 +32,11 @@ from crossplane.function.proto.v1 import run_function_pb2 as fnv1 from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1 from models.ai.modelplane.inferencecluster import v1alpha1 as icv1alpha1 +from models.ai.modelplane.modelcache import v1alpha1 as mcv1alpha1 from models.ai.modelplane.modelreplica import v1alpha1 from models.io.crossplane.m.kubernetes.object import v1alpha1 as k8sobjv1alpha1 -from function import routing +from function import cache, routing from function.backends import base, grove, llmd, native # Condition types and reasons for the ModelReplica XR. @@ -44,6 +45,7 @@ CONDITION_TYPE_MODEL_READY = "ModelReady" CONDITION_REASON_WAITING_FOR_CLUSTER = "WaitingForCluster" +CONDITION_REASON_WAITING_FOR_CACHE = "WaitingForCache" CONDITION_REASON_WAITING_FOR_MODEL = "WaitingForModel" CONDITION_REASON_DEPLOYING = "Deploying" CONDITION_REASON_ACCEPTED = "Accepted" @@ -93,11 +95,58 @@ def __init__(self, req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse) def compose(self) -> None: if not self.resolve_inputs(): return - self.compose_model_serving() + found, mount = self.resolve_cache() + if not found: + return + self.compose_model_serving(mount) self.derive_conditions() + def resolve_cache(self) -> tuple[bool, mcv1alpha1.Mount | None]: + """How the replica's engines read its ModelCache, if it reads one. + + Returns (True, mount) once the cache publishes a mount for the + replica's cluster, or (True, None) when the replica reads no cache. + A new replica waits for the mount, and composes nothing until then, so + its engines don't start before the model is there. A replica whose + engines already run keeps the cache volume, mount and env their + workload has now, or no cache if it has none, until the cache + publishes a mount. So a cache that stops being Ready, is deleted, or + is added or switched to before it's Ready doesn't roll a serving + model's engines. Returns (False, None) while waiting. + """ + resolution = cache.resolve(self.req, self.xr) + if resolution is None: + return True, None + if resolution.mount is not None: + return True, resolution.mount + keys = base.workload_keys(self.xr) + if any(k in self.req.observed.resources for k in keys): + kept = cache.kept(self.xr, self.req.observed.resources, keys) + response.warning(self.rsp, kept.waiting) + return True, kept.mount + response.set_conditions( + self.rsp, + resource.Condition( + typ=CONDITION_TYPE_MODEL_ACCEPTED, + status="False", + reason=CONDITION_REASON_WAITING_FOR_CACHE, + message=resolution.waiting, + ), + resource.Condition( + typ=CONDITION_TYPE_MODEL_SERVING, status="False", reason=CONDITION_REASON_WAITING_FOR_MODEL + ), + resource.Condition( + typ=CONDITION_TYPE_MODEL_READY, status="False", reason=CONDITION_REASON_WAITING_FOR_MODEL + ), + ) + # Nothing is composed while waiting, and an XR with no composed + # resources would otherwise be trivially ready. + self.rsp.desired.composite.ready = fnv1.READY_FALSE + return False, None + def resolve_inputs(self) -> bool: - """Declare and fetch the referenced InferenceCluster.""" + """Declare the referenced InferenceCluster and ModelCache, and fetch the cluster.""" + cache.require(self.rsp, self.xr) response.require_resources( self.rsp, name="cluster", @@ -153,7 +202,7 @@ def resolve_inputs(self) -> bool: return True - def compose_model_serving(self) -> None: + def compose_model_serving(self, mount: mcv1alpha1.Mount | None) -> None: """Compose each engine's workload, then the replica's routing surface. Every engine composes to a Deployment, LeaderWorkerSet, or PodCliqueSet (with its @@ -192,7 +241,7 @@ def compose_model_serving(self) -> None: composed: dict[str, k8sobjv1alpha1.Object] = {} for engine in self.xr.spec.engines: backend = _BACKENDS[base.select_backend(engine, stack)]() - composed.update(backend.build(self.xr, engine, pc, label, stack)) + composed.update(backend.build(self.xr, engine, pc, label, stack, mount)) composed = routing.apply(composed, self.xr, pc) for key, obj in composed.items(): resource.update(self.rsp.desired.resources[key], obj) diff --git a/functions/compose-model-replica/tests/test_backends.py b/functions/compose-model-replica/tests/test_backends.py index 743d86430..74dce9d2e 100644 --- a/functions/compose-model-replica/tests/test_backends.py +++ b/functions/compose-model-replica/tests/test_backends.py @@ -31,6 +31,7 @@ from crossplane.function.proto.v1 import run_function_pb2 as fnv1 from function import routing from function.backends import base, grove, llmd, native +from models.ai.modelplane.modelcache import v1alpha1 as mcv1alpha1 from models.ai.modelplane.modelreplica import v1alpha1 from models.io.crossplane.m.kubernetes.object import v1alpha1 as k8sobjv1alpha1 from models.io.k8s.apimachinery.pkg.apis.meta import v1 as metav1 @@ -47,6 +48,7 @@ class BuildCase: provider_config: str serving_label: str stack: str + cache: mcv1alpha1.Mount | None want: dict[str, dict] @@ -67,7 +69,7 @@ class CacheMountsCase: name: str reason: str - replica: v1alpha1.ModelReplica + cache: mcv1alpha1.Mount | None want: tuple[list[dict], list[dict]] @@ -77,10 +79,22 @@ class CacheEnvCase: name: str reason: str - replica: v1alpha1.ModelReplica + cache: mcv1alpha1.Mount | None want: list[dict] +@dataclasses.dataclass +class ModelExpressSecurityContextCase: + """A test case for base.modelexpress_security_context.""" + + name: str + reason: str + replica: v1alpha1.ModelReplica + stack: str + cache: mcv1alpha1.Mount | None + want: dict | None + + @dataclasses.dataclass class ApplyCase: """A test case for routing.apply.""" @@ -695,6 +709,19 @@ def _replica( ) +def _mount(*, claim_name: str) -> mcv1alpha1.Mount: + """The mount a HuggingFace ModelCache publishes for a cluster: its claim at /mnt/models, found by repo ID.""" + return mcv1alpha1.Mount( + volumes=[ + mcv1alpha1.Volume( + name="model-cache", persistentVolumeClaim=mcv1alpha1.PersistentVolumeClaim(claimName=claim_name) + ) + ], + volumeMounts=[mcv1alpha1.VolumeMount(name="model-cache", mountPath="/mnt/models")], + env=[mcv1alpha1.EnvItem(name="HF_HUB_CACHE", value="/mnt/models")], + ) + + def _to_dicts(*, objects: dict[str, k8sobjv1alpha1.Object]) -> dict[str, dict]: """objects as dicts of only the fields the code set.""" return {key: obj.model_dump(exclude_unset=True, by_alias=True) for key, obj in objects.items()} @@ -762,6 +789,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -859,6 +887,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -968,6 +997,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -1069,6 +1099,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -1162,6 +1193,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -1250,6 +1282,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="dep-clusterA", stack="Standard", + cache=None, want={ "model-serving-main": _deployment( name="dep-clusterA-main-3d1d5", @@ -1339,6 +1372,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="dep-clusterB", stack="Standard", + cache=None, want={ "model-serving-main": _deployment( name="dep-clusterB-main-d6c52", @@ -1463,6 +1497,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-prefill": _deployment( name="r-prefill-d90b0", @@ -1575,6 +1610,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -1694,6 +1730,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=_mount(claim_name="modelcache-ml-team-qwen-17db2"), want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -1791,6 +1828,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=_mount(claim_name="modelcache-ml-team-qwen-17db2"), want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -1897,6 +1935,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=_mount(claim_name="modelcache-ml-team-qwen-17db2"), want={ "model-serving-main": _deployment( name="r-main-bb4e3", @@ -2070,6 +2109,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=None, want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -2289,6 +2329,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=None, want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -2503,6 +2544,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=None, want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -2720,6 +2762,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=None, want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -2942,6 +2985,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=None, want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -3145,6 +3189,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=None, want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -3346,6 +3391,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=None, want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -3545,6 +3591,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=_mount(claim_name="modelcache-ml-team-kimi-aa322"), want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -3813,6 +3860,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=_mount(claim_name="modelcache-ml-team-kimi-aa322"), want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -4092,6 +4140,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Dynamo", + cache=_mount(claim_name="modelcache-ml-team-qwen-17db2"), want={ "scaling-group-main": _scaling_group(pcs_name="r-main-bb4e3"), "model-serving-main": _pod_clique_set( @@ -4359,6 +4408,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _leader_worker_set( name="r-main-bb4e3", @@ -4483,6 +4533,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="qwen2.5", stack="Standard", + cache=None, want={ "model-serving-main": _leader_worker_set( name="lws-qwen2-5-main-b2784-6d975", @@ -4610,6 +4661,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _leader_worker_set( name="r-main-bb4e3", @@ -4739,6 +4791,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _leader_worker_set( name="r-main-bb4e3", @@ -4885,6 +4938,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _leader_worker_set( name="r-main-bb4e3", @@ -5013,6 +5067,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=None, want={ "model-serving-main": _leader_worker_set( name="r-main-bb4e3", @@ -5134,6 +5189,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: provider_config="cluster-a-pc", serving_label="r", stack="Standard", + cache=_mount(claim_name="modelcache-ml-team-c-c5cc6"), want={ "model-serving-main": _leader_worker_set( name="r-main-bb4e3", @@ -5215,7 +5271,9 @@ def test_build(case: BuildCase) -> None: # build composes one engine, so build each of the replica's engines in turn. got: dict[str, k8sobjv1alpha1.Object] = {} for engine in case.replica.spec.engines: - got.update(case.backend.build(case.replica, engine, case.provider_config, case.serving_label, case.stack)) + got.update( + case.backend.build(case.replica, engine, case.provider_config, case.serving_label, case.stack, case.cache) + ) assert _sorted(objects=_to_dicts(objects=got)) == _sorted(objects=case.want), case.reason @@ -5419,54 +5477,85 @@ def test_select_backend(case: SelectBackendCase) -> None: CacheMountsCase( name="NoCache", reason="A replica with no cache gets no volume or mount.", - replica=_replica( - name=None, - namespace="ml-team", - cluster_name="c", - labels=None, - model_cache_ref=None, - serving=None, - engines=[ - v1alpha1.Engine( - name="main", - copies=1, - members=[ - v1alpha1.Member( - role="Standalone", - nodePoolName="frontier", - deviceRequests=[ - v1alpha1.DeviceRequest( - name="gpu", - deviceClassName="gpu.nvidia.com", - count=1, - selectors=[ - v1alpha1.Selector( - cel='device.capacity["gpu.nvidia.com"].memory.compareTo(quantity("80Gi")) >= 0' - ) - ], - ) - ], - template=v1alpha1.Template( - spec=v1alpha1.Spec( - containers=[ - v1alpha1.Container( - name="engine", - image="vllm/vllm-openai:latest", - args=[], - ) - ] - ) - ), - ) - ], - ) - ], - ), + cache=None, want=([], []), ), CacheMountsCase( name="WithCache", reason="A replica with a cache gets a volume for its PVC and a mount for it.", + cache=_mount(claim_name="modelcache-ml-team-qwen-17db2"), + want=( + [{"name": "model-cache", "persistentVolumeClaim": {"claimName": "modelcache-ml-team-qwen-17db2"}}], + [{"name": "model-cache", "mountPath": "/mnt/models"}], + ), + ), + CacheMountsCase( + name="CopiesTheCacheMount", + reason="A replica copies the volumes and mounts its cache publishes, including a read-only mount.", + cache=mcv1alpha1.Mount( + volumes=[ + mcv1alpha1.Volume( + name="weights", persistentVolumeClaim=mcv1alpha1.PersistentVolumeClaim(claimName="filled") + ) + ], + volumeMounts=[mcv1alpha1.VolumeMount(name="weights", mountPath="/mnt/models", readOnly=True)], + ), + want=( + [{"name": "weights", "persistentVolumeClaim": {"claimName": "filled"}}], + [{"name": "weights", "mountPath": "/mnt/models", "readOnly": True}], + ), + ), +] + + +@pytest.mark.parametrize("case", CACHE_MOUNTS_CASES, ids=lambda case: case.name) +def test_cache_mounts(case: CacheMountsCase) -> None: + """A replica's cache contributes a volume and a mount.""" + assert base.cache_mounts(case.cache) == case.want, case.reason + + +CACHE_ENV_CASES = [ + CacheEnvCase( + name="NoCache", + reason="A replica with no cache gets no env.", + cache=None, + want=[], + ), + # The cache is staged in HuggingFace's cache layout, so pointing + # HF_HUB_CACHE at the mount is what lets an engine's own --model= + # resolve against it instead of pulling from HuggingFace (#407). There's no + # HF_HUB_OFFLINE: it would break an engine that fetches a different repo at + # startup, like kimi-k2's separately gated tokenizer, and resolution doesn't + # need it. + CacheEnvCase( + name="WithCache", + reason="A replica with a cache gets HF_HUB_CACHE pointing at the mount, and no HF_HUB_OFFLINE.", + cache=_mount(claim_name="modelcache-ml-team-qwen-17db2"), + want=[{"name": "HF_HUB_CACHE", "value": "/mnt/models"}], + ), + CacheEnvCase( + name="CopiesTheCacheEnv", + reason="A replica copies every variable its cache publishes, in order.", + cache=mcv1alpha1.Mount( + env=[mcv1alpha1.EnvItem(name="HF_HUB_CACHE", value="/mnt/models"), mcv1alpha1.EnvItem(name="A", value="b")] + ), + want=[{"name": "HF_HUB_CACHE", "value": "/mnt/models"}, {"name": "A", "value": "b"}], + ), +] + + +@pytest.mark.parametrize("case", CACHE_ENV_CASES, ids=lambda case: case.name) +def test_cache_env(case: CacheEnvCase) -> None: + """A replica's cache contributes the env that resolves a model against it.""" + assert base.cache_env(case.cache) == case.want, case.reason + + +# modelexpress_security_context reads only a replica's cache reference, so the +# replicas here have no name and a placeholder cluster, c. +MODELEXPRESS_SECURITY_CONTEXT_CASES = [ + ModelExpressSecurityContextCase( + name="MountedOnDynamo", + reason="An engine that mounts its cache on a Dynamo cluster gets IPC_LOCK.", replica=_replica( name=None, namespace="ml-team", @@ -5477,32 +5566,13 @@ def test_select_backend(case: SelectBackendCase) -> None: engines=[ v1alpha1.Engine( name="main", - copies=1, members=[ v1alpha1.Member( role="Standalone", nodePoolName="frontier", - deviceRequests=[ - v1alpha1.DeviceRequest( - name="gpu", - deviceClassName="gpu.nvidia.com", - count=1, - selectors=[ - v1alpha1.Selector( - cel='device.capacity["gpu.nvidia.com"].memory.compareTo(quantity("80Gi")) >= 0' - ) - ], - ) - ], template=v1alpha1.Template( spec=v1alpha1.Spec( - containers=[ - v1alpha1.Container( - name="engine", - image="vllm/vllm-openai:latest", - args=[], - ) - ] + containers=[v1alpha1.Container(name="engine", image="vllm/vllm-openai:latest")] ) ), ) @@ -5510,62 +5580,32 @@ def test_select_backend(case: SelectBackendCase) -> None: ) ], ), - want=( - [{"name": "model-cache", "persistentVolumeClaim": {"claimName": "modelcache-ml-team-qwen-17db2"}}], - [{"name": "model-cache", "mountPath": "/mnt/models"}], - ), + stack="Dynamo", + cache=_mount(claim_name="modelcache-ml-team-qwen-17db2"), + want={"capabilities": {"add": ["IPC_LOCK"]}}, ), -] - - -@pytest.mark.parametrize("case", CACHE_MOUNTS_CASES, ids=lambda case: case.name) -def test_cache_mounts(case: CacheMountsCase) -> None: - """A replica's cache contributes a volume and a mount.""" - assert base.cache_mounts(case.replica) == case.want, case.reason - - -# cache_env reads only whether a replica has a cache, so the replicas here have -# no name and a placeholder cluster, c. -CACHE_ENV_CASES = [ - CacheEnvCase( - name="NoCache", - reason="A replica with no cache gets no env.", + # A running replica that gains a cache reference keeps its engines as they + # are until the cache is Ready, and adding a capability would roll them. + ModelExpressSecurityContextCase( + name="NotMountedOnDynamo", + reason="An engine that references a cache it doesn't mount yet gets no IPC_LOCK.", replica=_replica( name=None, namespace="ml-team", cluster_name="c", labels=None, - model_cache_ref=None, + model_cache_ref=v1alpha1.ModelCacheRef(name="qwen"), serving=None, engines=[ v1alpha1.Engine( name="main", - copies=1, members=[ v1alpha1.Member( role="Standalone", nodePoolName="frontier", - deviceRequests=[ - v1alpha1.DeviceRequest( - name="gpu", - deviceClassName="gpu.nvidia.com", - count=1, - selectors=[ - v1alpha1.Selector( - cel='device.capacity["gpu.nvidia.com"].memory.compareTo(quantity("80Gi")) >= 0' - ) - ], - ) - ], template=v1alpha1.Template( spec=v1alpha1.Spec( - containers=[ - v1alpha1.Container( - name="engine", - image="vllm/vllm-openai:latest", - args=[], - ) - ] + containers=[v1alpha1.Container(name="engine", image="vllm/vllm-openai:latest")] ) ), ) @@ -5573,17 +5613,13 @@ def test_cache_mounts(case: CacheMountsCase) -> None: ) ], ), - want=[], + stack="Dynamo", + cache=None, + want=None, ), - # The cache is staged in HuggingFace's cache layout, so pointing - # HF_HUB_CACHE at the mount is what lets an engine's own --model= - # resolve against it instead of pulling from HuggingFace (#407). There's no - # HF_HUB_OFFLINE: it would break an engine that fetches a different repo at - # startup, like kimi-k2's separately gated tokenizer, and resolution doesn't - # need it. - CacheEnvCase( - name="WithCache", - reason="A replica with a cache gets HF_HUB_CACHE pointing at the mount, and no HF_HUB_OFFLINE.", + ModelExpressSecurityContextCase( + name="MountedOnStandard", + reason="An engine that mounts its cache off Dynamo gets no IPC_LOCK.", replica=_replica( name=None, namespace="ml-team", @@ -5594,32 +5630,13 @@ def test_cache_mounts(case: CacheMountsCase) -> None: engines=[ v1alpha1.Engine( name="main", - copies=1, members=[ v1alpha1.Member( role="Standalone", nodePoolName="frontier", - deviceRequests=[ - v1alpha1.DeviceRequest( - name="gpu", - deviceClassName="gpu.nvidia.com", - count=1, - selectors=[ - v1alpha1.Selector( - cel='device.capacity["gpu.nvidia.com"].memory.compareTo(quantity("80Gi")) >= 0' - ) - ], - ) - ], template=v1alpha1.Template( spec=v1alpha1.Spec( - containers=[ - v1alpha1.Container( - name="engine", - image="vllm/vllm-openai:latest", - args=[], - ) - ] + containers=[v1alpha1.Container(name="engine", image="vllm/vllm-openai:latest")] ) ), ) @@ -5627,15 +5644,17 @@ def test_cache_mounts(case: CacheMountsCase) -> None: ) ], ), - want=[{"name": "HF_HUB_CACHE", "value": "/mnt/models"}], + stack="Standard", + cache=_mount(claim_name="modelcache-ml-team-qwen-17db2"), + want=None, ), ] -@pytest.mark.parametrize("case", CACHE_ENV_CASES, ids=lambda case: case.name) -def test_cache_env(case: CacheEnvCase) -> None: - """A replica's cache contributes the env that resolves a model against it.""" - assert base.cache_env(case.replica) == case.want, case.reason +@pytest.mark.parametrize("case", MODELEXPRESS_SECURITY_CONTEXT_CASES, ids=lambda case: case.name) +def test_modelexpress_security_context(case: ModelExpressSecurityContextCase) -> None: + """An engine that mounts its cache on a Dynamo cluster can pin memory for GPUDirect RDMA.""" + assert base.modelexpress_security_context(case.replica, case.stack, case.cache) == case.want, case.reason APPLY_CASES = [ diff --git a/functions/compose-model-replica/tests/test_cache.py b/functions/compose-model-replica/tests/test_cache.py new file mode 100644 index 000000000..e7ae08365 --- /dev/null +++ b/functions/compose-model-replica/tests/test_cache.py @@ -0,0 +1,154 @@ +# Copyright 2026 The Modelplane Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for compose-model-replica's reading of a replica's ModelCache.""" + +import dataclasses + +import pytest +from function import cache +from models.ai.modelplane.modelcache import v1alpha1 as mcv1alpha1 + + +@dataclasses.dataclass +class MountedCase: + """A test case for cache.mounted.""" + + name: str + reason: str + manifest: dict + want: mcv1alpha1.Mount | None + + +def _pod_spec(*, claim_name: str | None) -> dict: + """An engine pod's spec, mounting the named cache claim if any, beside the shared memory every engine mounts.""" + engine: dict = { + "name": "engine", + "image": "vllm/vllm-openai:latest", + "volumeMounts": [{"name": "dshm", "mountPath": "/dev/shm"}], + } + volumes: list[dict] = [{"name": "dshm", "emptyDir": {"medium": "Memory"}}] + if claim_name is not None: + engine["env"] = [ + {"name": "HF_HUB_CACHE", "value": "/mnt/models"}, + {"name": "MX_MODEL_REVISION", "value": claim_name}, + ] + engine["volumeMounts"].append({"name": "model-cache", "mountPath": "/mnt/models"}) + volumes.append({"name": "model-cache", "persistentVolumeClaim": {"claimName": claim_name}}) + return {"containers": [{"name": "sidecar", "image": "busybox:1.36"}, engine], "volumes": volumes} + + +def _mount(*, claim_name: str) -> mcv1alpha1.Mount: + """The mount a HuggingFace ModelCache publishes, reading the named claim.""" + return mcv1alpha1.Mount( + volumes=[ + mcv1alpha1.Volume( + name="model-cache", persistentVolumeClaim=mcv1alpha1.PersistentVolumeClaim(claimName=claim_name) + ) + ], + volumeMounts=[mcv1alpha1.VolumeMount(name="model-cache", mountPath="/mnt/models")], + env=[mcv1alpha1.EnvItem(name="HF_HUB_CACHE", value="/mnt/models")], + ) + + +MOUNTED_CASES = [ + MountedCase( + name="Deployment", + reason="A Deployment's pod template mounts the cache its replica composed it with.", + manifest={ + "kind": "Deployment", + "spec": {"template": {"spec": _pod_spec(claim_name="modelcache-ml-team-qwen-17db2")}}, + }, + want=_mount(claim_name="modelcache-ml-team-qwen-17db2"), + ), + MountedCase( + name="LeaderWorkerSet", + reason="A LeaderWorkerSet's leader and worker templates mount the cache its replica composed it with.", + manifest={ + "kind": "LeaderWorkerSet", + "spec": { + "leaderWorkerTemplate": { + "leaderTemplate": {"spec": _pod_spec(claim_name="modelcache-ml-team-kimi-aa322")}, + "workerTemplate": {"spec": _pod_spec(claim_name="modelcache-ml-team-kimi-aa322")}, + } + }, + }, + want=_mount(claim_name="modelcache-ml-team-kimi-aa322"), + ), + MountedCase( + name="PodCliqueSet", + reason="A PodCliqueSet's cliques mount the cache its replica composed it with.", + manifest={ + "kind": "PodCliqueSet", + "spec": { + "template": { + "cliques": [ + {"name": "leader", "spec": {"podSpec": _pod_spec(claim_name="modelcache-ml-team-c-c5cc6")}}, + {"name": "worker", "spec": {"podSpec": _pod_spec(claim_name="modelcache-ml-team-c-c5cc6")}}, + ] + } + }, + }, + want=_mount(claim_name="modelcache-ml-team-c-c5cc6"), + ), + MountedCase( + name="EngineSetsCacheEnv", + reason="An engine's own HF_HUB_CACHE follows the cache's, and isn't part of the cache's mount.", + manifest={ + "kind": "Deployment", + "spec": { + "template": { + "spec": { + "containers": [ + { + "name": "engine", + "image": "vllm/vllm-openai:latest", + "env": [ + {"name": "HF_HUB_CACHE", "value": "/mnt/models"}, + {"name": "HF_HUB_CACHE", "value": "/mnt/models/hub"}, + ], + "volumeMounts": [{"name": "model-cache", "mountPath": "/mnt/models"}], + } + ], + "volumes": [ + { + "name": "model-cache", + "persistentVolumeClaim": {"claimName": "modelcache-ml-team-qwen-17db2"}, + } + ], + } + } + }, + }, + want=_mount(claim_name="modelcache-ml-team-qwen-17db2"), + ), + MountedCase( + name="NoCache", + reason="A workload whose pods mount no cache has no cache mount.", + manifest={"kind": "Deployment", "spec": {"template": {"spec": _pod_spec(claim_name=None)}}}, + want=None, + ), + MountedCase( + name="NoPods", + reason="A workload manifest with no pod template, as an observe-only scaling group has, has no cache mount.", + manifest={"kind": "PodCliqueScalingGroup"}, + want=None, + ), +] + + +@pytest.mark.parametrize("case", MOUNTED_CASES, ids=lambda case: case.name) +def test_mounted(case: MountedCase) -> None: + """A replica reads the cache volume, mount and env its workload's pods already have.""" + assert cache.mounted(case.manifest) == case.want, case.reason diff --git a/functions/compose-model-replica/tests/test_fn.py b/functions/compose-model-replica/tests/test_fn.py index c2fd025c7..fabc3742a 100644 --- a/functions/compose-model-replica/tests/test_fn.py +++ b/functions/compose-model-replica/tests/test_fn.py @@ -21,6 +21,7 @@ import asyncio import dataclasses import json +from typing import Any import pytest from crossplane.function import resource @@ -43,8 +44,8 @@ class Case: want: fnv1.RunFunctionResponse -def _model_replica(*, copies: int) -> fnv1.Resource: - """The ModelReplica XR test-replica in ml-team, with one Standalone engine. +def _model_replica(*, copies: int, model_cache_ref: str | None) -> fnv1.Resource: + """The ModelReplica XR test-replica in ml-team, with one Standalone engine, reading the named ModelCache if any. The device request's CEL selector is as compose-model-deployment stamps it. """ @@ -59,6 +60,7 @@ def _model_replica(*, copies: int) -> fnv1.Resource: ), spec=v1alpha1.SpecModel1( clusterName="cluster-a", + modelCacheRef=v1alpha1.ModelCacheRef(name=model_cache_ref) if model_cache_ref else None, engines=[ v1alpha1.Engine( name="main", @@ -210,12 +212,33 @@ def _cluster(*, provider_config_ref: str | None) -> fnv1.Resource: return fnv1.Resource(resource=resource.dict_to_struct(cluster)) -def _deployment(*, ready: fnv1.Ready, name: str, claim_template_name: str, labels: dict, copies: int) -> fnv1.Resource: - """The Object composing an engine's Deployment. +def _deployment( + *, ready: fnv1.Ready, name: str, claim_template_name: str, labels: dict, copies: int, cache_claim: str | None +) -> fnv1.Resource: + """The Object composing an engine's Deployment, mounting the cache claim if any. Its pod carries the labels the collector attributes the engine's metrics by: the deployment, the engine and the member's role. """ + engine: dict[str, Any] = { + "name": "engine", + "image": "vllm/vllm-openai:latest", + "args": ["--model=Qwen/Qwen3-0.6B"], + "ports": [{"name": "http", "containerPort": 8000}], + "volumeMounts": [{"name": "dshm", "mountPath": "/dev/shm"}], + "readinessProbe": { + "httpGet": {"path": "/health", "port": 8000}, + "initialDelaySeconds": 30, + "periodSeconds": 10, + "timeoutSeconds": 5, + }, + "resources": {"claims": [{"name": "devices"}]}, + } + volumes: list[dict] = [{"name": "dshm", "emptyDir": {"medium": "Memory"}}] + if cache_claim is not None: + engine["volumeMounts"].append({"name": "model-cache", "mountPath": "/mnt/models"}) + engine["env"] = [{"name": "HF_HUB_CACHE", "value": "/mnt/models"}] + volumes.append({"name": "model-cache", "persistentVolumeClaim": {"claimName": cache_claim}}) return fnv1.Resource( resource=resource.dict_to_struct( { @@ -242,23 +265,8 @@ def _deployment(*, ready: fnv1.Ready, name: str, claim_template_name: str, label "template": { "metadata": {"labels": labels}, "spec": { - "containers": [ - { - "name": "engine", - "image": "vllm/vllm-openai:latest", - "args": ["--model=Qwen/Qwen3-0.6B"], - "ports": [{"name": "http", "containerPort": 8000}], - "volumeMounts": [{"name": "dshm", "mountPath": "/dev/shm"}], - "readinessProbe": { - "httpGet": {"path": "/health", "port": 8000}, - "initialDelaySeconds": 30, - "periodSeconds": 10, - "timeoutSeconds": 5, - }, - "resources": {"claims": [{"name": "devices"}]}, - } - ], - "volumes": [{"name": "dshm", "emptyDir": {"medium": "Memory"}}], + "containers": [engine], + "volumes": volumes, "nodeSelector": {"modelplane.ai/pool": "frontier"}, "resourceClaims": [ { @@ -695,8 +703,8 @@ def _epp_service(*, ready: fnv1.Ready) -> fnv1.Resource: ) -def _observed_deployment(*, ready: bool, ready_replicas: int) -> fnv1.Resource: - """The engine's Deployment Object as observed back. +def _observed_deployment(*, ready: bool, ready_replicas: int, cache_claim: str | None) -> fnv1.Resource: + """The engine's Deployment Object as observed back, its pods mounting the cache claim if any. ready sets the Object's derived Ready condition, and ready_replicas the Deployment's status.readyReplicas, which a Deployment leaves out at 0. @@ -704,12 +712,26 @@ def _observed_deployment(*, ready: bool, ready_replicas: int) -> fnv1.Resource: observed_manifest: dict = {"kind": "Deployment"} if ready_replicas: observed_manifest["status"] = {"readyReplicas": ready_replicas} + engine: dict[str, Any] = { + "name": "engine", + "image": "vllm/vllm-openai:latest", + "volumeMounts": [{"name": "dshm", "mountPath": "/dev/shm"}], + } + volumes: list[dict] = [{"name": "dshm", "emptyDir": {"medium": "Memory"}}] + if cache_claim is not None: + engine["volumeMounts"].append({"name": "model-cache", "mountPath": "/mnt/models"}) + engine["env"] = [{"name": "HF_HUB_CACHE", "value": "/mnt/models"}] + volumes.append({"name": "model-cache", "persistentVolumeClaim": {"claimName": cache_claim}}) + applied_manifest = { + "kind": "Deployment", + "spec": {"template": {"spec": {"containers": [engine], "volumes": volumes}}}, + } return fnv1.Resource( resource=resource.dict_to_struct( { "apiVersion": "kubernetes.m.crossplane.io/v1alpha1", "kind": "Object", - "spec": {"forProvider": {"manifest": {"kind": "Deployment"}}}, + "spec": {"forProvider": {"manifest": applied_manifest}}, "status": { "atProvider": {"manifest": observed_manifest}, "conditions": [ @@ -752,6 +774,30 @@ def _observed_object(*, ready: bool) -> fnv1.Resource: ) +def _model_cache(*, clusters: list[dict]) -> fnv1.Resource: + """The ModelCache qwen in ml-team, reporting clusters as its per-cluster status.""" + return fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "modelplane.ai/v1alpha1", + "kind": "ModelCache", + "metadata": {"name": "qwen", "namespace": "ml-team"}, + "spec": {"source": "HuggingFace", "huggingFace": {"repo": "Qwen/Qwen3-0.6B", "sizeGiB": 20}}, + "status": {"clusters": clusters}, + } + ) + ) + + +def _cache_mount(*, claim_name: str) -> dict: + """The mount a HuggingFace ModelCache publishes for a cluster it has staged the model on.""" + return { + "volumes": [{"name": "model-cache", "persistentVolumeClaim": {"claimName": claim_name}}], + "volumeMounts": [{"name": "model-cache", "mountPath": "/mnt/models"}], + "env": [{"name": "HF_HUB_CACHE", "value": "/mnt/models"}], + } + + def _to_dict(msg: message.Message) -> dict: """msg as a dict with sorted keys, so pytest's diff of two lines them up.""" return json.loads(json_format.MessageToJson(msg, sort_keys=True)) @@ -771,8 +817,8 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_prefill_decode_replica(serving=v1alpha1.Serving(mode="PrefillDecode")), resources={ - "model-serving-prefill": _observed_deployment(ready=True, ready_replicas=1), - "model-serving-decode": _observed_deployment(ready=True, ready_replicas=1), + "model-serving-prefill": _observed_deployment(ready=True, ready_replicas=1, cache_claim=None), + "model-serving-decode": _observed_deployment(ready=True, ready_replicas=1, cache_claim=None), }, ), required_resources={"cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")])}, @@ -818,8 +864,8 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_prefill_decode_replica(serving=v1alpha1.Serving(mode="PrefillDecode")), resources={ - "model-serving-prefill": _observed_deployment(ready=True, ready_replicas=1), - "model-serving-decode": _observed_deployment(ready=False, ready_replicas=0), + "model-serving-prefill": _observed_deployment(ready=True, ready_replicas=1, cache_claim=None), + "model-serving-decode": _observed_deployment(ready=False, ready_replicas=0, cache_claim=None), }, ), required_resources={"cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")])}, @@ -948,6 +994,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-prefill": _deployment( + cache_claim=None, copies=1, ready=fnv1.READY_UNSPECIFIED, name="test-replica-prefill-10d7e", @@ -966,6 +1013,7 @@ def _to_dict(msg: message.Message) -> dict: ready=fnv1.READY_UNSPECIFIED, name="test-replica-prefill-standalone-devices-51762" ), "model-serving-decode": _deployment( + cache_claim=None, copies=1, ready=fnv1.READY_UNSPECIFIED, name="test-replica-decode-44185", @@ -1067,7 +1115,7 @@ def _to_dict(msg: message.Message) -> dict: "endpoint picker, and marks none of it ready." ), req=fnv1.RunFunctionRequest( - observed=fnv1.State(composite=_model_replica(copies=1)), + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref=None)), required_resources={"cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")])}, ), want=fnv1.RunFunctionResponse( @@ -1075,6 +1123,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + cache_claim=None, copies=1, ready=fnv1.READY_UNSPECIFIED, name="test-replica-main-6b608", @@ -1137,7 +1186,7 @@ def _to_dict(msg: message.Message) -> dict: name="ClusterUnresolved", reason="Before its cluster resolves, a replica composes nothing and reports waiting, with the XR not ready.", req=fnv1.RunFunctionRequest( - observed=fnv1.State(composite=_model_replica(copies=1)), + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref=None)), ), want=fnv1.RunFunctionResponse( meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), @@ -1172,7 +1221,7 @@ def _to_dict(msg: message.Message) -> dict: "not ready." ), req=fnv1.RunFunctionRequest( - observed=fnv1.State(composite=_model_replica(copies=1)), + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref=None)), required_resources={"cluster": fnv1.Resources(items=[_cluster(provider_config_ref=None)])}, ), want=fnv1.RunFunctionResponse( @@ -1213,9 +1262,9 @@ def _to_dict(msg: message.Message) -> dict: ), req=fnv1.RunFunctionRequest( observed=fnv1.State( - composite=_model_replica(copies=1), + composite=_model_replica(copies=1, model_cache_ref=None), resources={ - "model-serving-main": _observed_deployment(ready=True, ready_replicas=1), + "model-serving-main": _observed_deployment(ready=True, ready_replicas=1, cache_claim=None), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), }, @@ -1227,6 +1276,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + cache_claim=None, copies=1, ready=fnv1.READY_TRUE, name="test-replica-main-6b608", @@ -1288,9 +1338,9 @@ def _to_dict(msg: message.Message) -> dict: ), req=fnv1.RunFunctionRequest( observed=fnv1.State( - composite=_model_replica(copies=1), + composite=_model_replica(copies=1, model_cache_ref=None), resources={ - "model-serving-main": _observed_deployment(ready=True, ready_replicas=1), + "model-serving-main": _observed_deployment(ready=True, ready_replicas=1, cache_claim=None), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1309,6 +1359,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + cache_claim=None, copies=1, ready=fnv1.READY_TRUE, name="test-replica-main-6b608", @@ -1370,9 +1421,9 @@ def _to_dict(msg: message.Message) -> dict: ), req=fnv1.RunFunctionRequest( observed=fnv1.State( - composite=_model_replica(copies=1), + composite=_model_replica(copies=1, model_cache_ref=None), resources={ - "model-serving-main": _observed_deployment(ready=True, ready_replicas=1), + "model-serving-main": _observed_deployment(ready=True, ready_replicas=1, cache_claim=None), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1391,6 +1442,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + cache_claim=None, copies=1, ready=fnv1.READY_TRUE, name="test-replica-main-6b608", @@ -1449,9 +1501,9 @@ def _to_dict(msg: message.Message) -> dict: ), req=fnv1.RunFunctionRequest( observed=fnv1.State( - composite=_model_replica(copies=2), + composite=_model_replica(copies=2, model_cache_ref=None), resources={ - "model-serving-main": _observed_deployment(ready=False, ready_replicas=1), + "model-serving-main": _observed_deployment(ready=False, ready_replicas=1, cache_claim=None), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1470,6 +1522,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + cache_claim=None, ready=fnv1.READY_UNSPECIFIED, copies=2, name="test-replica-main-6b608", @@ -1525,9 +1578,9 @@ def _to_dict(msg: message.Message) -> dict: reason=("With no copy ready, the replica has started but isn't serving, whatever its routing."), req=fnv1.RunFunctionRequest( observed=fnv1.State( - composite=_model_replica(copies=2), + composite=_model_replica(copies=2, model_cache_ref=None), resources={ - "model-serving-main": _observed_deployment(ready=False, ready_replicas=0), + "model-serving-main": _observed_deployment(ready=False, ready_replicas=0, cache_claim=None), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1546,6 +1599,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + cache_claim=None, ready=fnv1.READY_UNSPECIFIED, copies=2, name="test-replica-main-6b608", @@ -1604,9 +1658,9 @@ def _to_dict(msg: message.Message) -> dict: ), req=fnv1.RunFunctionRequest( observed=fnv1.State( - composite=_model_replica(copies=2), + composite=_model_replica(copies=2, model_cache_ref=None), resources={ - "model-serving-main": _observed_deployment(ready=False, ready_replicas=2), + "model-serving-main": _observed_deployment(ready=False, ready_replicas=2, cache_claim=None), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1625,6 +1679,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + cache_claim=None, ready=fnv1.READY_UNSPECIFIED, copies=2, name="test-replica-main-6b608", @@ -1680,9 +1735,9 @@ def _to_dict(msg: message.Message) -> dict: reason=("With both copies ready and the routing in place, the replica is serving at full capacity."), req=fnv1.RunFunctionRequest( observed=fnv1.State( - composite=_model_replica(copies=2), + composite=_model_replica(copies=2, model_cache_ref=None), resources={ - "model-serving-main": _observed_deployment(ready=True, ready_replicas=2), + "model-serving-main": _observed_deployment(ready=True, ready_replicas=2, cache_claim=None), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1701,6 +1756,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + cache_claim=None, ready=fnv1.READY_TRUE, copies=2, name="test-replica-main-6b608", @@ -1751,6 +1807,1351 @@ def _to_dict(msg: message.Message) -> dict: ], ), ), + Case( + name="CacheReady", + reason="A replica whose ModelCache is Ready on its cluster mounts the cache the way the ModelCache says to.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref="qwen")), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Ready", + "mount": _cache_mount(claim_name="modelcache-ml-team-qwen-17db2"), + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": _deployment( + cache_claim="modelcache-ml-team-qwen-17db2", + copies=1, + ready=fnv1.READY_UNSPECIFIED, + name="test-replica-main-6b608", + claim_template_name="test-replica-main-standalone-devices-e609d", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Standalone", + "modelplane.ai/serving": "test-replica", + "modelplane.ai/workload": "test-replica-main-6b608", + }, + ), + "model-route": _route(ready=fnv1.READY_UNSPECIFIED), + "resource-claim-main-standalone": _claim_template( + ready=fnv1.READY_UNSPECIFIED, name="test-replica-main-standalone-devices-e609d" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_UNSPECIFIED), + "epp": _epp( + ready=fnv1.READY_UNSPECIFIED, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_UNSPECIFIED), + "epp-role": _epp_role(ready=fnv1.READY_UNSPECIFIED), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_UNSPECIFIED), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_UNSPECIFIED), + "epp-service": _epp_service(ready=fnv1.READY_UNSPECIFIED), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_NORMAL, + message="Composing vllm/vllm-openai:latest on cluster-a", + ), + ], + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_FALSE, reason="Deploying"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForModel", + message="0 of 1 engine copies ready", + ), + ], + ), + ), + Case( + name="CacheHydrating", + reason="Before its ModelCache is Ready on its cluster, a new replica composes nothing and says why it waits.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref="qwen")), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Hydrating", + "message": "Downloading Qwen/Qwen3-0.6B into PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2", + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State(composite=fnv1.Resource(ready=fnv1.READY_FALSE)), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition( + type="ModelAccepted", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForCache", + message="ModelCache qwen is Hydrating on cluster cluster-a: Downloading Qwen/Qwen3-0.6B into PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2", + ), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + fnv1.Condition(type="ModelReady", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + ], + ), + ), + Case( + name="CacheNotOnCluster", + reason="A new replica on a cluster its ModelCache doesn't stage to composes nothing and says so.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref="qwen")), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-b", + "phase": "Ready", + "mount": _cache_mount(claim_name="modelcache-ml-team-qwen-17db2"), + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State(composite=fnv1.Resource(ready=fnv1.READY_FALSE)), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition( + type="ModelAccepted", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForCache", + message="ModelCache qwen isn't staged on cluster cluster-a", + ), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + fnv1.Condition(type="ModelReady", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + ], + ), + ), + Case( + name="CacheReadyWithoutMount", + reason="A new replica whose cache is Ready but hasn't published how to mount it waits for the mount.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref="qwen")), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Ready", + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State(composite=fnv1.Resource(ready=fnv1.READY_FALSE)), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition( + type="ModelAccepted", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForCache", + message="Waiting for ModelCache qwen to publish how to mount it on cluster cluster-a", + ), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + fnv1.Condition(type="ModelReady", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + ], + ), + ), + Case( + name="CacheNotFound", + reason="A new replica whose ModelCache doesn't exist composes nothing and says so.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref="qwen")), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources(items=[]), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State(composite=fnv1.Resource(ready=fnv1.READY_FALSE)), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition( + type="ModelAccepted", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForCache", + message="ModelCache qwen not found in namespace ml-team", + ), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + fnv1.Condition(type="ModelReady", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + ], + ), + ), + Case( + name="CacheUnresolved", + reason="Before its ModelCache resolves, a new replica composes nothing.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State(composite=_model_replica(copies=1, model_cache_ref="qwen")), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State(composite=fnv1.Resource(ready=fnv1.READY_FALSE)), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition( + type="ModelAccepted", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForCache", + message="Waiting for ModelCache qwen", + ), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + fnv1.Condition(type="ModelReady", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForModel"), + ], + ), + ), + Case( + name="CacheLostKeepsEngines", + reason="A replica whose engines already run keeps the cache they mount when the cache stops being Ready.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_replica(copies=1, model_cache_ref="qwen"), + resources={ + "model-serving-main": _observed_deployment( + ready=True, ready_replicas=1, cache_claim="modelcache-ml-team-qwen-17db2" + ), + "model-route": _observed_object(ready=True), + "resource-claim-main-standalone": _observed_object(ready=True), + }, + ), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Pending", + "message": "Waiting for PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2 to bind", + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": _deployment( + cache_claim="modelcache-ml-team-qwen-17db2", + copies=1, + ready=fnv1.READY_TRUE, + name="test-replica-main-6b608", + claim_template_name="test-replica-main-standalone-devices-e609d", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Standalone", + "modelplane.ai/serving": "test-replica", + "modelplane.ai/workload": "test-replica-main-6b608", + }, + ), + "model-route": _route(ready=fnv1.READY_TRUE), + "resource-claim-main-standalone": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-standalone-devices-e609d" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_UNSPECIFIED), + "epp": _epp( + ready=fnv1.READY_UNSPECIFIED, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_UNSPECIFIED), + "epp-role": _epp_role(ready=fnv1.READY_UNSPECIFIED), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_UNSPECIFIED), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_UNSPECIFIED), + "epp-service": _epp_service(ready=fnv1.READY_UNSPECIFIED), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_WARNING, + message=( + "ModelCache qwen publishes no mount for cluster cluster-a. The replica's engines keep the " + "cache they mount now." + ), + ), + ], + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_TRUE, reason="Accepted"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForRouting"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForRouting", + message="1 of 1 engine copies ready", + ), + ], + ), + ), + Case( + name="UpgradeKeepsCacheMount", + reason="A running replica whose Ready cache hasn't published a mount yet, as after an upgrade, composes its workload as before.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_replica(copies=1, model_cache_ref="qwen"), + resources={ + "model-serving-main": _observed_deployment( + ready=True, ready_replicas=1, cache_claim="modelcache-ml-team-qwen-17db2" + ), + "model-route": _observed_object(ready=True), + "resource-claim-main-standalone": _observed_object(ready=True), + }, + ), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Ready", + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": _deployment( + cache_claim="modelcache-ml-team-qwen-17db2", + copies=1, + ready=fnv1.READY_TRUE, + name="test-replica-main-6b608", + claim_template_name="test-replica-main-standalone-devices-e609d", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Standalone", + "modelplane.ai/serving": "test-replica", + "modelplane.ai/workload": "test-replica-main-6b608", + }, + ), + "model-route": _route(ready=fnv1.READY_TRUE), + "resource-claim-main-standalone": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-standalone-devices-e609d" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_UNSPECIFIED), + "epp": _epp( + ready=fnv1.READY_UNSPECIFIED, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_UNSPECIFIED), + "epp-role": _epp_role(ready=fnv1.READY_UNSPECIFIED), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_UNSPECIFIED), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_UNSPECIFIED), + "epp-service": _epp_service(ready=fnv1.READY_UNSPECIFIED), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_WARNING, + message=( + "ModelCache qwen publishes no mount for cluster cluster-a. The replica's engines keep the " + "cache they mount now." + ), + ), + ], + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_TRUE, reason="Accepted"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForRouting"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForRouting", + message="1 of 1 engine copies ready", + ), + ], + ), + ), + Case( + name="CacheAddedKeepsPods", + reason="A running replica that gains a cache that isn't Ready keeps its engines without one.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_replica(copies=1, model_cache_ref="qwen"), + resources={ + "model-serving-main": _observed_deployment(ready=True, ready_replicas=1, cache_claim=None), + "model-route": _observed_object(ready=True), + "resource-claim-main-standalone": _observed_object(ready=True), + }, + ), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Hydrating", + "message": "Downloading Qwen/Qwen3-0.6B into PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2", + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": _deployment( + cache_claim=None, + copies=1, + ready=fnv1.READY_TRUE, + name="test-replica-main-6b608", + claim_template_name="test-replica-main-standalone-devices-e609d", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Standalone", + "modelplane.ai/serving": "test-replica", + "modelplane.ai/workload": "test-replica-main-6b608", + }, + ), + "model-route": _route(ready=fnv1.READY_TRUE), + "resource-claim-main-standalone": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-standalone-devices-e609d" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_UNSPECIFIED), + "epp": _epp( + ready=fnv1.READY_UNSPECIFIED, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_UNSPECIFIED), + "epp-role": _epp_role(ready=fnv1.READY_UNSPECIFIED), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_UNSPECIFIED), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_UNSPECIFIED), + "epp-service": _epp_service(ready=fnv1.READY_UNSPECIFIED), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_WARNING, + message=( + "ModelCache qwen publishes no mount for cluster cluster-a. The replica's engines keep " + "running without a cache." + ), + ), + ], + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_TRUE, reason="Accepted"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForRouting"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForRouting", + message="1 of 1 engine copies ready", + ), + ], + ), + ), + Case( + name="CacheAddedThenReady", + reason="A running replica that gains a cache mounts it once the cache is Ready on its cluster.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_replica(copies=1, model_cache_ref="qwen"), + resources={ + "model-serving-main": _observed_deployment(ready=True, ready_replicas=1, cache_claim=None), + "model-route": _observed_object(ready=True), + "resource-claim-main-standalone": _observed_object(ready=True), + }, + ), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Ready", + "mount": _cache_mount(claim_name="modelcache-ml-team-qwen-17db2"), + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": _deployment( + cache_claim="modelcache-ml-team-qwen-17db2", + copies=1, + ready=fnv1.READY_TRUE, + name="test-replica-main-6b608", + claim_template_name="test-replica-main-standalone-devices-e609d", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Standalone", + "modelplane.ai/serving": "test-replica", + "modelplane.ai/workload": "test-replica-main-6b608", + }, + ), + "model-route": _route(ready=fnv1.READY_TRUE), + "resource-claim-main-standalone": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-standalone-devices-e609d" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_UNSPECIFIED), + "epp": _epp( + ready=fnv1.READY_UNSPECIFIED, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_UNSPECIFIED), + "epp-role": _epp_role(ready=fnv1.READY_UNSPECIFIED), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_UNSPECIFIED), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_UNSPECIFIED), + "epp-service": _epp_service(ready=fnv1.READY_UNSPECIFIED), + }, + ), + results=[], + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_TRUE, reason="Accepted"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForRouting"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForRouting", + message="1 of 1 engine copies ready", + ), + ], + ), + ), + Case( + name="SwitchedCacheKeepsMount", + reason="A running replica switched to a cache that isn't Ready keeps the cache its engines mount now.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_replica(copies=1, model_cache_ref="qwen"), + resources={ + "model-serving-main": _observed_deployment( + ready=True, ready_replicas=1, cache_claim="modelcache-ml-team-llama-0a1b2" + ), + "model-route": _observed_object(ready=True), + "resource-claim-main-standalone": _observed_object(ready=True), + }, + ), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Pending", + "message": "Waiting for PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-qwen-17db2 to bind", + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": _deployment( + cache_claim="modelcache-ml-team-llama-0a1b2", + copies=1, + ready=fnv1.READY_TRUE, + name="test-replica-main-6b608", + claim_template_name="test-replica-main-standalone-devices-e609d", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Standalone", + "modelplane.ai/serving": "test-replica", + "modelplane.ai/workload": "test-replica-main-6b608", + }, + ), + "model-route": _route(ready=fnv1.READY_TRUE), + "resource-claim-main-standalone": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-standalone-devices-e609d" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_UNSPECIFIED), + "epp": _epp( + ready=fnv1.READY_UNSPECIFIED, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_UNSPECIFIED), + "epp-role": _epp_role(ready=fnv1.READY_UNSPECIFIED), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_UNSPECIFIED), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_UNSPECIFIED), + "epp-service": _epp_service(ready=fnv1.READY_UNSPECIFIED), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_WARNING, + message=( + "ModelCache qwen publishes no mount for cluster cluster-a. The replica's engines keep the " + "cache they mount now." + ), + ), + ], + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_TRUE, reason="Accepted"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForRouting"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForRouting", + message="1 of 1 engine copies ready", + ), + ], + ), + ), + Case( + name="CacheDeletedKeepsMount", + reason="A running replica whose cache is deleted keeps the cache its engines mount now.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_replica(copies=1, model_cache_ref="qwen"), + resources={ + "model-serving-main": _observed_deployment( + ready=True, ready_replicas=1, cache_claim="modelcache-ml-team-qwen-17db2" + ), + "model-route": _observed_object(ready=True), + "resource-claim-main-standalone": _observed_object(ready=True), + }, + ), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources(items=[]), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": _deployment( + cache_claim="modelcache-ml-team-qwen-17db2", + copies=1, + ready=fnv1.READY_TRUE, + name="test-replica-main-6b608", + claim_template_name="test-replica-main-standalone-devices-e609d", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Standalone", + "modelplane.ai/serving": "test-replica", + "modelplane.ai/workload": "test-replica-main-6b608", + }, + ), + "model-route": _route(ready=fnv1.READY_TRUE), + "resource-claim-main-standalone": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-standalone-devices-e609d" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_UNSPECIFIED), + "epp": _epp( + ready=fnv1.READY_UNSPECIFIED, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_UNSPECIFIED), + "epp-role": _epp_role(ready=fnv1.READY_UNSPECIFIED), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_UNSPECIFIED), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_UNSPECIFIED), + "epp-service": _epp_service(ready=fnv1.READY_UNSPECIFIED), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_WARNING, + message=( + "ModelCache qwen publishes no mount for cluster cluster-a. The replica's engines keep the " + "cache they mount now." + ), + ), + ], + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="qwen", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_TRUE, reason="Accepted"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForRouting"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForRouting", + message="1 of 1 engine copies ready", + ), + ], + ), + ), + Case( + name="GangKeepsCacheMount", + reason="A running LeaderWorkerSet replica whose cache isn't Ready keeps the cache its leader and workers mount.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=fnv1.Resource( + resource=resource.dict_to_struct( + v1alpha1.ModelReplica( + metadata=metav1.ObjectMeta( + name="test-replica", + namespace="ml-team", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/cluster": "cluster-a", + }, + ), + spec=v1alpha1.SpecModel1( + clusterName="cluster-a", + modelCacheRef=v1alpha1.ModelCacheRef(name="kimi"), + engines=[ + v1alpha1.Engine( + name="main", + copies=1, + members=[ + v1alpha1.Member( + role="Leader", + nodePoolName="frontier", + deviceRequests=[ + v1alpha1.DeviceRequest( + name="gpu", + deviceClassName="gpu.nvidia.com", + count=1, + selectors=[ + v1alpha1.Selector( + cel='device.capacity["gpu.nvidia.com"].memory.compareTo(quantity("80Gi")) >= 0' + ) + ], + ) + ], + template=v1alpha1.Template( + spec=v1alpha1.Spec( + containers=[ + v1alpha1.Container( + name="engine", + image="vllm/vllm-openai:latest", + args=["--model=moonshotai/Kimi-K2-Instruct"], + ) + ] + ) + ), + ), + v1alpha1.Member( + role="Worker", + nodePoolName="frontier", + worker=v1alpha1.Worker(nodes=1), + deviceRequests=[ + v1alpha1.DeviceRequest( + name="gpu", + deviceClassName="gpu.nvidia.com", + count=1, + selectors=[ + v1alpha1.Selector( + cel='device.capacity["gpu.nvidia.com"].memory.compareTo(quantity("80Gi")) >= 0' + ) + ], + ) + ], + template=v1alpha1.Template( + spec=v1alpha1.Spec( + containers=[ + v1alpha1.Container( + name="engine", + image="vllm/vllm-openai:latest", + args=["--model=moonshotai/Kimi-K2-Instruct"], + ) + ] + ) + ), + ), + ], + ) + ], + ), + ).model_dump(exclude_none=True, mode="json", by_alias=True) + ) + ), + resources={ + # Only the pod templates of the LeaderWorkerSet it last + # composed matter here, so the rest is left out. + "model-serving-main": fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "kubernetes.m.crossplane.io/v1alpha1", + "kind": "Object", + "spec": { + "forProvider": { + "manifest": { + "apiVersion": "leaderworkerset.x-k8s.io/v1", + "kind": "LeaderWorkerSet", + "spec": { + "leaderWorkerTemplate": { + "leaderTemplate": { + "spec": { + "containers": [ + { + "name": "engine", + "env": [ + {"name": "HF_HUB_CACHE", "value": "/mnt/models"} + ], + "volumeMounts": [ + {"name": "dshm", "mountPath": "/dev/shm"}, + { + "name": "model-cache", + "mountPath": "/mnt/models", + }, + ], + } + ], + "volumes": [ + {"name": "dshm", "emptyDir": {"medium": "Memory"}}, + { + "name": "model-cache", + "persistentVolumeClaim": { + "claimName": "modelcache-ml-team-kimi-aa322" + }, + }, + ], + } + }, + "workerTemplate": { + "spec": { + "containers": [ + { + "name": "engine", + "env": [ + {"name": "HF_HUB_CACHE", "value": "/mnt/models"} + ], + "volumeMounts": [ + {"name": "dshm", "mountPath": "/dev/shm"}, + { + "name": "model-cache", + "mountPath": "/mnt/models", + }, + ], + } + ], + "volumes": [ + {"name": "dshm", "emptyDir": {"medium": "Memory"}}, + { + "name": "model-cache", + "persistentVolumeClaim": { + "claimName": "modelcache-ml-team-kimi-aa322" + }, + }, + ], + } + }, + } + }, + } + } + }, + "status": { + "atProvider": {"manifest": {"status": {"readyReplicas": 1}}}, + "conditions": [ + { + "type": "Ready", + "status": "True", + "reason": "Available", + "lastTransitionTime": "2025-01-01T00:00:00Z", + } + ], + }, + } + ) + ), + "model-route": _observed_object(ready=True), + "resource-claim-main-leader": _observed_object(ready=True), + "resource-claim-main-worker": _observed_object(ready=True), + }, + ), + required_resources={ + "cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")]), + "cache": fnv1.Resources( + items=[ + _model_cache( + clusters=[ + { + "name": "cluster-a", + "phase": "Hydrating", + "message": "Downloading moonshotai/Kimi-K2-Instruct into PersistentVolumeClaim mp-ml-team-51733/modelcache-ml-team-kimi-aa322", + } + ] + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "kubernetes.m.crossplane.io/v1alpha1", + "kind": "Object", + "spec": { + "forProvider": { + "manifest": { + "apiVersion": "leaderworkerset.x-k8s.io/v1", + "kind": "LeaderWorkerSet", + "metadata": { + "name": "test-replica-main-6b608", + "namespace": "mp-ml-team-51733", + }, + "spec": { + "leaderWorkerTemplate": { + "leaderTemplate": { + "metadata": { + "labels": { + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/lws-role": "leader", + "modelplane.ai/role": "Leader", + "modelplane.ai/serving": "test-replica", + } + }, + "spec": { + "containers": [ + { + "args": ["--model=moonshotai/Kimi-K2-Instruct"], + "env": [ + { + "name": "MODELPLANE_LEADER_ADDRESS", + "value": "$(LWS_LEADER_ADDRESS)", + }, + { + "name": "MODELPLANE_RANK", + "value": "$(LWS_WORKER_INDEX)", + }, + { + "name": "HF_HUB_CACHE", + "value": "/mnt/models", + }, + ], + "image": "vllm/vllm-openai:latest", + "name": "engine", + "ports": [{"containerPort": 8000, "name": "http"}], + "readinessProbe": { + "httpGet": {"path": "/health", "port": 8000}, + "initialDelaySeconds": 30, + "periodSeconds": 10, + "timeoutSeconds": 5, + }, + "resources": {"claims": [{"name": "devices"}]}, + "volumeMounts": [ + {"mountPath": "/dev/shm", "name": "dshm"}, + { + "mountPath": "/mnt/models", + "name": "model-cache", + }, + ], + } + ], + "nodeSelector": {"modelplane.ai/pool": "frontier"}, + "resourceClaims": [ + { + "name": "devices", + "resourceClaimTemplateName": "test-replica-main-leader-devices-34ec6", + } + ], + "tolerations": [ + { + "effect": "NoSchedule", + "key": "nvidia.com/gpu", + "operator": "Exists", + } + ], + "volumes": [ + {"emptyDir": {"medium": "Memory"}, "name": "dshm"}, + { + "name": "model-cache", + "persistentVolumeClaim": { + "claimName": "modelcache-ml-team-kimi-aa322" + }, + }, + ], + }, + }, + "size": 2, + "workerTemplate": { + "metadata": { + "labels": { + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Worker", + } + }, + "spec": { + "containers": [ + { + "args": ["--model=moonshotai/Kimi-K2-Instruct"], + "env": [ + { + "name": "MODELPLANE_LEADER_ADDRESS", + "value": "$(LWS_LEADER_ADDRESS)", + }, + { + "name": "MODELPLANE_RANK", + "value": "$(LWS_WORKER_INDEX)", + }, + { + "name": "HF_HUB_CACHE", + "value": "/mnt/models", + }, + ], + "image": "vllm/vllm-openai:latest", + "name": "engine", + "resources": {"claims": [{"name": "devices"}]}, + "volumeMounts": [ + {"mountPath": "/dev/shm", "name": "dshm"}, + { + "mountPath": "/mnt/models", + "name": "model-cache", + }, + ], + } + ], + "nodeSelector": {"modelplane.ai/pool": "frontier"}, + "resourceClaims": [ + { + "name": "devices", + "resourceClaimTemplateName": "test-replica-main-worker-devices-db183", + } + ], + "tolerations": [ + { + "effect": "NoSchedule", + "key": "nvidia.com/gpu", + "operator": "Exists", + } + ], + "volumes": [ + {"emptyDir": {"medium": "Memory"}, "name": "dshm"}, + { + "name": "model-cache", + "persistentVolumeClaim": { + "claimName": "modelcache-ml-team-kimi-aa322" + }, + }, + ], + }, + }, + }, + "replicas": 1, + }, + } + }, + "providerConfigRef": {"kind": "ClusterProviderConfig", "name": "cluster-a-pc"}, + "readiness": { + "celQuery": "has(object.status.conditions) && object.status.conditions.exists(c, " + 'c.type == "Available" && c.status == "True") && ' + "has(object.status.readyReplicas) && object.status.readyReplicas >= " + "object.spec.replicas", + "policy": "DeriveFromCelQuery", + }, + "watch": True, + }, + } + ), + ready=fnv1.READY_TRUE, + ), + "model-route": _route(ready=fnv1.READY_TRUE), + "resource-claim-main-leader": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-leader-devices-34ec6" + ), + "resource-claim-main-worker": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-worker-devices-db183" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_UNSPECIFIED), + "epp": _epp( + ready=fnv1.READY_UNSPECIFIED, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_UNSPECIFIED), + "epp-role": _epp_role(ready=fnv1.READY_UNSPECIFIED), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_UNSPECIFIED), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_UNSPECIFIED), + "epp-service": _epp_service(ready=fnv1.READY_UNSPECIFIED), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_WARNING, + message=( + "ModelCache kimi publishes no mount for cluster cluster-a. The replica's engines keep the " + "cache they mount now." + ), + ), + ], + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + "cache": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelCache", + match_name="kimi", + namespace="ml-team", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_TRUE, reason="Accepted"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForRouting"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForRouting", + message="1 of 1 engine copies ready", + ), + ], + ), + ), ] diff --git a/schemas/.lock.json b/schemas/.lock.json index ad2bd8e12..dc7eabefc 100644 --- a/schemas/.lock.json +++ b/schemas/.lock.json @@ -1,6 +1,6 @@ { "packages": { - "fs://apis": "1e73064210fec7fa5508d4af954e85d2adba93ccb40101324d926f950e8d2e2f", + "fs://apis": "df6c53d90327907620d69c4cb69dd656ff82294a2b062028e64bbe7296c022d6", "git://https://github.com/crossplane/crossplane/cluster/crds": "90d8b72ad8b829f0bcd7d7d5a98eaa0d579f244a", "xpkg://xpkg.upbound.io/upbound/provider-aws-ec2:v2.8.1": "sha256:ca2e9e3b2e3a8b6ca44a9700d5abf7abd733cfa388d1afe9bb2bf7c847cd39ef", "xpkg://xpkg.upbound.io/upbound/provider-aws-efs:v2.8.1": "sha256:ebb1bcd8dc9a7e60e97a324652fbd9d1b0609ddbb1107db518ce999cbc1113bd", diff --git a/schemas/python/models/ai/modelplane/modelcache/v1alpha1.py b/schemas/python/models/ai/modelplane/modelcache/v1alpha1.py index a63baed4c..2ca4ebe8d 100644 --- a/schemas/python/models/ai/modelplane/modelcache/v1alpha1.py +++ b/schemas/python/models/ai/modelplane/modelcache/v1alpha1.py @@ -5,7 +5,7 @@ from typing import Literal -from pydantic import AwareDatetime, BaseModel, conint, constr +from pydantic import AwareDatetime, BaseModel, Field, conint, constr from ....io.k8s.apimachinery.pkg.apis.meta import v1 @@ -88,8 +88,50 @@ class Spec(BaseModel): """ +class EnvItem(BaseModel): + name: str + value: str + + +class VolumeMount(BaseModel): + mountPath: str + name: str + readOnly: bool | None = None + + +class PersistentVolumeClaim(BaseModel): + claimName: str + + +class Volume(BaseModel): + name: str + persistentVolumeClaim: PersistentVolumeClaim | None = None + + +class Mount(BaseModel): + env: list[EnvItem] | None = Field(None, max_length=8) + """ + Container environment variables. A variable the engine sets itself takes precedence. + """ + volumeMounts: list[VolumeMount] | None = Field(None, max_length=8) + """ + Container volume mounts, each a subset of a Kubernetes VolumeMount. + """ + volumes: list[Volume] | None = Field(None, max_length=8) + """ + Pod volumes, each a subset of a Kubernetes Volume. + """ + + class Cluster(BaseModel): message: str | None = None + """ + Why the model isn't readable on this cluster yet, while phase isn't Ready. + """ + mount: Mount | None = None + """ + How an engine pod reads the model on this cluster. Set only while phase is Ready. A ModelReplica adds these to each of its engine pods. + """ name: str phase: Literal['Pending', 'Hydrating', 'Ready', 'Failed'] | None = None