diff --git a/apis/capacityreservations/composition.yaml b/apis/capacityreservations/composition.yaml new file mode 100644 index 000000000..6e41a33f8 --- /dev/null +++ b/apis/capacityreservations/composition.yaml @@ -0,0 +1,13 @@ +apiVersion: apiextensions.crossplane.io/v1 +kind: Composition +metadata: + name: capacityreservations.modelplane.ai +spec: + compositeTypeRef: + apiVersion: modelplane.ai/v1alpha1 + kind: CapacityReservation + mode: Pipeline + pipeline: + - functionRef: + name: modelplane-modelplanecompose-capacity-reservation + step: compose-capacity-reservation diff --git a/apis/capacityreservations/definition.yaml b/apis/capacityreservations/definition.yaml new file mode 100644 index 000000000..855c6e198 --- /dev/null +++ b/apis/capacityreservations/definition.yaml @@ -0,0 +1,126 @@ +apiVersion: apiextensions.crossplane.io/v2 +kind: CompositeResourceDefinition +metadata: + name: capacityreservations.modelplane.ai +spec: + group: modelplane.ai + names: + categories: [crossplane, modelplane, platform] + kind: CapacityReservation + plural: capacityreservations + scope: Cluster + versions: + - name: v1alpha1 + served: true + referenceable: true + additionalPrinterColumns: + - name: PROVIDER + type: string + jsonPath: .spec.provider + schema: + openAPIV3Schema: + type: object + description: >- + An existing cloud capacity reservation that InferenceCluster node + pools can reference by name. Modelplane doesn't create or modify the + reservation; it only directs node pools to consume it. + required: [spec] + properties: + spec: + type: object + required: [provider] + x-kubernetes-validations: + - rule: "self.provider != 'GKE' || has(self.gke)" + message: spec.gke is required when spec.provider is GKE. + - rule: "self.provider != 'EKS' || has(self.eks)" + message: spec.eks is required when spec.provider is EKS. + - rule: "self.provider == 'GKE' ? !has(self.eks) : !has(self.gke)" + message: Set only the block for spec.provider. + - rule: "self.provider == oldSelf.provider" + message: spec.provider is immutable. + properties: + provider: + type: string + description: >- + The cloud the reservation belongs to. A node pool can only + reference a reservation whose provider matches its cluster's + source. + enum: [GKE, EKS] + gke: + type: object + description: >- + A Compute Engine reservation, consumed as a specific + reservation. Any reservation type works, including future + reservations in calendar mode once their reservation period + starts. Required when provider is GKE. Immutable, because + node pools that reference this CapacityReservation can't + move to another reservation in place. + required: [zone, name] + x-kubernetes-validations: + - rule: "self == oldSelf" + message: spec.gke is immutable. Create a new CapacityReservation instead. + properties: + zone: + type: string + description: >- + Zone the reservation is in (e.g. us-central1-a). A node + pool backed by the reservation must be restricted to + exactly this zone. + minLength: 1 + maxLength: 63 + name: + type: string + description: Name of the reservation. + pattern: '^[a-z]([-a-z0-9]{0,61}[a-z0-9])?$' + maxLength: 63 + project: + type: string + description: >- + Project that owns the reservation, for a reservation + shared from another project. The reservation must be + shared with the cluster's project, and the cluster + project's GKE service agent must be allowed to list + reservations in this project. Omit for a reservation + in the cluster's own project. + pattern: '^[a-z][-a-z0-9]{4,28}[a-z0-9]$' + maxLength: 30 + eks: + type: object + description: >- + An EC2 Capacity Reservation. Required when provider is EKS. + Immutable, because node pools that reference this + CapacityReservation can't move to another reservation in + place. + required: [region, type, id] + x-kubernetes-validations: + - rule: "self == oldSelf" + message: spec.eks is immutable. Create a new CapacityReservation instead. + properties: + region: + type: string + description: AWS region the reservation is in (e.g. us-east-2). + minLength: 1 + maxLength: 32 + type: + type: string + description: >- + The kind of Capacity Reservation. CapacityBlock is a + Capacity Block for ML. OnDemand is an On-Demand + Capacity Reservation. + enum: [CapacityBlock, OnDemand] + id: + type: string + description: >- + ID of the Capacity Reservation (e.g. + cr-0123456789abcdef0). Purchasing a Capacity Block + yields this ID. + pattern: '^cr-[0-9a-f]+$' + minLength: 4 + maxLength: 64 + status: + type: object + properties: + conditions: + type: array + items: + type: object diff --git a/apis/eksclusters/definition.yaml b/apis/eksclusters/definition.yaml index 8bfa2dddd..e3718703b 100644 --- a/apis/eksclusters/definition.yaml +++ b/apis/eksclusters/definition.yaml @@ -188,29 +188,35 @@ spec: by the instance type. minLength: 1 maxLength: 63 - capacityBlock: + capacityReservation: type: object description: >- - Capacity Block reservation backing this node group. - Large GPU instances (e.g. p5en.48xlarge) are rarely - available on demand; AWS allocates them via Capacity - Blocks for ML. Set this to back the node group with a - Capacity Block you have purchased. When set, Modelplane - composes a launch template targeting the reservation - and creates the node group with CAPACITY_BLOCK capacity - type. The node group's zones must match the - reservation's Availability Zone, and nodeCount must not - exceed the reserved instance count. Omit for on-demand - node groups. + Capacity Reservation backing this node group. When set, + Modelplane composes a launch template that launches + instances only into the reservation. A Capacity Block + node group also uses the CAPACITY_BLOCK capacity type. + The node group's zones must match the reservation's + Availability Zone, and nodeCount must not exceed the + reserved instance count. Omit for on-demand node + groups. required: - - capacityReservationId + - type + - id properties: - capacityReservationId: + type: type: string description: >- - The ID of the Capacity Reservation backing the - Capacity Block (e.g. cr-0123456789abcdef0). - Purchasing a Capacity Block yields this ID. + The kind of Capacity Reservation. CapacityBlock is + a Capacity Block for ML. OnDemand is an On-Demand + Capacity Reservation. + enum: + - CapacityBlock + - OnDemand + id: + type: string + description: >- + The ID of the Capacity Reservation (e.g. + cr-0123456789abcdef0). pattern: "^cr-[0-9a-f]+$" minLength: 4 maxLength: 64 diff --git a/apis/gkeclusters/definition.yaml b/apis/gkeclusters/definition.yaml index 588edc7d0..2c7165716 100644 --- a/apis/gkeclusters/definition.yaml +++ b/apis/gkeclusters/definition.yaml @@ -197,6 +197,30 @@ spec: type: string minLength: 1 maxLength: 63 + reservation: + type: object + description: >- + Compute Engine reservation backing this node pool. The + pool consumes only this reservation, as a specific + reservation, so its size is limited to the + reservation's capacity. The pool's zones must be the + reservation's zone. Omit for on-demand pools. + required: + - name + properties: + name: + type: string + description: Name of the reservation. + minLength: 1 + maxLength: 63 + project: + type: string + description: >- + Project that owns the reservation, for a + reservation shared from another project. Omit for + a reservation in the cluster's project. + minLength: 1 + maxLength: 30 type: object status: description: GKEClusterStatus defines the observed state of GKECluster. diff --git a/apis/inferenceclusters/definition.yaml b/apis/inferenceclusters/definition.yaml index 0e0c8ca3c..e4a44d8de 100644 --- a/apis/inferenceclusters/definition.yaml +++ b/apis/inferenceclusters/definition.yaml @@ -32,6 +32,8 @@ spec: x-kubernetes-validations: - rule: "self.cluster.source != 'Nebius' || !has(self.nodePools) || self.nodePools.all(p, !has(p.fabric) || p.fabric.type != 'InfiniBand' || has(p.fabric.infiniband))" message: fabric.infiniband is required when fabric.type is InfiniBand and cluster.source is Nebius. + - rule: "self.cluster.source in ['GKE', 'EKS'] || !has(self.nodePools) || self.nodePools.all(p, !has(p.reservationRef))" + message: nodePools[].reservationRef is only supported when cluster.source is GKE or EKS. properties: cluster: type: object @@ -459,6 +461,9 @@ spec: items: type: object required: [name, className] + x-kubernetes-validations: + - rule: "has(self.reservationRef) == has(oldSelf.reservationRef) && (!has(self.reservationRef) || self.reservationRef.name == oldSelf.reservationRef.name)" + message: A node pool's reservationRef is immutable. Add a new pool instead. properties: name: type: string @@ -497,29 +502,26 @@ spec: region have every GPU type. items: type: string - capacityBlock: + reservationRef: type: object description: >- - Capacity Block reservation backing this node pool. - EKS only. Large GPU instances (e.g. p5en.48xlarge) - are rarely available on demand; AWS allocates them - via Capacity Blocks for ML. Set this to back the - pool with a Capacity Block you have purchased. The - pool's zones must match the reservation's - Availability Zone, and nodeCount must not exceed the - reserved instance count. Omit for on-demand pools. - required: - - capacityReservationId + CapacityReservation backing this node pool. GKE and + EKS only. The pool only uses the referenced + reservation, and never falls back to on-demand + capacity, so nodeCount must not exceed the reserved + instance count. The reservation's provider must match + the cluster's source. On GKE the pool's zones must be + exactly the reservation's zone; on EKS they must be + the reservation's Availability Zone. Immutable, + because neither cloud can move an existing node pool + onto or off a reservation. Omit for on-demand pools. + required: [name] properties: - capacityReservationId: + name: type: string - description: >- - The ID of the Capacity Reservation backing the - Capacity Block (e.g. cr-0123456789abcdef0). - Purchasing a Capacity Block yields this ID. - pattern: "^cr-[0-9a-f]+$" - minLength: 4 - maxLength: 64 + description: Name of the CapacityReservation. + minLength: 1 + maxLength: 253 fabric: type: object description: >- diff --git a/crossplane-project.yaml b/crossplane-project.yaml index 3b8fafbc9..9a979e9c5 100644 --- a/crossplane-project.yaml +++ b/crossplane-project.yaml @@ -19,6 +19,10 @@ spec: tarball: name: compose-aks-cluster pathPrefix: _output/functions/compose-aks-cluster + - source: Tarball + tarball: + name: compose-capacity-reservation + pathPrefix: _output/functions/compose-capacity-reservation - source: Tarball tarball: name: compose-civo-cluster diff --git a/docs/content/platform/inference-cluster.md b/docs/content/platform/inference-cluster.md index 55f2eef83..40151aa8f 100644 --- a/docs/content/platform/inference-cluster.md +++ b/docs/content/platform/inference-cluster.md @@ -147,6 +147,34 @@ the rank a gang computes until then. Composing a full Dynamo graph deployment, for its frontend and request routing, is planned. +## Capacity reservations + +Large GPU machines are often only available through a capacity reservation. On +GKE and EKS a node pool can consume one you've already bought. Register the +reservation as a `CapacityReservation`, then reference it from the pool's +`reservationRef`: + +{{< manifests path="concepts/capacity-reservation-gke.yaml" apply="false" >}} + +A pool that references a reservation only gets nodes from that reservation, and +never falls back to on-demand capacity. Keep its `nodeCount` within the +reserved instance count. On GKE, restrict the pool's `zones` to the +reservation's zone. On EKS, restrict them to the reservation's Availability +Zone. + +A `CapacityReservation` describes any GKE reservation, including a future +reservation in calendar mode once its reservation period starts. On EKS it +describes a Capacity Block for ML (`type: CapacityBlock`) or an On-Demand +Capacity Reservation (`type: OnDemand`). Modelplane doesn't create, change, or +delete the reservation itself. It blocks deleting a `CapacityReservation`, or an +`InferenceClass`, while a node pool references it. + +When a Capacity Block or a calendar-mode reservation ends, the cloud removes +the nodes that consume it. A pool's reservation can't change, and neither can +a `CapacityReservation`'s. To move to a new reservation, add a node pool that +references it, ideally before the old reservation ends, then remove the old +pool. + ## Examples {{< tabs >}} diff --git a/docs/content/recipes/qwen3-coder.md b/docs/content/recipes/qwen3-coder.md index 60c94cbae..ddc94aa7c 100644 --- a/docs/content/recipes/qwen3-coder.md +++ b/docs/content/recipes/qwen3-coder.md @@ -24,7 +24,7 @@ cache. Both deployments were run end to end; the `InferenceClass` and `ModelDeployment` are the exact manifests from those runs. Apply the platform side first, then the ML side. Edit the EC2 capacity reservation placeholder in -the `InferenceCluster` before applying it. +the `CapacityReservation` before applying it. ## Validated deployments diff --git a/docs/data/apigroups.yaml b/docs/data/apigroups.yaml index fc508a746..524467523 100644 --- a/docs/data/apigroups.yaml +++ b/docs/data/apigroups.yaml @@ -28,6 +28,7 @@ groups: # Link each kind back to its concept doc in the Documentation tab (by content # path). Kinds without an entry (machine-generated / infra) just omit the link. concepts: + CapacityReservation: /platform/inference-cluster InferenceGateway: /platform/inference-gateway InferenceClass: /platform/inference-class InferenceCluster: /platform/inference-cluster diff --git a/docs/manifests/concepts/capacity-reservation-gke.yaml b/docs/manifests/concepts/capacity-reservation-gke.yaml new file mode 100644 index 000000000..6c9a922ad --- /dev/null +++ b/docs/manifests/concepts/capacity-reservation-gke.yaml @@ -0,0 +1,30 @@ +# A GKE InferenceCluster whose L4 pool consumes a Compute Engine reservation. +# The pool is restricted to the reservation's zone. +apiVersion: modelplane.ai/v1alpha1 +kind: CapacityReservation +metadata: + name: l4-reservation +spec: + provider: GKE + gke: + zone: us-central1-a + name: l4-reservation + # project: shared-reservations # for a reservation shared from another project +--- +apiVersion: modelplane.ai/v1alpha1 +kind: InferenceCluster +metadata: + name: gke-reserved +spec: + cluster: + source: GKE + gke: + region: us-central1 + nodePools: + - name: l4-pool + className: gke-l4-1x-g2 + nodeCount: 2 + maxNodeCount: 2 + zones: [us-central1-a] + reservationRef: + name: l4-reservation diff --git a/docs/manifests/recipes/qwen3-coder/inference-cluster.yaml b/docs/manifests/recipes/qwen3-coder/inference-cluster.yaml index b9082376e..8766269e3 100644 --- a/docs/manifests/recipes/qwen3-coder/inference-cluster.yaml +++ b/docs/manifests/recipes/qwen3-coder/inference-cluster.yaml @@ -1,11 +1,23 @@ # An EKS InferenceCluster with a two-node H200 pool over EFA, validated serving # Qwen3-Coder-480B as a multi-node gang. The H200 nodes come from an EC2 -# Capacity Block reserved for ML. +# Capacity Block reserved for ML, registered as a CapacityReservation that the +# pool references. # # fabric.type: EFA turns on Elastic Fabric Adapter for the gang's cross-node # traffic; without it multi-node NCCL falls back to TCP, which is slow and # unstable. apiVersion: modelplane.ai/v1alpha1 +kind: CapacityReservation +metadata: + name: h200-capacity-block +spec: + provider: EKS + eks: + region: us-east-2 + type: CapacityBlock + id: cr-0123456789abcdef0 # replace with your reservation ID +--- +apiVersion: modelplane.ai/v1alpha1 kind: InferenceCluster metadata: name: eks-coder @@ -23,8 +35,8 @@ spec: minNodeCount: 2 maxNodeCount: 2 zones: - - us-east-2b + - us-east-2b # the Capacity Block's Availability Zone fabric: type: EFA - capacityBlock: - capacityReservationId: cr-0123456789abcdef0 # replace with your reservation ID + reservationRef: + name: h200-capacity-block diff --git a/docs/manifests/reference/capacityreservations.yaml b/docs/manifests/reference/capacityreservations.yaml new file mode 100644 index 000000000..3b2dc41ec --- /dev/null +++ b/docs/manifests/reference/capacityreservations.yaml @@ -0,0 +1,9 @@ +apiVersion: modelplane.ai/v1alpha1 +kind: CapacityReservation +metadata: + name: a3-mega-calendar +spec: + provider: GKE + gke: + zone: us-central1-a + name: a3-mega-calendar diff --git a/flake.nix b/flake.nix index d1f6a70db..76d7f1cb0 100644 --- a/flake.nix +++ b/flake.nix @@ -51,6 +51,7 @@ # The composition functions that make up Modelplane. functionNames = [ "compose-aks-cluster" + "compose-capacity-reservation" "compose-civo-cluster" "compose-eks-cluster" "compose-gke-cluster" diff --git a/functions/compose-capacity-reservation/function/__init__.py b/functions/compose-capacity-reservation/function/__init__.py new file mode 100644 index 000000000..b53d39d12 --- /dev/null +++ b/functions/compose-capacity-reservation/function/__init__.py @@ -0,0 +1,14 @@ +# 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. + diff --git a/functions/compose-capacity-reservation/function/fn.py b/functions/compose-capacity-reservation/function/fn.py new file mode 100644 index 000000000..120a5f579 --- /dev/null +++ b/functions/compose-capacity-reservation/function/fn.py @@ -0,0 +1,49 @@ +# 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. + +"""Compose a CapacityReservation. + +CapacityReservation is a data resource: it references a cloud capacity +reservation that InferenceCluster node pools consume. It has no composed +children. This function just marks the XR Ready. +""" + +import grpc +from crossplane.function import logging, resource, response +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.capacityreservation import v1alpha1 + + +class FunctionRunner(grpcv1.FunctionRunnerServiceServicer): + """A FunctionRunner handles gRPC RunFunctionRequests.""" + + def __init__(self) -> None: + """Create a new FunctionRunner.""" + self.log = logging.get_logger() + + async def RunFunction( + self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext | None + ) -> fnv1.RunFunctionResponse: # ty: ignore[invalid-method-override] # the generated grpc servicer base is untyped + """Run the function.""" + log = self.log.bind(tag=req.meta.tag) + log.info("Running function") + + rsp = response.to(req) + + resource.update_status(rsp.desired.composite, v1alpha1.Status()) + response.set_conditions(rsp, resource.Condition(typ="Accepted", status="True", reason="Available")) + rsp.desired.composite.ready = fnv1.READY_TRUE + + return rsp diff --git a/functions/compose-capacity-reservation/function/main.py b/functions/compose-capacity-reservation/function/main.py new file mode 100644 index 000000000..2e8441dac --- /dev/null +++ b/functions/compose-capacity-reservation/function/main.py @@ -0,0 +1,55 @@ +# 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. + +"""The composition function's main CLI.""" + +import click +from crossplane.function import logging, runtime + +from function import fn + + +@click.command() +@click.option("--debug", "-d", is_flag=True, help="Emit debug logs.") +@click.option( + "--address", + default="0.0.0.0:9443", + show_default=True, + help="Address at which to listen for gRPC connections", +) +@click.option("--tls-certs-dir", help="Serve using mTLS certificates.", envvar="TLS_SERVER_CERTS_DIR") +@click.option( + "--insecure", + is_flag=True, + help="Run without mTLS credentials. If you supply this flag --tls-certs-dir will be ignored.", +) +def cli(debug: bool, address: str, tls_certs_dir: str, insecure: bool) -> None: + """A Crossplane composition function.""" + try: + level = logging.Level.INFO + if debug: + level = logging.Level.DEBUG + logging.configure(level=level) + runtime.serve( + fn.FunctionRunner(), + address, + creds=runtime.load_credentials(tls_certs_dir), + insecure=insecure, + ) + except Exception as e: + click.echo(f"Cannot run function: {e}") + + +if __name__ == "__main__": + cli() diff --git a/functions/compose-capacity-reservation/pyproject.toml b/functions/compose-capacity-reservation/pyproject.toml new file mode 100644 index 000000000..176cebb14 --- /dev/null +++ b/functions/compose-capacity-reservation/pyproject.toml @@ -0,0 +1,26 @@ +[build-system] +requires = ["uv_build>=0.11.0,<0.12"] +build-backend = "uv_build" + +[project] +name = "compose-capacity-reservation" +version = "0.0.0" +description = "Mark a CapacityReservation as ready." +requires-python = ">=3.11,<3.14" +license = "Apache-2.0" +dependencies = [ + "crossplane-function-sdk-python>=0.14.0", + "click>=8.1.0", + "grpcio>=1.73.1", + "crossplane-models", +] + +[tool.uv.sources] +crossplane-models = { workspace = true } + +[project.scripts] +function = "function.main:cli" + +[tool.uv.build-backend] +module-name = "function" +module-root = "" diff --git a/functions/compose-capacity-reservation/tests/test_fn.py b/functions/compose-capacity-reservation/tests/test_fn.py new file mode 100644 index 000000000..068d4bc05 --- /dev/null +++ b/functions/compose-capacity-reservation/tests/test_fn.py @@ -0,0 +1,94 @@ +# 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 the compose-capacity-reservation function.""" + +import asyncio +import dataclasses +import json + +import pytest +from crossplane.function import resource +from crossplane.function.proto.v1 import run_function_pb2 as fnv1 +from function import fn +from google.protobuf import duration_pb2 as durationpb +from google.protobuf import json_format, message +from google.protobuf import struct_pb2 as structpb +from models.ai.modelplane.capacityreservation import v1alpha1 +from models.io.k8s.apimachinery.pkg.apis.meta import v1 as metav1 + + +@dataclasses.dataclass +class Case: + """A test case for compose-capacity-reservation.""" + + name: str + reason: str + req: fnv1.RunFunctionRequest + want: fnv1.RunFunctionResponse + + +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)) + + +COMPOSE_CASES = [ + Case( + name="GKEReservation", + reason="A CapacityReservation referencing a GKE reservation is ready and Accepted, with an empty status.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=fnv1.Resource( + resource=resource.dict_to_struct( + v1alpha1.CapacityReservation( + metadata=metav1.ObjectMeta(name="a3-mega-calendar"), + spec=v1alpha1.Spec( + provider="GKE", + gke=v1alpha1.Gke( + zone="us-central1-a", + name="a3-mega-calendar", + ), + ), + ).model_dump(exclude_none=True, mode="json", by_alias=True) + ), + ), + ), + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + composite=fnv1.Resource( + resource=resource.dict_to_struct({"status": {}}), + ready=fnv1.READY_TRUE, + ), + ), + context=structpb.Struct(), + conditions=[ + fnv1.Condition( + type="Accepted", + status=fnv1.STATUS_CONDITION_TRUE, + reason="Available", + ), + ], + ), + ), +] + + +@pytest.mark.parametrize("case", COMPOSE_CASES, ids=lambda case: case.name) +def test_compose(case: Case) -> None: + """RunFunction marks the CapacityReservation ready.""" + got = asyncio.run(fn.FunctionRunner().RunFunction(case.req, None)) + assert _to_dict(got) == _to_dict(case.want), case.reason diff --git a/functions/compose-eks-cluster/function/fn.py b/functions/compose-eks-cluster/function/fn.py index 14759acee..cce585b31 100644 --- a/functions/compose-eks-cluster/function/fn.py +++ b/functions/compose-eks-cluster/function/fn.py @@ -165,13 +165,14 @@ def _namespace(meta: metav1.ObjectMeta | None) -> str: # since EKS won't accept the node group's diskSize when a template is set. _ROOT_DEVICE_NAME = "/dev/xvda" -# Capacity Block backing. Large GPU instances (e.g. p5en.48xlarge) are -# rarely available on demand; AWS allocates them via Capacity Blocks for -# ML. A node group backed by a Capacity Block uses the CAPACITY_BLOCK -# capacity type and a launch template that targets the reservation: it -# sets the instance market type to "capacity-block" and points at the -# reservation ID. EKS requires the instance type to come from the launch -# template (not the node group's instanceTypes) for capacity-block groups. +# Capacity Reservation backing. A node group backed by a Capacity Reservation +# uses a launch template that launches instances only into the reservation. A +# Capacity Block for ML additionally needs the "capacity-block" instance market +# type in the template and the CAPACITY_BLOCK capacity type on the node group; +# an On-Demand Capacity Reservation uses the default on-demand market. EKS +# requires the instance type to come from the launch template (not the node +# group's instanceTypes) whenever a template is set. +_RESERVATION_TYPE_CAPACITY_BLOCK = "CapacityBlock" _CAPACITY_TYPE_CAPACITY_BLOCK = "CAPACITY_BLOCK" _MARKET_TYPE_CAPACITY_BLOCK = "capacity-block" _CR_PREFERENCE_ONLY = "capacity-reservations-only" @@ -850,15 +851,15 @@ def compose_node_groups(self) -> None: """Compose the system and user-declared GPU node groups.""" self._compose_system_node_group() for pool in self.xr.spec.nodePools: - capacity_block = pool.capacityBlock + reservation = pool.capacityReservation efa = pool.fabric == _FABRIC_EFA - # A launch template is needed to carry the capacity-block market - # options, the EFA interfaces, or both. EKS takes the instance type - # from the launch template whenever one is set, so the node group - # must not also set instanceTypes in that case. - uses_launch_template = bool(capacity_block) or efa + # A launch template is needed to carry the capacity reservation, the + # EFA interfaces, or both. EKS takes the instance type from the + # launch template whenever one is set, so the node group must not + # also set instanceTypes in that case. + uses_launch_template = bool(reservation) or efa if uses_launch_template: - self._compose_launch_template(pool, capacity_block, efa=efa) + self._compose_launch_template(pool, reservation, efa=efa) if efa: self._compose_efa_security_group() @@ -884,15 +885,15 @@ def compose_node_groups(self) -> None: if uses_launch_template: # EKS takes the instance type from the launch template, so the # node group must not also set instanceTypes. The template - # carries the instance type, the disk size, any capacity-block - # market options, and any EFA interfaces. EKS rejects a node + # carries the instance type, the disk size, any capacity + # reservation, and any EFA interfaces. EKS rejects a node # group that sets diskSize alongside a launch template, so the # disk size lives only in the template in this case. fp.launchTemplate = ngv1beta1.LaunchTemplate( name=self._launch_template_name(pool), version="$Latest", ) - if capacity_block: + if reservation and reservation.type == _RESERVATION_TYPE_CAPACITY_BLOCK: fp.capacityType = _CAPACITY_TYPE_CAPACITY_BLOCK else: fp.instanceTypes = [pool.instanceType] @@ -950,7 +951,7 @@ def compose_node_groups(self) -> None: ) def _launch_template_name(self, pool: v1alpha1.NodePool) -> str: - """Derive the EC2 launch template name for a Capacity Block pool. + """Derive the EC2 launch template name for a pool that needs one. The EKS NodeGroup references the launch template by name, so it must be stable and match the launchTemplate.forProvider.name set @@ -959,15 +960,15 @@ def _launch_template_name(self, pool: v1alpha1.NodePool) -> str: return resource.child_name(_name(self.xr.metadata), f"lt-{pool.name}") def _compose_launch_template( - self, pool: v1alpha1.NodePool, capacity_block: v1alpha1.CapacityBlock | None, *, efa: bool + self, pool: v1alpha1.NodePool, reservation: v1alpha1.CapacityReservation | None, *, efa: bool ) -> None: """Compose an EC2 launch template for a node group. EKS launches the node group's instances from this template. It carries - the instance type and, depending on the pool, the capacity-block market - options (so instances come from the reservation) and/or EFA network + the instance type and, depending on the pool, the capacity reservation + (so instances come only from the reservation) and/or EFA network interfaces (so nodes get GPUDirect RDMA). The node group references it - by name; for a capacity-block pool it also sets capacityType. + by name; for a Capacity Block pool it also sets capacityType. The template must not set subnets - EKS rejects a launch template with subnet configuration for managed node groups and takes the subnets from @@ -987,16 +988,17 @@ def _compose_launch_template( ], ) - if capacity_block: - fp.instanceMarketOptions = ltv1beta1.InstanceMarketOptions( - marketType=_MARKET_TYPE_CAPACITY_BLOCK, - ) + if reservation: fp.capacityReservationSpecification = ltv1beta1.CapacityReservationSpecification( capacityReservationPreference=_CR_PREFERENCE_ONLY, capacityReservationTarget=ltv1beta1.CapacityReservationTarget( - capacityReservationId=capacity_block.capacityReservationId, + capacityReservationId=reservation.id, ), ) + if reservation and reservation.type == _RESERVATION_TYPE_CAPACITY_BLOCK: + fp.instanceMarketOptions = ltv1beta1.InstanceMarketOptions( + marketType=_MARKET_TYPE_CAPACITY_BLOCK, + ) if efa: fp.networkInterfaces = self._efa_network_interfaces(pool) @@ -1756,7 +1758,9 @@ def mark_readiness(self) -> None: managed_resources.append(f"efs-mount-target-{i}") managed_resources += [f"nodegroup-{p.name}" for p in self.xr.spec.nodePools] managed_resources += [ - f"launch-template-{p.name}" for p in self.xr.spec.nodePools if p.capacityBlock or p.fabric == _FABRIC_EFA + f"launch-template-{p.name}" + for p in self.xr.spec.nodePools + if p.capacityReservation or p.fabric == _FABRIC_EFA ] if any(p.fabric == _FABRIC_EFA for p in self.xr.spec.nodePools): managed_resources += [ diff --git a/functions/compose-eks-cluster/tests/test_fn.py b/functions/compose-eks-cluster/tests/test_fn.py index bd16d8e9c..83a6d50ec 100644 --- a/functions/compose-eks-cluster/tests/test_fn.py +++ b/functions/compose-eks-cluster/tests/test_fn.py @@ -1822,7 +1822,9 @@ def _to_dict(msg: message.Message) -> dict: maxNodeCount=2, diskSizeGb=1024, gpu=v1alpha1.Gpu(acceleratorType="nvidia-h200"), - capacityBlock=v1alpha1.CapacityBlock(capacityReservationId="cr-0123456789abcdef0"), + capacityReservation=v1alpha1.CapacityReservation( + type="CapacityBlock", id="cr-0123456789abcdef0" + ), zones=[v1alpha1.Zone("us-west-2a")], ), ], @@ -2067,6 +2069,270 @@ def _to_dict(msg: message.Message) -> dict: context=structpb.Struct(), ), ), + # An On-Demand Capacity Reservation needs no market type, and its node group + # keeps the default on-demand capacity type. + Case( + name="OnDemandCapacityReservation", + reason="An EKSCluster with an On-Demand Capacity Reservation pool composes a launch template targeting its reservation, with no market type, and an on-demand node group that uses the template.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_xr( + credentials=None, + node_pools=[ + v1alpha1.NodePool( + name="gpu-h200", + role="GPU", + instanceType="p5en.48xlarge", + nodeCount=2, + minNodeCount=0, + maxNodeCount=2, + diskSizeGb=1024, + gpu=v1alpha1.Gpu(acceleratorType="nvidia-h200"), + capacityReservation=v1alpha1.CapacityReservation( + type="OnDemand", id="cr-0123456789abcdef0" + ), + zones=[v1alpha1.Zone("us-west-2a")], + ), + ], + ), + ), + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + composite=_desired_xr(), + resources={ + "vpc": _vpc(cred_kind="ClusterProviderConfig", cred_name="default"), + "subnet-0": _subnet( + name="test-cluster-subnet-us-west-2a-952dc", + az="us-west-2a", + cidr="10.0.0.0/20", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "subnet-1": _subnet( + name="test-cluster-subnet-us-west-2b-2b80f", + az="us-west-2b", + cidr="10.0.16.0/20", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "subnet-2": _subnet( + name="test-cluster-subnet-us-west-2c-03273", + az="us-west-2c", + cidr="10.0.32.0/20", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "private-subnet-0": _private_subnet( + name="test-cluster-private-subnet-us-west-2a-6a89f", + az="us-west-2a", + cidr="10.0.48.0/20", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "private-subnet-1": _private_subnet( + name="test-cluster-private-subnet-us-west-2b-b7832", + az="us-west-2b", + cidr="10.0.64.0/20", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "private-subnet-2": _private_subnet( + name="test-cluster-private-subnet-us-west-2c-ef57d", + az="us-west-2c", + cidr="10.0.80.0/20", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "internet-gateway": _internet_gateway(cred_kind="ClusterProviderConfig", cred_name="default"), + "nat-eip": _nat_eip(cred_kind="ClusterProviderConfig", cred_name="default"), + "nat-gateway": _nat_gateway(cred_kind="ClusterProviderConfig", cred_name="default"), + "route-table": _route_table(cred_kind="ClusterProviderConfig", cred_name="default"), + "route-default": _route_default(cred_kind="ClusterProviderConfig", cred_name="default"), + "private-route-table": _private_route_table(cred_kind="ClusterProviderConfig", cred_name="default"), + "private-route-default": _private_route_default( + cred_kind="ClusterProviderConfig", cred_name="default" + ), + "route-table-association-0": _route_table_association( + az="us-west-2a", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "route-table-association-1": _route_table_association( + az="us-west-2b", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "route-table-association-2": _route_table_association( + az="us-west-2c", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "private-route-table-association-0": _private_route_table_association( + az="us-west-2a", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "private-route-table-association-1": _private_route_table_association( + az="us-west-2b", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "private-route-table-association-2": _private_route_table_association( + az="us-west-2c", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "iam-role-cluster": _cluster_role(cred_kind="ClusterProviderConfig", cred_name="default"), + "iam-attach-cluster-policy": _role_policy_attachment( + role="cluster", + arn="arn:aws:iam::aws:policy/AmazonEKSClusterPolicy", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "iam-role-node": _node_role(cred_kind="ClusterProviderConfig", cred_name="default"), + "iam-attach-node-worker": _role_policy_attachment( + role="node", + arn="arn:aws:iam::aws:policy/AmazonEKSWorkerNodePolicy", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "iam-attach-node-cni": _role_policy_attachment( + role="node", + arn="arn:aws:iam::aws:policy/AmazonEKS_CNI_Policy", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "iam-attach-node-ecr": _role_policy_attachment( + role="node", + arn="arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "cluster": _eks_cluster( + cred_kind="ClusterProviderConfig", cred_name="default", ready=fnv1.READY_UNSPECIFIED + ), + "cluster-auth": _cluster_auth( + cred_kind="ClusterProviderConfig", cred_name="default", ready=fnv1.READY_UNSPECIFIED + ), + "nodegroup-system": _nodegroup_system(cred_kind="ClusterProviderConfig", cred_name="default"), + "launch-template-gpu-h200": fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "ec2.aws.m.upbound.io/v1beta1", + "kind": "LaunchTemplate", + "spec": { + "providerConfigRef": {"kind": "ClusterProviderConfig", "name": "default"}, + "forProvider": { + "region": "us-west-2", + "name": "test-cluster-lt-gpu-h200-83c00", + "instanceType": "p5en.48xlarge", + "blockDeviceMappings": [ + {"deviceName": "/dev/xvda", "ebs": {"volumeSize": 1024}}, + ], + "capacityReservationSpecification": { + "capacityReservationPreference": "capacity-reservations-only", + "capacityReservationTarget": { + "capacityReservationId": "cr-0123456789abcdef0", + }, + }, + }, + }, + } + ), + ), + "nodegroup-gpu-h200": fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "eks.aws.m.upbound.io/v1beta1", + "kind": "NodeGroup", + "spec": { + "providerConfigRef": {"kind": "ClusterProviderConfig", "name": "default"}, + "managementPolicies": ["Observe", "Create", "Update", "Delete"], + "initProvider": {"scalingConfig": {"desiredSize": 2}}, + "forProvider": { + "region": "us-west-2", + "amiType": "AL2023_x86_64_NVIDIA", + "clusterNameSelector": {"matchControllerRef": True}, + "nodeRoleArnSelector": { + "matchControllerRef": True, + "matchLabels": {"modelplane.ai/iam-role": "node"}, + }, + "launchTemplate": { + "name": "test-cluster-lt-gpu-h200-83c00", + "version": "$Latest", + }, + "subnetIdRefs": [{"name": "test-cluster-private-subnet-us-west-2a-6a89f"}], + "scalingConfig": {"minSize": 0, "maxSize": 2}, + "labels": { + "modelplane.ai/gpu": "nvidia-h200", + "modelplane.ai/pool": "gpu-h200", + }, + "taint": [ + { + "key": "nvidia.com/gpu", + "value": "true", + "effect": "NO_SCHEDULE", + }, + ], + }, + }, + } + ), + ), + "addon-vpc-cni": _addon(name="vpc-cni", cred_kind="ClusterProviderConfig", cred_name="default"), + "addon-kube-proxy": _addon( + name="kube-proxy", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "addon-coredns": _addon(name="coredns", cred_kind="ClusterProviderConfig", cred_name="default"), + "efs-filesystem": _efs_filesystem( + cred_kind="ClusterProviderConfig", cred_name="default", ready=fnv1.READY_UNSPECIFIED + ), + "efs-security-group": _efs_security_group(cred_kind="ClusterProviderConfig", cred_name="default"), + "efs-security-group-ingress": _efs_security_group_ingress( + cred_kind="ClusterProviderConfig", cred_name="default" + ), + "efs-mount-target-0": _efs_mount_target( + subnet_name="test-cluster-private-subnet-us-west-2a-6a89f", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "efs-mount-target-1": _efs_mount_target( + subnet_name="test-cluster-private-subnet-us-west-2b-b7832", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "efs-mount-target-2": _efs_mount_target( + subnet_name="test-cluster-private-subnet-us-west-2c-ef57d", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "iam-role-efs-csi": _pod_identity_role( + role="efs-csi", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "iam-attach-efs-csi": _role_policy_attachment( + role="efs-csi", + arn="arn:aws:iam::aws:policy/service-role/AmazonEFSCSIDriverPolicy", + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "addon-eks-pod-identity-agent": _addon( + name="eks-pod-identity-agent", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "pod-identity-efs-csi": _pod_identity_association( + cred_kind="ClusterProviderConfig", cred_name="default" + ), + "addon-aws-efs-csi-driver": _addon( + name="aws-efs-csi-driver", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "iam-policy-cluster-autoscaler": _autoscaler_policy( + cred_kind="ClusterProviderConfig", cred_name="default" + ), + "iam-role-cluster-autoscaler": _pod_identity_role( + role="cluster-autoscaler", cred_kind="ClusterProviderConfig", cred_name="default" + ), + "iam-attach-cluster-autoscaler": _autoscaler_attachment( + cred_kind="ClusterProviderConfig", cred_name="default" + ), + "pod-identity-cluster-autoscaler": _autoscaler_pod_identity( + cred_kind="ClusterProviderConfig", cred_name="default" + ), + "provider-config-kubernetes": _provider_config(api_version="kubernetes.m.crossplane.io/v1alpha1"), + "provider-config-helm": _provider_config(api_version="helm.m.crossplane.io/v1beta1"), + }, + ), + context=structpb.Struct(), + ), + ), # The node group's launch template carries one EFA interface per network card # (card 0 keeps device index 0 for the node's IP traffic, the rest device # index 1 for RDMA), the cluster gets an EFA security group with diff --git a/functions/compose-gke-cluster/function/fn.py b/functions/compose-gke-cluster/function/fn.py index a45542bd0..c008583cb 100644 --- a/functions/compose-gke-cluster/function/fn.py +++ b/functions/compose-gke-cluster/function/fn.py @@ -85,6 +85,14 @@ # minNodeCount 0. _LABEL_GPU_DRA_DRIVER = "cloud.google.com/gke-nvidia-gpu-dra-driver" +# A node pool backed by a reservation consumes it as a specific reservation, so +# GKE creates the pool's nodes only from that reservation and never from +# on-demand capacity. GKE identifies the reservation by this key, with the +# reservation's name, or projects//reservations/ for a +# reservation shared from another project. +_RESERVATION_AFFINITY_SPECIFIC = "SPECIFIC_RESERVATION" +_RESERVATION_AFFINITY_KEY = "compute.googleapis.com/reservation-name" + # Secret types written to XR status. compose-inference-cluster reads # these to wire the kubeconfig and SA key into ProviderConfigs. The SA key's # type is the provider identity it authenticates as (see _IDENTITY_TYPE_GCP). @@ -414,6 +422,16 @@ def compose_node_pools(self) -> None: _LABEL_POOL: pool.name, } + if pool.reservation: + name = pool.reservation.name + if pool.reservation.project: + name = f"projects/{pool.reservation.project}/reservations/{name}" + node_config.reservationAffinity = nodepoolv1beta1.ReservationAffinity( + consumeReservationType=_RESERVATION_AFFINITY_SPECIFIC, + key=_RESERVATION_AFFINITY_KEY, + values=[name], + ) + np = nodepoolv1beta1.NodePool( spec=nodepoolv1beta1.Spec( providerConfigRef=nodepoolv1beta1.ProviderConfigRef( diff --git a/functions/compose-gke-cluster/tests/test_fn.py b/functions/compose-gke-cluster/tests/test_fn.py index fe12aabf4..0ec5e71fc 100644 --- a/functions/compose-gke-cluster/tests/test_fn.py +++ b/functions/compose-gke-cluster/tests/test_fn.py @@ -236,8 +236,32 @@ def _nodepool_system(*, cred_kind: str, cred_name: str) -> fnv1.Resource: ) -def _nodepool_gpu(*, cred_kind: str, cred_name: str) -> fnv1.Resource: - """The composed NodePool for the XR's gpu-pool.""" +def _nodepool_gpu(*, cred_kind: str, cred_name: str, reservation_affinity: dict | None) -> fnv1.Resource: + """The composed NodePool for the XR's gpu-pool, consuming a reservation unless reservation_affinity is None.""" + node_config: dict = { + "machineType": "a2-highgpu-8g", + "diskSizeGb": 100, + "imageType": "COS_CONTAINERD", + "oauthScopes": [ + "https://www.googleapis.com/auth/cloud-platform", + ], + "guestAccelerator": [ + { + "type": "nvidia-tesla-a100", + "count": 8, + "gpuDriverInstallationConfig": { + "gpuDriverVersion": "DEFAULT", + }, + }, + ], + "labels": { + "modelplane.ai/gpu": "nvidia-tesla-a100", + "modelplane.ai/pool": "gpu-pool", + "cloud.google.com/gke-nvidia-gpu-dra-driver": "true", + }, + } + if reservation_affinity is not None: + node_config["reservationAffinity"] = reservation_affinity return fnv1.Resource( resource=resource.dict_to_struct( { @@ -250,28 +274,7 @@ def _nodepool_gpu(*, cred_kind: str, cred_name: str) -> fnv1.Resource: "clusterSelector": {"matchControllerRef": True}, "initialNodeCount": 1, "autoscaling": {"minNodeCount": 0, "maxNodeCount": 8}, - "nodeConfig": { - "machineType": "a2-highgpu-8g", - "diskSizeGb": 100, - "imageType": "COS_CONTAINERD", - "oauthScopes": [ - "https://www.googleapis.com/auth/cloud-platform", - ], - "guestAccelerator": [ - { - "type": "nvidia-tesla-a100", - "count": 8, - "gpuDriverInstallationConfig": { - "gpuDriverVersion": "DEFAULT", - }, - }, - ], - "labels": { - "modelplane.ai/gpu": "nvidia-tesla-a100", - "modelplane.ai/pool": "gpu-pool", - "cloud.google.com/gke-nvidia-gpu-dra-driver": "true", - }, - }, + "nodeConfig": node_config, }, }, } @@ -438,7 +441,153 @@ def _to_dict(msg: message.Message) -> dict: "subnet": _subnet(cred_kind="ClusterProviderConfig", cred_name="default"), "cluster": _cluster(cred_kind="ClusterProviderConfig", cred_name="default"), "nodepool-system": _nodepool_system(cred_kind="ClusterProviderConfig", cred_name="default"), - "nodepool-gpu-pool": _nodepool_gpu(cred_kind="ClusterProviderConfig", cred_name="default"), + "nodepool-gpu-pool": _nodepool_gpu( + cred_kind="ClusterProviderConfig", cred_name="default", reservation_affinity=None + ), + "service-account": _service_account( + cred_kind="ClusterProviderConfig", + cred_name="default", + ready=fnv1.READY_UNSPECIFIED, + ), + "service-account-key": _service_account_key(cred_kind="ClusterProviderConfig", cred_name="default"), + "provider-config-kubernetes": _provider_config(api_version="kubernetes.m.crossplane.io/v1alpha1"), + "provider-config-helm": _provider_config(api_version="helm.m.crossplane.io/v1beta1"), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "gcp-provider-config": fnv1.ResourceSelector( + api_version="gcp.m.upbound.io/v1beta1", + kind="ClusterProviderConfig", + match_name="default", + ), + }, + ), + ), + ), + Case( + name="Reservation", + reason="A GPU pool backed by a reservation in the cluster's project consumes it as a specific reservation, named by its name alone.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_xr( + credentials=None, + node_pools=[ + v1alpha1.NodePool( + name="gpu-pool", + role="GPU", + machineType="a2-highgpu-8g", + gpu=v1alpha1.Gpu(acceleratorType="nvidia-tesla-a100", acceleratorCount=8), + reservation=v1alpha1.Reservation(name="a100-calendar"), + ), + ], + ), + ), + required_resources={ + "gcp-provider-config": fnv1.Resources( + items=[_gcp_provider_config(kind="ClusterProviderConfig", name="default", namespace=None)], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + composite=_desired_xr(), + resources={ + "network": _network( + cred_kind="ClusterProviderConfig", + cred_name="default", + ready=fnv1.READY_UNSPECIFIED, + ), + "projectservice-filestore": _projectservice_filestore( + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "subnet": _subnet(cred_kind="ClusterProviderConfig", cred_name="default"), + "cluster": _cluster(cred_kind="ClusterProviderConfig", cred_name="default"), + "nodepool-system": _nodepool_system(cred_kind="ClusterProviderConfig", cred_name="default"), + "nodepool-gpu-pool": _nodepool_gpu( + cred_kind="ClusterProviderConfig", + cred_name="default", + reservation_affinity={ + "consumeReservationType": "SPECIFIC_RESERVATION", + "key": "compute.googleapis.com/reservation-name", + "values": ["a100-calendar"], + }, + ), + "service-account": _service_account( + cred_kind="ClusterProviderConfig", + cred_name="default", + ready=fnv1.READY_UNSPECIFIED, + ), + "service-account-key": _service_account_key(cred_kind="ClusterProviderConfig", cred_name="default"), + "provider-config-kubernetes": _provider_config(api_version="kubernetes.m.crossplane.io/v1alpha1"), + "provider-config-helm": _provider_config(api_version="helm.m.crossplane.io/v1beta1"), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "gcp-provider-config": fnv1.ResourceSelector( + api_version="gcp.m.upbound.io/v1beta1", + kind="ClusterProviderConfig", + match_name="default", + ), + }, + ), + ), + ), + Case( + name="SharedReservation", + reason="A GPU pool backed by a reservation shared from another project consumes it as a specific reservation, named by its project path.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_xr( + credentials=None, + node_pools=[ + v1alpha1.NodePool( + name="gpu-pool", + role="GPU", + machineType="a2-highgpu-8g", + gpu=v1alpha1.Gpu(acceleratorType="nvidia-tesla-a100", acceleratorCount=8), + reservation=v1alpha1.Reservation(name="a100-calendar", project="shared-reservations"), + ), + ], + ), + ), + required_resources={ + "gcp-provider-config": fnv1.Resources( + items=[_gcp_provider_config(kind="ClusterProviderConfig", name="default", namespace=None)], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + composite=_desired_xr(), + resources={ + "network": _network( + cred_kind="ClusterProviderConfig", + cred_name="default", + ready=fnv1.READY_UNSPECIFIED, + ), + "projectservice-filestore": _projectservice_filestore( + cred_kind="ClusterProviderConfig", + cred_name="default", + ), + "subnet": _subnet(cred_kind="ClusterProviderConfig", cred_name="default"), + "cluster": _cluster(cred_kind="ClusterProviderConfig", cred_name="default"), + "nodepool-system": _nodepool_system(cred_kind="ClusterProviderConfig", cred_name="default"), + "nodepool-gpu-pool": _nodepool_gpu( + cred_kind="ClusterProviderConfig", + cred_name="default", + reservation_affinity={ + "consumeReservationType": "SPECIFIC_RESERVATION", + "key": "compute.googleapis.com/reservation-name", + "values": ["projects/shared-reservations/reservations/a100-calendar"], + }, + ), "service-account": _service_account( cred_kind="ClusterProviderConfig", cred_name="default", @@ -695,7 +844,9 @@ def _to_dict(msg: message.Message) -> dict: "subnet": _subnet(cred_kind="ClusterProviderConfig", cred_name="default"), "cluster": _cluster(cred_kind="ClusterProviderConfig", cred_name="default"), "nodepool-system": _nodepool_system(cred_kind="ClusterProviderConfig", cred_name="default"), - "nodepool-gpu-pool": _nodepool_gpu(cred_kind="ClusterProviderConfig", cred_name="default"), + "nodepool-gpu-pool": _nodepool_gpu( + cred_kind="ClusterProviderConfig", cred_name="default", reservation_affinity=None + ), "service-account": _service_account( cred_kind="ClusterProviderConfig", cred_name="default", @@ -798,7 +949,9 @@ def _to_dict(msg: message.Message) -> dict: "subnet": _subnet(cred_kind="ProviderConfig", cred_name="my-gcp-account"), "cluster": _cluster(cred_kind="ProviderConfig", cred_name="my-gcp-account"), "nodepool-system": _nodepool_system(cred_kind="ProviderConfig", cred_name="my-gcp-account"), - "nodepool-gpu-pool": _nodepool_gpu(cred_kind="ProviderConfig", cred_name="my-gcp-account"), + "nodepool-gpu-pool": _nodepool_gpu( + cred_kind="ProviderConfig", cred_name="my-gcp-account", reservation_affinity=None + ), "service-account": _service_account( cred_kind="ProviderConfig", cred_name="my-gcp-account", diff --git a/functions/compose-inference-cluster/function/fn.py b/functions/compose-inference-cluster/function/fn.py index 822c54424..07d4a5b47 100644 --- a/functions/compose-inference-cluster/function/fn.py +++ b/functions/compose-inference-cluster/function/fn.py @@ -39,6 +39,7 @@ 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.agent.poolobservation import v1alpha1 as pov1alpha1 +from models.ai.modelplane.capacityreservation import v1alpha1 as crv1alpha1 from models.ai.modelplane.inferenceclass import v1alpha1 as iclv1alpha1 from models.ai.modelplane.inferencecluster import v1alpha1 from models.ai.modelplane.inferencegateway import v1alpha1 as igv1alpha1 @@ -81,6 +82,7 @@ CONDITION_REASON_PROVISIONING = "Provisioning" CONDITION_REASON_WAITING_FOR_CLUSTER = "WaitingForCluster" CONDITION_REASON_WAITING_FOR_CLASSES = "WaitingForClasses" +CONDITION_REASON_WAITING_FOR_RESERVATIONS = "WaitingForReservations" CONDITION_REASON_BACKEND_HEALTHY = "BackendHealthy" CONDITION_REASON_INSTALLING = "Installing" CONDITION_REASON_INVALID_NODE_POOL = "InvalidNodePool" @@ -325,6 +327,9 @@ def __init__(self, req: fnv1.RunFunctionRequest, rsp: fnv1.RunFunctionResponse) # Resolved InferenceClasses, keyed by class name. Populated by # resolve_classes(). self.classes: dict[str, iclv1alpha1.InferenceClass] = {} + # Resolved CapacityReservations, keyed by reservation name. Populated by + # resolve_reservations(). + self.reservations: dict[str, crv1alpha1.CapacityReservation] = {} # Client CA per InferenceGateway, keyed by gateway name. self.gateway_cas: dict[str, str] = {} @@ -339,10 +344,10 @@ def compose(self) -> None: self.require_inputs() # The deletion guard, namespaces, and gateway CAs all run before any - # early return: none depends on the cluster's source or whether its - # classes resolve. Gating the guard behind those would drop its - # ClusterUsage on a reconcile where classes are transiently unresolved - # and let the cluster be deleted while something still uses it. + # early return: none depends on the cluster's source. Gating the guard + # behind it would drop its ClusterUsage on a reconcile that returns + # early without failing, and let the cluster be deleted while something + # still uses it. self.compose_deletion_guard() self.compose_namespaces() self.resolve_gateway_cas() @@ -352,9 +357,11 @@ def compose(self) -> None: response.warning(self.rsp, "spec.cluster is required") return - if not self.resolve_classes(): + if not self.resolve_classes() or not self.resolve_reservations(): return + self.compose_input_usages() + source = cluster.source if not self.check_devices_supported(source): return @@ -424,7 +431,8 @@ def require_inputs(self) -> None: _LABEL_CLUSTER), and the ModelCaches staging onto it, drive the deletion guard and the mirrored namespaces; the InferenceGateways carry the client CAs the cluster gateway must accept, and those running here also - drive the guard; and the InferenceClasses back the node pools. + drive the guard; and the InferenceClasses and CapacityReservations back + the node pools. """ name = _name(self.xr.metadata) for kind, key in (("ModelReplica", "model-replicas"), ("ModelRoute", "model-routes")): @@ -460,6 +468,16 @@ def require_inputs(self) -> None: kind="InferenceClass", match_name=class_name, ) + for reservation_name in sorted( + {p.reservationRef.name for p in (self.xr.spec.nodePools or []) if p.reservationRef} + ): + response.require_resources( + self.rsp, + name=f"reservation-{reservation_name}", + api_version="modelplane.ai/v1alpha1", + kind="CapacityReservation", + match_name=reservation_name, + ) def compose_deletion_guard(self) -> None: """Block deletion of the InferenceCluster while anything composes onto it. @@ -520,6 +538,42 @@ def compose_deletion_guard(self) -> None: ) self.rsp.desired.resources[_DELETION_GUARD_RESOURCE_KEY].ready = fnv1.READY_TRUE + def compose_input_usages(self) -> None: + """Block deletion of the InferenceClasses and CapacityReservations the + node pools reference. + + The function fails every reconcile while one of them is missing, so + deleting one in use would stop this cluster reconciling until it's + recreated. Each ClusterUsage blocks that delete while this cluster + references the class or reservation. Once it doesn't, the function stops + composing the ClusterUsage, and replayDeletion re-issues a delete that + was already attempted. + + The ClusterUsages give a reason rather than naming this cluster as + their `by`. Crossplane keeps a Usage with a `by` until the `by` is gone, + and a foreground delete of this cluster waits for its composed + ClusterUsages, so neither would ever finish. + """ + name = _name(self.xr.metadata) + inputs = [("InferenceClass", c) for c in self.classes] + [("CapacityReservation", r) for r in self.reservations] + for kind, ref in inputs: + key = f"usage-{kind.lower()}-{ref}" + resource.update( + self.rsp.desired.resources[key], + clusterusagev1beta1.ClusterUsage( + spec=clusterusagev1beta1.Spec( + of=clusterusagev1beta1.Of( + apiVersion="modelplane.ai/v1alpha1", + kind=kind, + resourceRef=clusterusagev1beta1.ResourceRef(name=ref), + ), + reason=f"InferenceCluster {name} uses this {kind}", + replayDeletion=True, + ), + ), + ) + self.rsp.desired.resources[key].ready = fnv1.READY_TRUE + def caches_on_cluster(self) -> list[mcv1alpha1.ModelCache]: """The ModelCaches staging onto this cluster. @@ -687,12 +741,17 @@ def resolve_gateway_cas(self) -> None: def resolve_classes(self) -> bool: """Declare and fetch every InferenceClass referenced by - spec.nodePools[].className (required in require_inputs). Returns False if - any is missing, in which case the function gates and waits.""" + spec.nodePools[].className (required in require_inputs). Returns False, + having failed the reconcile, if any is missing.""" class_names = sorted({p.className for p in (self.xr.spec.nodePools or [])}) missing: list[str] = [] for name in class_names: + # Crossplane only fetches a required resource after the function + # asks for it, and won't fetch at all if the result is fatal. An + # absent key means Crossplane hasn't fetched it yet, so wait for it. + if f"class-{name}" not in self.req.required_resources: + return False d = request.get_required_resource(self.req, f"class-{name}") if d is None: missing.append(name) @@ -700,19 +759,32 @@ def resolve_classes(self) -> bool: self.classes[name] = iclv1alpha1.InferenceClass.model_validate(d) if missing: - response.set_conditions( - self.rsp, - resource.Condition( - typ=CONDITION_TYPE_CLUSTER_READY, - status="False", - reason=CONDITION_REASON_WAITING_FOR_CLASSES, - message=f"Waiting for InferenceClasses: {', '.join(missing)}", - ), + self.fail(CONDITION_REASON_WAITING_FOR_CLASSES, f"Waiting for InferenceClasses: {', '.join(missing)}") + return False + + return True + + def resolve_reservations(self) -> bool: + """Fetch every CapacityReservation referenced by + spec.nodePools[].reservationRef (required in require_inputs). Returns + False, having failed the reconcile, if any is missing.""" + names = sorted({p.reservationRef.name for p in (self.xr.spec.nodePools or []) if p.reservationRef}) + + missing: list[str] = [] + for name in names: + # Wait for Crossplane to fetch it. See resolve_classes. + if f"reservation-{name}" not in self.req.required_resources: + return False + d = request.get_required_resource(self.req, f"reservation-{name}") + if d is None: + missing.append(name) + continue + self.reservations[name] = crv1alpha1.CapacityReservation.model_validate(d) + + if missing: + self.fail( + CONDITION_REASON_WAITING_FOR_RESERVATIONS, f"Waiting for CapacityReservations: {', '.join(missing)}" ) - response.normal(self.rsp, f"Waiting for InferenceClasses: {', '.join(missing)}") - # Only the guard and namespaces, both marked ready, are composed - # while waiting, so the XR would otherwise be ready. - self.rsp.desired.composite.ready = fnv1.READY_FALSE return False return True @@ -760,25 +832,29 @@ def check_devices_supported(self, source: Cloud) -> bool: if not offending: return True msg = f"{'; '.join(offending)}; {source} clusters support {', '.join(sorted(supported))}" - response.set_conditions( - self.rsp, - resource.Condition( - typ=CONDITION_TYPE_CLUSTER_READY, - status="False", - reason=CONDITION_REASON_UNSUPPORTED_DEVICES, - message=msg, - ), - ) # A fatal result, not a bare return: everything after this gate # composes the cluster, so returning an empty desired state would # have Crossplane garbage-collect a running cluster - the cloud # cluster XR, ServingStack and Usages - the moment a shared class - # starts claiming an unsupported driver. Crossplane stops before - # applying desired state or collecting anything on a fatal result, - # and still applies the condition above. - response.fatal(self.rsp, msg) + # starts claiming an unsupported driver. + self.fail(CONDITION_REASON_UNSUPPORTED_DEVICES, msg) return False + def fail(self, reason: str, msg: str) -> None: + """Stop composing the cluster, reporting why on its ClusterReady condition. + + The result is fatal, so Crossplane applies none of this reconcile's + desired state. Its caller returns before composing the cluster XR, and + Crossplane would otherwise delete an existing cluster XR that's missing + from desired state, for example because someone deleted an + InferenceClass or edited a node pool. + """ + response.set_conditions( + self.rsp, + resource.Condition(typ=CONDITION_TYPE_CLUSTER_READY, status="False", reason=reason, message=msg), + ) + response.fatal(self.rsp, msg) + def compose_gke(self, gke: v1alpha1.Gke | None) -> None: """Compose an InferenceCluster backed by a Modelplane-provisioned GKE cluster. Composes the GKECluster XR, waits for it to be ready, @@ -790,7 +866,8 @@ def compose_gke(self, gke: v1alpha1.Gke | None) -> None: self.compose_activation(_ACTIVATE_GCP) if self._activation_ready(_ACTIVATE_GCP) or "gke-cluster" in self.req.observed.resources: self.rsp.desired.resources["activation"].ready = fnv1.READY_TRUE - self.compose_gke_cluster(gke) + if not self.compose_gke_cluster(gke): + return gke_ready = resource.get_condition(self.req.observed.resources.get("gke-cluster"), "Ready").status == "True" kubeconfig_secret = self.observed_gke_secret(_SECRET_TYPE_KUBECONFIG) @@ -837,7 +914,8 @@ def compose_eks(self, eks: v1alpha1.Eks | None) -> None: self.compose_activation(_ACTIVATE_AWS) if self._activation_ready(_ACTIVATE_AWS) or "eks-cluster" in self.req.observed.resources: self.rsp.desired.resources["activation"].ready = fnv1.READY_TRUE - self.compose_eks_cluster(eks) + if not self.compose_eks_cluster(eks): + return eks_ready = resource.get_condition(self.req.observed.resources.get("eks-cluster"), "Ready").status == "True" kubeconfig = self.observed_eks_secret(_SECRET_TYPE_KUBECONFIG) @@ -876,7 +954,8 @@ def compose_aks(self, aks: v1alpha1.Aks | None) -> None: self.compose_activation(_ACTIVATE_AZURE) if self._activation_ready(_ACTIVATE_AZURE) or "aks-cluster" in self.req.observed.resources: self.rsp.desired.resources["activation"].ready = fnv1.READY_TRUE - self.compose_aks_cluster(aks) + if not self.compose_aks_cluster(aks): + return aks_ready = resource.get_condition(self.req.observed.resources.get("aks-cluster"), "Ready").status == "True" kubeconfig = self.observed_aks_secret(_SECRET_TYPE_KUBECONFIG) @@ -917,7 +996,8 @@ def compose_nebius(self, nebius: v1alpha1.Nebius | None) -> None: self.compose_activation(_ACTIVATE_NEBIUS) if self._activation_ready(_ACTIVATE_NEBIUS) or "nebius-cluster" in self.req.observed.resources: self.rsp.desired.resources["activation"].ready = fnv1.READY_TRUE - self.compose_nebius_cluster(nebius) + if not self.compose_nebius_cluster(nebius): + return nebius_ready = ( resource.get_condition(self.req.observed.resources.get("nebius-cluster"), "Ready").status == "True" @@ -964,7 +1044,8 @@ def compose_vultr(self, vultr: v1alpha1.Vultr | None) -> None: self.compose_activation(_ACTIVATE_VULTR) if self._activation_ready(_ACTIVATE_VULTR) or "vultr-cluster" in self.req.observed.resources: self.rsp.desired.resources["activation"].ready = fnv1.READY_TRUE - self.compose_vultr_cluster(vultr) + if not self.compose_vultr_cluster(vultr): + return vultr_ready = resource.get_condition(self.req.observed.resources.get("vultr-cluster"), "Ready").status == "True" kubeconfig = self.observed_vultr_secret(_SECRET_TYPE_KUBECONFIG) @@ -1003,7 +1084,8 @@ def compose_civo(self, civo: v1alpha1.Civo | None) -> None: self.compose_activation(_ACTIVATE_CIVO) if self._activation_ready(_ACTIVATE_CIVO) or "civo-cluster" in self.req.observed.resources: self.rsp.desired.resources["activation"].ready = fnv1.READY_TRUE - self.compose_civo_cluster(civo) + if not self.compose_civo_cluster(civo): + return civo_ready = resource.get_condition(self.req.observed.resources.get("civo-cluster"), "Ready").status == "True" kubeconfig = self.observed_civo_secret(_SECRET_TYPE_KUBECONFIG) @@ -1237,13 +1319,15 @@ def derive_conditions(self, *, cluster_ready: bool) -> None: ), ) - def compose_gke_cluster(self, gke: v1alpha1.Gke) -> None: + def compose_gke_cluster(self, gke: v1alpha1.Gke) -> bool: """Compose a GKECluster XR. Combines the cluster-level config (region) with the GPU pools derived from the user's node pools + referenced classes. The project is derived by compose-gke-cluster from the referenced ProviderConfig. The system pool is injected by compose-gke-cluster. + + Returns False, having failed the reconcile, if a node pool is invalid. """ gke_node_pools: list[gkev1alpha1.NodePool] = [] @@ -1251,17 +1335,8 @@ def compose_gke_cluster(self, gke: v1alpha1.Gke) -> None: cls = self.classes.get(pool.className) if not cls or not cls.spec.provisioning or not cls.spec.provisioning.gke: msg = f"InferenceClass {pool.className} has no GKE provisioning block" - response.set_conditions( - self.rsp, - resource.Condition( - typ=CONDITION_TYPE_CLUSTER_READY, - status="False", - reason=CONDITION_REASON_INVALID_NODE_POOL, - message=msg, - ), - ) - response.warning(self.rsp, msg) - return + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False prov = cls.spec.provisioning.gke # A class with no accelerator describes a CPU pool: a standard # node image, no GPU attachment, and the modelplane.ai/cpu @@ -1286,6 +1361,26 @@ def compose_gke_cluster(self, gke: v1alpha1.Gke) -> None: # empty list out of the GKECluster spec. if pool.zones: node_pool.zones = [gkev1alpha1.Zone(z) for z in pool.zones] + # Only set reservation when the pool has one, so an on-demand pool's + # GKECluster spec stays free of reservation: null. + if pool.reservationRef: + r = self.reservations[pool.reservationRef.name] + if r.spec.provider != CLUSTER_SOURCE_GKE or not r.spec.gke: + msg = f"CapacityReservation {pool.reservationRef.name} is for {r.spec.provider}, not GKE" + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False + # A specific reservation only has capacity in its own zone, so a + # pool that may create nodes in any other zone can't use it. + if pool.zones != [r.spec.gke.zone]: + msg = ( + f"Node pool {pool.name} must set zones to [{r.spec.gke.zone}] to use " + f"CapacityReservation {pool.reservationRef.name}" + ) + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False + node_pool.reservation = gkev1alpha1.Reservation(name=r.spec.gke.name) + if r.spec.gke.project: + node_pool.reservation.project = r.spec.gke.project gke_node_pools.append(node_pool) gke_spec = gkev1alpha1.Spec( @@ -1308,13 +1403,16 @@ def compose_gke_cluster(self, gke: v1alpha1.Gke) -> None: spec=gke_spec, ), ) + return True - def compose_eks_cluster(self, eks: v1alpha1.Eks) -> None: + def compose_eks_cluster(self, eks: v1alpha1.Eks) -> bool: """Compose an EKSCluster XR. Combines the cluster-level config (region) with GPU node pools derived from the user's node pools + referenced classes. The system pool is injected by compose-eks-cluster. + + Returns False, having failed the reconcile, if a node pool is invalid. """ eks_node_pools: list[eksv1alpha1.NodePool] = [] @@ -1322,17 +1420,8 @@ def compose_eks_cluster(self, eks: v1alpha1.Eks) -> None: cls = self.classes.get(pool.className) if not cls or not cls.spec.provisioning or not cls.spec.provisioning.eks: msg = f"InferenceClass {pool.className} has no EKS provisioning block" - response.set_conditions( - self.rsp, - resource.Condition( - typ=CONDITION_TYPE_CLUSTER_READY, - status="False", - reason=CONDITION_REASON_INVALID_NODE_POOL, - message=msg, - ), - ) - response.warning(self.rsp, msg) - return + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False prov = cls.spec.provisioning.eks # A class with no accelerator describes a CPU pool (see the GKE # builder above). @@ -1354,12 +1443,25 @@ def compose_eks_cluster(self, eks: v1alpha1.Eks) -> None: # serializes with exclude_unset). if pool.zones: node_pool.zones = [eksv1alpha1.Zone(z) for z in pool.zones] - # Only set capacityBlock when the pool has one. resource.update - # serializes with exclude_unset, so leaving it unset keeps it out - # of the EKSCluster spec rather than emitting capacityBlock: null. - if pool.capacityBlock: - node_pool.capacityBlock = eksv1alpha1.CapacityBlock( - capacityReservationId=pool.capacityBlock.capacityReservationId, + # Only set capacityReservation when the pool has one. resource.update + # serializes with exclude_unset, so leaving it unset keeps it out of + # the EKSCluster spec rather than emitting capacityReservation: null. + if pool.reservationRef: + r = self.reservations[pool.reservationRef.name] + if r.spec.provider != CLUSTER_SOURCE_EKS or not r.spec.eks: + msg = f"CapacityReservation {pool.reservationRef.name} is for {r.spec.provider}, not EKS" + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False + if r.spec.eks.region != eks.region: + msg = ( + f"CapacityReservation {pool.reservationRef.name} is in region {r.spec.eks.region}, " + f"but the cluster is in {eks.region}" + ) + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False + node_pool.capacityReservation = eksv1alpha1.CapacityReservation( + type=r.spec.eks.type, + id=r.spec.eks.id, ) # Likewise only set fabric when the pool opts into one, so an # on-demand pool's EKSCluster spec stays free of fabric: None. @@ -1387,13 +1489,16 @@ def compose_eks_cluster(self, eks: v1alpha1.Eks) -> None: spec=eks_spec, ), ) + return True - def compose_aks_cluster(self, aks: v1alpha1.Aks) -> None: + def compose_aks_cluster(self, aks: v1alpha1.Aks) -> bool: """Compose an AKSCluster XR. Combines the cluster-level config (location) with GPU node pools derived from the user's node pools + referenced classes. The system pool is injected by compose-aks-cluster. + + Returns False, having failed the reconcile, if a node pool is invalid. """ aks_node_pools: list[aksv1alpha1.NodePool] = [] @@ -1401,17 +1506,8 @@ def compose_aks_cluster(self, aks: v1alpha1.Aks) -> None: cls = self.classes.get(pool.className) if not cls or not cls.spec.provisioning or not cls.spec.provisioning.aks: msg = f"InferenceClass {pool.className} has no AKS provisioning block" - response.set_conditions( - self.rsp, - resource.Condition( - typ=CONDITION_TYPE_CLUSTER_READY, - status="False", - reason=CONDITION_REASON_INVALID_NODE_POOL, - message=msg, - ), - ) - response.warning(self.rsp, msg) - return + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False prov = cls.spec.provisioning.aks # A class with no accelerator describes a CPU pool (see the GKE # builder above). @@ -1461,13 +1557,16 @@ def compose_aks_cluster(self, aks: v1alpha1.Aks) -> None: spec=aks_spec, ), ) + return True - def compose_nebius_cluster(self, nebius: v1alpha1.Nebius) -> None: + def compose_nebius_cluster(self, nebius: v1alpha1.Nebius) -> bool: """Compose a NebiusCluster XR. Combines the cluster-level config (project, credentials) with GPU node pools derived from the user's node pools + referenced classes. The system pool is injected by compose-nebius-cluster. + + Returns False, having failed the reconcile, if a node pool is invalid. """ nebius_node_pools: list[nebiusv1alpha1.NodePool] = [] @@ -1475,17 +1574,8 @@ def compose_nebius_cluster(self, nebius: v1alpha1.Nebius) -> None: cls = self.classes.get(pool.className) if not cls or not cls.spec.provisioning or not cls.spec.provisioning.nebius: msg = f"InferenceClass {pool.className} has no Nebius provisioning block" - response.set_conditions( - self.rsp, - resource.Condition( - typ=CONDITION_TYPE_CLUSTER_READY, - status="False", - reason=CONDITION_REASON_INVALID_NODE_POOL, - message=msg, - ), - ) - response.warning(self.rsp, msg) - return + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False prov = cls.spec.provisioning.nebius # A class with no accelerator describes a CPU pool (see the GKE # builder above). driversPreset rides the gpu block: a CPU @@ -1542,6 +1632,7 @@ def compose_nebius_cluster(self, nebius: v1alpha1.Nebius) -> None: spec=spec, ), ) + return True def compose_nebius_usage(self) -> None: """Block NebiusCluster deletion until the backend is deleted.""" @@ -1623,12 +1714,14 @@ def observed_nebius_secret(self, secret_type: str) -> nebiusv1alpha1.Secret | No return None return next((s for s in nebius_secrets if s.type == secret_type), None) - def compose_vultr_cluster(self, vultr: v1alpha1.Vultr) -> None: + def compose_vultr_cluster(self, vultr: v1alpha1.Vultr) -> bool: """Compose a VultrCluster XR. Combines the cluster-level config (region) with GPU node pools derived from the user's node pools + referenced classes. The system pool is injected by compose-vultr-cluster. + + Returns False, having failed the reconcile, if a node pool is invalid. """ vultr_node_pools: list[vultrv1alpha1.NodePool] = [] @@ -1636,17 +1729,8 @@ def compose_vultr_cluster(self, vultr: v1alpha1.Vultr) -> None: cls = self.classes.get(pool.className) if not cls or not cls.spec.provisioning or not cls.spec.provisioning.vultr: msg = f"InferenceClass {pool.className} has no Vultr provisioning block" - response.set_conditions( - self.rsp, - resource.Condition( - typ=CONDITION_TYPE_CLUSTER_READY, - status="False", - reason=CONDITION_REASON_INVALID_NODE_POOL, - message=msg, - ), - ) - response.warning(self.rsp, msg) - return + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False prov = cls.spec.provisioning.vultr # A class with no accelerator describes a CPU pool (see the GKE # builder above). @@ -1691,6 +1775,7 @@ def compose_vultr_cluster(self, vultr: v1alpha1.Vultr) -> None: spec=vultr_spec, ), ) + return True def compose_vultr_usage(self) -> None: """Block VultrCluster deletion until the backend is deleted.""" @@ -1773,12 +1858,14 @@ def civo_nvlink_disabled_pools(self) -> list[str]: pools.append(pool.name) return pools - def compose_civo_cluster(self, civo: v1alpha1.Civo) -> None: + def compose_civo_cluster(self, civo: v1alpha1.Civo) -> bool: """Compose a CivoCluster XR. Combines the cluster-level config (region) with GPU node pools derived from the user's node pools + referenced classes. The system pool is injected by compose-civo-cluster. + + Returns False, having failed the reconcile, if a node pool is invalid. """ civo_node_pools: list[civov1alpha1.NodePool] = [] @@ -1786,17 +1873,8 @@ def compose_civo_cluster(self, civo: v1alpha1.Civo) -> None: cls = self.classes.get(pool.className) if not cls or not cls.spec.provisioning or not cls.spec.provisioning.civo: msg = f"InferenceClass {pool.className} has no Civo provisioning block" - response.set_conditions( - self.rsp, - resource.Condition( - typ=CONDITION_TYPE_CLUSTER_READY, - status="False", - reason=CONDITION_REASON_INVALID_NODE_POOL, - message=msg, - ), - ) - response.warning(self.rsp, msg) - return + self.fail(CONDITION_REASON_INVALID_NODE_POOL, msg) + return False prov = cls.spec.provisioning.civo # A class with no accelerator describes a CPU pool (see the GKE # builder above). @@ -1844,6 +1922,7 @@ def compose_civo_cluster(self, civo: v1alpha1.Civo) -> None: spec=civo_spec, ), ) + return True def compose_civo_usage(self) -> None: """Block CivoCluster deletion until the backend is deleted.""" diff --git a/functions/compose-inference-cluster/tests/test_fn.py b/functions/compose-inference-cluster/tests/test_fn.py index 6a9e7104d..467e41e83 100644 --- a/functions/compose-inference-cluster/tests/test_fn.py +++ b/functions/compose-inference-cluster/tests/test_fn.py @@ -129,6 +129,20 @@ def _cpu_inference_class(*, name: str, cpus: str, provisioning: dict) -> fnv1.Re ) +def _capacity_reservation(*, name: str, spec: dict) -> fnv1.Resource: + """A CapacityReservation with spec, as its reservation requirement returns it.""" + return fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "modelplane.ai/v1alpha1", + "kind": "CapacityReservation", + "metadata": {"name": name}, + "spec": spec, + } + ) + ) + + def _inference_gateway(*, name: str, cluster: str, status: dict | None) -> fnv1.Resource: """An InferenceGateway on cluster, as the gateways requirement returns it, with status unless it's None.""" gateway: dict = { @@ -198,24 +212,26 @@ def _activation_policy(*, activate: list[str], ready: fnv1.Ready) -> fnv1.Resour ) -def _gke_cluster(*, credentials: dict | None, ready: fnv1.Ready) -> fnv1.Resource: - """The composed GKECluster with l4-pool, using credentials unless they're None.""" +def _gke_cluster(*, credentials: dict | None, reservation: dict | None, ready: fnv1.Ready) -> fnv1.Resource: + """The composed GKECluster with l4-pool, using credentials and backing l4-pool with reservation unless they're + None.""" + pool: dict = { + "name": "l4-pool", + "role": "GPU", + "machineType": "g2-standard-48", + "nodeCount": 2, + "minNodeCount": None, + "maxNodeCount": 4, + "diskSizeGb": 100, + "gpu": {"acceleratorType": "nvidia-l4", "acceleratorCount": 1}, + "zones": ["us-central1-a"], + } + if reservation is not None: + pool["reservation"] = reservation spec: dict = { "region": "us-central1", "kubernetesVersion": "1.35", - "nodePools": [ - { - "name": "l4-pool", - "role": "GPU", - "machineType": "g2-standard-48", - "nodeCount": 2, - "minNodeCount": None, - "maxNodeCount": 4, - "diskSizeGb": 100, - "gpu": {"acceleratorType": "nvidia-l4", "acceleratorCount": 1}, - "zones": ["us-central1-a"], - }, - ], + "nodePools": [pool], } if credentials is not None: spec["credentials"] = credentials @@ -233,9 +249,10 @@ def _gke_cluster(*, credentials: dict | None, ready: fnv1.Ready) -> fnv1.Resourc def _eks_cluster( - *, zones: list[str], capacity_block: dict | None, fabric: str | None, ready: fnv1.Ready + *, zones: list[str], capacity_reservation: dict | None, fabric: str | None, ready: fnv1.Ready ) -> fnv1.Resource: - """The composed EKSCluster with l4-pool in zones, setting its capacityBlock and fabric unless they're None.""" + """The composed EKSCluster with l4-pool in zones, setting its capacityReservation and fabric unless they're + None.""" pool: dict = { "name": "l4-pool", "role": "GPU", @@ -247,8 +264,8 @@ def _eks_cluster( "gpu": {"acceleratorType": "nvidia-l4"}, "zones": zones, } - if capacity_block is not None: - pool["capacityBlock"] = capacity_block + if capacity_reservation is not None: + pool["capacityReservation"] = capacity_reservation if fabric is not None: pool["fabric"] = fabric return fnv1.Resource( @@ -451,6 +468,24 @@ def _guard_cluster_usage(*, reason: str) -> fnv1.Resource: ) +def _input_usage(*, kind: str, name: str) -> fnv1.Resource: + """The ClusterUsage blocking deletion of an InferenceClass or CapacityReservation test-cluster uses.""" + return fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "protection.crossplane.io/v1beta1", + "kind": "ClusterUsage", + "spec": { + "of": {"apiVersion": "modelplane.ai/v1alpha1", "kind": kind, "resourceRef": {"name": name}}, + "reason": f"InferenceCluster test-cluster uses this {kind}", + "replayDeletion": True, + }, + } + ), + ready=fnv1.READY_TRUE, + ) + + def _namespace_object(*, team: str, name: str) -> fnv1.Resource: """The composed Object that mirrors team's namespace onto the cluster as the Namespace name.""" return fnv1.Resource( @@ -635,6 +670,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -740,6 +776,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity={ @@ -870,23 +907,670 @@ def _to_dict(msg: message.Message) -> dict: cache=None, gateway=None, ), - resources={ - "activation": _activation_policy( - activate=[ - "projectiammembers.cloudplatform.gcp.m.upbound.io", - "projectservices.cloudplatform.gcp.m.upbound.io", - "serviceaccounts.cloudplatform.gcp.m.upbound.io", - "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", - "networks.compute.gcp.m.upbound.io", - "subnetworks.compute.gcp.m.upbound.io", - "clusters.container.gcp.m.upbound.io", - "nodepools.container.gcp.m.upbound.io", - ], - ready=fnv1.READY_TRUE, - ), - "gke-cluster": _gke_cluster(credentials=None, ready=fnv1.READY_UNSPECIFIED), - }, - ), + resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), + "activation": _activation_policy( + activate=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ], + ready=fnv1.READY_TRUE, + ), + "gke-cluster": _gke_cluster(credentials=None, reservation=None, ready=fnv1.READY_UNSPECIFIED), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "model-replicas": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelReplica", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-routes": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelRoute", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-caches": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="ModelCache"), + "gateways": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceGateway"), + "class-gpu-l4": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4" + ), + }, + ), + conditions=[ + fnv1.Condition(type="ClusterReady", status=fnv1.STATUS_CONDITION_FALSE, reason="Provisioning"), + fnv1.Condition(type="BackendReady", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForCluster"), + ], + ), + ), + ComposeCase( + name="GKEClassWrongProvider", + reason="A GKE node pool whose InferenceClass has no GKE provisioning block is rejected with a fatal result.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_inference_cluster( + cluster=v1alpha1.Cluster(source="GKE", gke=v1alpha1.Gke(region="us-central1")), + node_pools=[ + v1alpha1.NodePool( + name="l4-pool", + className="gpu-l4", + nodeCount=2, + maxNodeCount=4, + zones=["us-central1-a"], + ), + ], + ), + resources={ + "activation": _observed_activation_policy( + activated=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ] + ), + }, + ), + required_resources={ + "class-gpu-l4": fnv1.Resources( + items=[ + _inference_class( + name="gpu-l4", + count=1, + memory="24Gi", + provisioning={ + "provider": "EKS", + "eks": { + "instanceType": "g6.xlarge", + "diskSizeGb": 100, + "accelerator": {"type": "nvidia-l4", "count": 1}, + }, + }, + ), + ], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), + "activation": _activation_policy( + activate=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ], + ready=fnv1.READY_TRUE, + ), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "model-replicas": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelReplica", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-routes": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelRoute", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-caches": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="ModelCache"), + "gateways": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceGateway"), + "class-gpu-l4": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4" + ), + }, + ), + results=[ + fnv1.Result(severity=fnv1.SEVERITY_FATAL, message="InferenceClass gpu-l4 has no GKE provisioning block") + ], + conditions=[ + fnv1.Condition( + type="ClusterReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="InvalidNodePool", + message="InferenceClass gpu-l4 has no GKE provisioning block", + ), + ], + ), + ), + ComposeCase( + name="GKEReservation", + reason="A node pool referencing a GKE CapacityReservation passes the reservation's name through to the GKECluster pool.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_inference_cluster( + cluster=v1alpha1.Cluster(source="GKE", gke=v1alpha1.Gke(region="us-central1")), + node_pools=[ + v1alpha1.NodePool( + name="l4-pool", + className="gpu-l4", + nodeCount=2, + maxNodeCount=4, + zones=["us-central1-a"], + reservationRef=v1alpha1.ReservationRef(name="l4-calendar"), + ), + ], + ), + resources={ + "activation": _observed_activation_policy( + activated=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ] + ), + }, + ), + required_resources={ + "class-gpu-l4": fnv1.Resources( + items=[ + _inference_class( + name="gpu-l4", + count=1, + memory="24Gi", + provisioning={ + "provider": "GKE", + "gke": { + "machineType": "g2-standard-48", + "diskSizeGb": 100, + "accelerator": {"type": "nvidia-l4", "count": 1}, + }, + }, + ), + ], + ), + "reservation-l4-calendar": fnv1.Resources( + items=[ + _capacity_reservation( + name="l4-calendar", + spec={ + "provider": "GKE", + "gke": {"zone": "us-central1-a", "name": "l4-calendar"}, + }, + ), + ], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + composite=_desired_inference_cluster( + schedulable_pools=[ + { + "name": "l4-pool", + "nodes": 4, + "devices": [ + { + "name": "gpu", + "claim": "DRA", + "driver": "gpu.nvidia.com", + "deviceClassName": "gpu.nvidia.com", + "count": 1, + "capacity": {"memory": {"value": "24Gi"}}, + }, + ], + }, + ], + cache=None, + gateway=None, + ), + resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), + "usage-capacityreservation-l4-calendar": _input_usage( + kind="CapacityReservation", name="l4-calendar" + ), + "activation": _activation_policy( + activate=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ], + ready=fnv1.READY_TRUE, + ), + "gke-cluster": _gke_cluster( + credentials=None, + reservation={"name": "l4-calendar"}, + ready=fnv1.READY_UNSPECIFIED, + ), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "model-replicas": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelReplica", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-routes": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelRoute", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-caches": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="ModelCache"), + "gateways": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceGateway"), + "class-gpu-l4": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4" + ), + "reservation-l4-calendar": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="CapacityReservation", match_name="l4-calendar" + ), + }, + ), + conditions=[ + fnv1.Condition(type="ClusterReady", status=fnv1.STATUS_CONDITION_FALSE, reason="Provisioning"), + fnv1.Condition(type="BackendReady", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForCluster"), + ], + ), + ), + # An invalid node pool fails the reconcile, so Crossplane applies none of its + # desired state, and an existing GKECluster isn't deleted for being missing + # from it. Crossplane still sets the conditions. + ComposeCase( + name="GKEReservationZoneMismatch", + reason="A node pool referencing a GKE CapacityReservation in a zone it isn't restricted to is rejected with a fatal result.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_inference_cluster( + cluster=v1alpha1.Cluster(source="GKE", gke=v1alpha1.Gke(region="us-central1")), + node_pools=[ + v1alpha1.NodePool( + name="l4-pool", + className="gpu-l4", + nodeCount=2, + maxNodeCount=4, + zones=["us-central1-a", "us-central1-b"], + reservationRef=v1alpha1.ReservationRef(name="l4-calendar"), + ), + ], + ), + resources={ + "activation": _observed_activation_policy( + activated=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ] + ), + }, + ), + required_resources={ + "class-gpu-l4": fnv1.Resources( + items=[ + _inference_class( + name="gpu-l4", + count=1, + memory="24Gi", + provisioning={ + "provider": "GKE", + "gke": { + "machineType": "g2-standard-48", + "diskSizeGb": 100, + "accelerator": {"type": "nvidia-l4", "count": 1}, + }, + }, + ), + ], + ), + "reservation-l4-calendar": fnv1.Resources( + items=[ + _capacity_reservation( + name="l4-calendar", + spec={"provider": "GKE", "gke": {"zone": "us-central1-a", "name": "l4-calendar"}}, + ), + ], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), + "usage-capacityreservation-l4-calendar": _input_usage( + kind="CapacityReservation", name="l4-calendar" + ), + "activation": _activation_policy( + activate=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ], + ready=fnv1.READY_TRUE, + ), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "model-replicas": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelReplica", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-routes": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelRoute", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-caches": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="ModelCache"), + "gateways": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceGateway"), + "class-gpu-l4": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4" + ), + "reservation-l4-calendar": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="CapacityReservation", match_name="l4-calendar" + ), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_FATAL, + message="Node pool l4-pool must set zones to [us-central1-a] to use CapacityReservation l4-calendar", + ), + ], + conditions=[ + fnv1.Condition( + type="ClusterReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="InvalidNodePool", + message="Node pool l4-pool must set zones to [us-central1-a] to use CapacityReservation l4-calendar", + ), + ], + ), + ), + ComposeCase( + name="GKEReservationWrongProvider", + reason="A GKE node pool referencing an EKS CapacityReservation is rejected with a fatal result.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_inference_cluster( + cluster=v1alpha1.Cluster(source="GKE", gke=v1alpha1.Gke(region="us-central1")), + node_pools=[ + v1alpha1.NodePool( + name="l4-pool", + className="gpu-l4", + nodeCount=2, + maxNodeCount=4, + zones=["us-central1-a"], + reservationRef=v1alpha1.ReservationRef(name="l4-block"), + ), + ], + ), + resources={ + "activation": _observed_activation_policy( + activated=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ] + ), + }, + ), + required_resources={ + "class-gpu-l4": fnv1.Resources( + items=[ + _inference_class( + name="gpu-l4", + count=1, + memory="24Gi", + provisioning={ + "provider": "GKE", + "gke": { + "machineType": "g2-standard-48", + "diskSizeGb": 100, + "accelerator": {"type": "nvidia-l4", "count": 1}, + }, + }, + ), + ], + ), + "reservation-l4-block": fnv1.Resources( + items=[ + _capacity_reservation( + name="l4-block", + spec={ + "provider": "EKS", + "eks": {"region": "us-west-2", "type": "CapacityBlock", "id": "cr-0123456789abcdef0"}, + }, + ), + ], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), + "usage-capacityreservation-l4-block": _input_usage(kind="CapacityReservation", name="l4-block"), + "activation": _activation_policy( + activate=[ + "projectiammembers.cloudplatform.gcp.m.upbound.io", + "projectservices.cloudplatform.gcp.m.upbound.io", + "serviceaccounts.cloudplatform.gcp.m.upbound.io", + "serviceaccountkeys.cloudplatform.gcp.m.upbound.io", + "networks.compute.gcp.m.upbound.io", + "subnetworks.compute.gcp.m.upbound.io", + "clusters.container.gcp.m.upbound.io", + "nodepools.container.gcp.m.upbound.io", + ], + ready=fnv1.READY_TRUE, + ), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "model-replicas": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelReplica", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-routes": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelRoute", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-caches": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="ModelCache"), + "gateways": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceGateway"), + "class-gpu-l4": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4" + ), + "reservation-l4-block": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="CapacityReservation", match_name="l4-block" + ), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_FATAL, + message="CapacityReservation l4-block is for EKS, not GKE", + ), + ], + conditions=[ + fnv1.Condition( + type="ClusterReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="InvalidNodePool", + message="CapacityReservation l4-block is for EKS, not GKE", + ), + ], + ), + ), + ComposeCase( + name="GKEReservationMissing", + reason="A node pool referencing a CapacityReservation that doesn't exist fails the reconcile until the reservation exists.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_inference_cluster( + cluster=v1alpha1.Cluster(source="GKE", gke=v1alpha1.Gke(region="us-central1")), + node_pools=[ + v1alpha1.NodePool( + name="l4-pool", + className="gpu-l4", + nodeCount=2, + maxNodeCount=4, + zones=["us-central1-a"], + reservationRef=v1alpha1.ReservationRef(name="l4-calendar"), + ), + ], + ), + ), + required_resources={ + "class-gpu-l4": fnv1.Resources( + items=[ + _inference_class( + name="gpu-l4", + count=1, + memory="24Gi", + provisioning={ + "provider": "GKE", + "gke": { + "machineType": "g2-standard-48", + "diskSizeGb": 100, + "accelerator": {"type": "nvidia-l4", "count": 1}, + }, + }, + ), + ], + ), + "reservation-l4-calendar": fnv1.Resources(), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State(), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "model-replicas": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelReplica", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-routes": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelRoute", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-caches": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="ModelCache"), + "gateways": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceGateway"), + "class-gpu-l4": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4" + ), + "reservation-l4-calendar": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="CapacityReservation", match_name="l4-calendar" + ), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_FATAL, + message="Waiting for CapacityReservations: l4-calendar", + ), + ], + conditions=[ + fnv1.Condition( + type="ClusterReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="WaitingForReservations", + message="Waiting for CapacityReservations: l4-calendar", + ), + ], + ), + ), + ComposeCase( + name="ReservationNotYetFetched", + reason="Before Crossplane has fetched its CapacityReservation, the function asks for it and waits, without failing the reconcile.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_inference_cluster( + cluster=v1alpha1.Cluster(source="GKE", gke=v1alpha1.Gke(region="us-central1")), + node_pools=[ + v1alpha1.NodePool( + name="l4-pool", + className="gpu-l4", + nodeCount=2, + maxNodeCount=4, + zones=["us-central1-a"], + reservationRef=v1alpha1.ReservationRef(name="l4-calendar"), + ), + ], + ), + ), + required_resources={ + "class-gpu-l4": fnv1.Resources( + items=[ + _inference_class( + name="gpu-l4", + count=1, + memory="24Gi", + provisioning={ + "provider": "GKE", + "gke": { + "machineType": "g2-standard-48", + "diskSizeGb": 100, + "accelerator": {"type": "nvidia-l4", "count": 1}, + }, + }, + ), + ], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State(), context=structpb.Struct(), requirements=fnv1.Requirements( resources={ @@ -905,12 +1589,11 @@ def _to_dict(msg: message.Message) -> dict: "class-gpu-l4": fnv1.ResourceSelector( api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4" ), + "reservation-l4-calendar": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="CapacityReservation", match_name="l4-calendar" + ), }, ), - conditions=[ - fnv1.Condition(type="ClusterReady", status=fnv1.STATUS_CONDITION_FALSE, reason="Provisioning"), - fnv1.Condition(type="BackendReady", status=fnv1.STATUS_CONDITION_FALSE, reason="WaitingForCluster"), - ], ), ), ComposeCase( @@ -995,6 +1678,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "activation": _activation_policy( activate=[ "projectiammembers.cloudplatform.gcp.m.upbound.io", @@ -1009,7 +1693,9 @@ def _to_dict(msg: message.Message) -> dict: ready=fnv1.READY_TRUE, ), "gke-cluster": _gke_cluster( - credentials={"type": "ProviderConfig", "name": "my-gcp-account"}, ready=fnv1.READY_UNSPECIFIED + credentials={"type": "ProviderConfig", "name": "my-gcp-account"}, + reservation=None, + ready=fnv1.READY_UNSPECIFIED, ), }, ), @@ -1101,6 +1787,7 @@ def _to_dict(msg: message.Message) -> dict: gateway={"address": "34.55.100.10"}, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -1249,6 +1936,8 @@ def _to_dict(msg: message.Message) -> dict: gateway={"address": "34.55.100.10"}, ), resources={ + "usage-inferenceclass-cpu": _input_usage(kind="InferenceClass", name="cpu"), + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -1390,6 +2079,7 @@ def _to_dict(msg: message.Message) -> dict: gateway={"address": "34.55.100.10"}, ), resources={ + "usage-inferenceclass-gpu-l4x4": _input_usage(kind="InferenceClass", name="gpu-l4x4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -1519,6 +2209,7 @@ def _to_dict(msg: message.Message) -> dict: gateway={"address": "34.55.100.10"}, ), resources={ + "usage-inferenceclass-gpu-l4x4": _input_usage(kind="InferenceClass", name="gpu-l4x4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -1657,6 +2348,7 @@ def _to_dict(msg: message.Message) -> dict: gateway={"address": "34.55.100.10"}, ), resources={ + "usage-inferenceclass-gpu-l4x4": _input_usage(kind="InferenceClass", name="gpu-l4x4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -1791,6 +2483,7 @@ def _to_dict(msg: message.Message) -> dict: gateway={"address": "34.55.100.10"}, ), resources={ + "usage-inferenceclass-gpu-l4x4": _input_usage(kind="InferenceClass", name="gpu-l4x4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -1946,6 +2639,7 @@ def _to_dict(msg: message.Message) -> dict: gateway={"address": "34.55.100.10"}, ), resources={ + "usage-inferenceclass-gpu-l4x4": _input_usage(kind="InferenceClass", name="gpu-l4x4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -2076,6 +2770,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4-eks": _input_usage(kind="InferenceClass", name="gpu-l4-eks"), "activation": _activation_policy( activate=[ "eips.ec2.aws.m.upbound.io", @@ -2105,7 +2800,7 @@ def _to_dict(msg: message.Message) -> dict: ), "eks-cluster": _eks_cluster( zones=["us-west-2a", "us-west-2b"], - capacity_block=None, + capacity_reservation=None, fabric=None, ready=fnv1.READY_UNSPECIFIED, ), @@ -2249,6 +2944,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4-eks": _input_usage(kind="InferenceClass", name="gpu-l4-eks"), "activation": _activation_policy( activate=[ "eips.ec2.aws.m.upbound.io", @@ -2278,7 +2974,7 @@ def _to_dict(msg: message.Message) -> dict: ), "eks-cluster": _eks_cluster( zones=["us-west-2a", "us-west-2b"], - capacity_block=None, + capacity_reservation=None, fabric=None, ready=fnv1.READY_UNSPECIFIED, ), @@ -2424,6 +3120,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "activation": _activation_policy( activate=[ "projectiammembers.cloudplatform.gcp.m.upbound.io", @@ -2437,7 +3134,7 @@ def _to_dict(msg: message.Message) -> dict: ], ready=fnv1.READY_TRUE, ), - "gke-cluster": _gke_cluster(credentials=None, ready=fnv1.READY_TRUE), + "gke-cluster": _gke_cluster(credentials=None, reservation=None, ready=fnv1.READY_TRUE), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="test-cluster-kubeconfig-abcde", identity={ @@ -2618,6 +3315,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4-eks": _input_usage(kind="InferenceClass", name="gpu-l4-eks"), "activation": _activation_policy( activate=[ "eips.ec2.aws.m.upbound.io", @@ -2646,7 +3344,10 @@ def _to_dict(msg: message.Message) -> dict: ready=fnv1.READY_TRUE, ), "eks-cluster": _eks_cluster( - zones=["us-west-2a", "us-west-2b"], capacity_block=None, fabric=None, ready=fnv1.READY_TRUE + zones=["us-west-2a", "us-west-2b"], + capacity_reservation=None, + fabric=None, + ready=fnv1.READY_TRUE, ), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="test-cluster-kubeconfig-abcde", identity=None @@ -2689,11 +3390,11 @@ def _to_dict(msg: message.Message) -> dict: ], ), ), - # compose-eks-cluster turns the pool's capacityBlock into a CAPACITY_BLOCK - # node group. + # compose-eks-cluster turns the pool's capacityReservation into a + # CAPACITY_BLOCK node group. ComposeCase( name="EKSCapacityBlock", - reason="A node pool backed by a Capacity Block passes its reservation ID through to the EKSCluster pool's capacityBlock.", + reason="A node pool referencing a Capacity Block CapacityReservation passes its type and ID through to the EKSCluster pool's capacityReservation.", req=fnv1.RunFunctionRequest( observed=fnv1.State( composite=_inference_cluster( @@ -2705,9 +3406,7 @@ def _to_dict(msg: message.Message) -> dict: nodeCount=2, maxNodeCount=4, zones=["us-west-2a"], - capacityBlock=v1alpha1.CapacityBlock( - capacityReservationId="cr-0123456789abcdef0", - ), + reservationRef=v1alpha1.ReservationRef(name="l4-block"), ), ], ), @@ -2758,6 +3457,17 @@ def _to_dict(msg: message.Message) -> dict: ), ], ), + "reservation-l4-block": fnv1.Resources( + items=[ + _capacity_reservation( + name="l4-block", + spec={ + "provider": "EKS", + "eks": {"region": "us-west-2", "type": "CapacityBlock", "id": "cr-0123456789abcdef0"}, + }, + ), + ], + ), }, ), want=fnv1.RunFunctionResponse( @@ -2784,6 +3494,8 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4-eks": _input_usage(kind="InferenceClass", name="gpu-l4-eks"), + "usage-capacityreservation-l4-block": _input_usage(kind="CapacityReservation", name="l4-block"), "activation": _activation_policy( activate=[ "eips.ec2.aws.m.upbound.io", @@ -2813,7 +3525,7 @@ def _to_dict(msg: message.Message) -> dict: ), "eks-cluster": _eks_cluster( zones=["us-west-2a"], - capacity_block={"capacityReservationId": "cr-0123456789abcdef0"}, + capacity_reservation={"type": "CapacityBlock", "id": "cr-0123456789abcdef0"}, fabric=None, ready=fnv1.READY_UNSPECIFIED, ), @@ -2837,6 +3549,9 @@ def _to_dict(msg: message.Message) -> dict: "class-gpu-l4-eks": fnv1.ResourceSelector( api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4-eks" ), + "reservation-l4-block": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="CapacityReservation", match_name="l4-block" + ), }, ), conditions=[ @@ -2847,6 +3562,160 @@ def _to_dict(msg: message.Message) -> dict: ), # compose-eks-cluster turns the pool's fabric into EFA launch-template # interfaces. + ComposeCase( + name="EKSReservationWrongRegion", + reason="A node pool referencing a CapacityReservation in another region is rejected with a fatal result.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_inference_cluster( + cluster=v1alpha1.Cluster(source="EKS", eks=v1alpha1.Eks(region="us-west-2")), + node_pools=[ + v1alpha1.NodePool( + name="l4-pool", + className="gpu-l4-eks", + nodeCount=2, + maxNodeCount=4, + zones=["us-west-2a"], + reservationRef=v1alpha1.ReservationRef(name="l4-block"), + ), + ], + ), + resources={ + "activation": _observed_activation_policy( + activated=[ + "eips.ec2.aws.m.upbound.io", + "internetgateways.ec2.aws.m.upbound.io", + "launchtemplates.ec2.aws.m.upbound.io", + "natgateways.ec2.aws.m.upbound.io", + "routes.ec2.aws.m.upbound.io", + "routetables.ec2.aws.m.upbound.io", + "routetableassociations.ec2.aws.m.upbound.io", + "securitygroups.ec2.aws.m.upbound.io", + "securitygroupegressrules.ec2.aws.m.upbound.io", + "securitygroupingressrules.ec2.aws.m.upbound.io", + "subnets.ec2.aws.m.upbound.io", + "vpcs.ec2.aws.m.upbound.io", + "filesystems.efs.aws.m.upbound.io", + "mounttargets.efs.aws.m.upbound.io", + "addons.eks.aws.m.upbound.io", + "clusters.eks.aws.m.upbound.io", + "clusterauths.eks.aws.m.upbound.io", + "nodegroups.eks.aws.m.upbound.io", + "podidentityassociations.eks.aws.m.upbound.io", + "policies.iam.aws.m.upbound.io", + "roles.iam.aws.m.upbound.io", + "rolepolicyattachments.iam.aws.m.upbound.io", + ] + ), + }, + ), + required_resources={ + "class-gpu-l4-eks": fnv1.Resources( + items=[ + _inference_class( + name="gpu-l4-eks", + count=1, + memory="24Gi", + provisioning={ + "provider": "EKS", + "eks": { + "instanceType": "g6.xlarge", + "diskSizeGb": 100, + "accelerator": {"type": "nvidia-l4", "count": 1}, + }, + }, + ), + ], + ), + "reservation-l4-block": fnv1.Resources( + items=[ + _capacity_reservation( + name="l4-block", + spec={ + "provider": "EKS", + "eks": {"region": "us-east-2", "type": "CapacityBlock", "id": "cr-0123456789abcdef0"}, + }, + ), + ], + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "usage-inferenceclass-gpu-l4-eks": _input_usage(kind="InferenceClass", name="gpu-l4-eks"), + "usage-capacityreservation-l4-block": _input_usage(kind="CapacityReservation", name="l4-block"), + "activation": _activation_policy( + activate=[ + "eips.ec2.aws.m.upbound.io", + "internetgateways.ec2.aws.m.upbound.io", + "launchtemplates.ec2.aws.m.upbound.io", + "natgateways.ec2.aws.m.upbound.io", + "routes.ec2.aws.m.upbound.io", + "routetables.ec2.aws.m.upbound.io", + "routetableassociations.ec2.aws.m.upbound.io", + "securitygroups.ec2.aws.m.upbound.io", + "securitygroupegressrules.ec2.aws.m.upbound.io", + "securitygroupingressrules.ec2.aws.m.upbound.io", + "subnets.ec2.aws.m.upbound.io", + "vpcs.ec2.aws.m.upbound.io", + "filesystems.efs.aws.m.upbound.io", + "mounttargets.efs.aws.m.upbound.io", + "addons.eks.aws.m.upbound.io", + "clusters.eks.aws.m.upbound.io", + "clusterauths.eks.aws.m.upbound.io", + "nodegroups.eks.aws.m.upbound.io", + "podidentityassociations.eks.aws.m.upbound.io", + "policies.iam.aws.m.upbound.io", + "roles.iam.aws.m.upbound.io", + "rolepolicyattachments.iam.aws.m.upbound.io", + ], + ready=fnv1.READY_TRUE, + ), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "model-replicas": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelReplica", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-routes": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelRoute", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-caches": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="ModelCache"), + "gateways": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceGateway"), + "class-gpu-l4-eks": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4-eks" + ), + "reservation-l4-block": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="CapacityReservation", match_name="l4-block" + ), + }, + ), + results=[ + fnv1.Result( + severity=fnv1.SEVERITY_FATAL, + message="CapacityReservation l4-block is in region us-east-2, but the cluster is in us-west-2", + ) + ], + conditions=[ + fnv1.Condition( + type="ClusterReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="InvalidNodePool", + message="CapacityReservation l4-block is in region us-east-2, but the cluster is in us-west-2", + ), + ], + ), + ), + # compose-eks-cluster turns the pool's fabric into EFA launch-template + # interfaces. ComposeCase( name="EKSFabricEFA", reason="A node pool that opts into the EFA fabric passes fabric.type through to the EKSCluster pool.", @@ -2938,6 +3807,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4-eks": _input_usage(kind="InferenceClass", name="gpu-l4-eks"), "activation": _activation_policy( activate=[ "eips.ec2.aws.m.upbound.io", @@ -2966,7 +3836,7 @@ def _to_dict(msg: message.Message) -> dict: ready=fnv1.READY_TRUE, ), "eks-cluster": _eks_cluster( - zones=["us-west-2a"], capacity_block=None, fabric="EFA", ready=fnv1.READY_UNSPECIFIED + zones=["us-west-2a"], capacity_reservation=None, fabric="EFA", ready=fnv1.READY_UNSPECIFIED ), }, ), @@ -3076,6 +3946,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-h100-nebius": _input_usage(kind="InferenceClass", name="gpu-h100-nebius"), "activation": _activation_policy( activate=[ "filesystems.compute.nebius.m.upbound.io", @@ -3264,6 +4135,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-h100-nebius": _input_usage(kind="InferenceClass", name="gpu-h100-nebius"), "activation": _activation_policy( activate=[ "filesystems.compute.nebius.m.upbound.io", @@ -3436,6 +4308,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-h100-aks": _input_usage(kind="InferenceClass", name="gpu-h100-aks"), "activation": _activation_policy( activate=[ "kubernetesclusters.containerservice.azure.m.upbound.io", @@ -3614,6 +4487,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-h100-aks": _input_usage(kind="InferenceClass", name="gpu-h100-aks"), "activation": _activation_policy( activate=[ "kubernetesclusters.containerservice.azure.m.upbound.io", @@ -3772,6 +4646,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "activation": _activation_policy( activate=[ "projectiammembers.cloudplatform.gcp.m.upbound.io", @@ -3916,6 +4791,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "activation": _activation_policy( activate=[ "projectiammembers.cloudplatform.gcp.m.upbound.io", @@ -3929,7 +4805,7 @@ def _to_dict(msg: message.Message) -> dict: ], ready=fnv1.READY_TRUE, ), - "gke-cluster": _gke_cluster(credentials=None, ready=fnv1.READY_TRUE), + "gke-cluster": _gke_cluster(credentials=None, reservation=None, ready=fnv1.READY_TRUE), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="test-cluster-kubeconfig-abcde", identity={ @@ -4049,6 +4925,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l40s-vultr": _input_usage(kind="InferenceClass", name="gpu-l40s-vultr"), "activation": _activation_policy( activate=["kubernetes.vke.vultr.m.upbound.io", "kubernetesnodepools.vke.vultr.m.upbound.io"], ready=fnv1.READY_TRUE, @@ -4148,6 +5025,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l40s-vultr": _input_usage(kind="InferenceClass", name="gpu-l40s-vultr"), "activation": _activation_policy( activate=["kubernetes.vke.vultr.m.upbound.io", "kubernetesnodepools.vke.vultr.m.upbound.io"], ready=fnv1.READY_TRUE, @@ -4283,6 +5161,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l40s-vultr": _input_usage(kind="InferenceClass", name="gpu-l40s-vultr"), "activation": _activation_policy( activate=["kubernetes.vke.vultr.m.upbound.io", "kubernetesnodepools.vke.vultr.m.upbound.io"], ready=fnv1.READY_TRUE, @@ -4425,6 +5304,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-cpu-vultr": _input_usage(kind="InferenceClass", name="cpu-vultr"), "activation": _activation_policy( activate=["kubernetes.vke.vultr.m.upbound.io", "kubernetesnodepools.vke.vultr.m.upbound.io"], ready=fnv1.READY_TRUE, @@ -4538,6 +5418,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l40s-civo": _input_usage(kind="InferenceClass", name="gpu-l40s-civo"), "activation": _activation_policy( activate=[ "clusters.kubernetes.civo.m.upbound.io", @@ -4660,6 +5541,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l40s-civo": _input_usage(kind="InferenceClass", name="gpu-l40s-civo"), "activation": _activation_policy( activate=[ "clusters.kubernetes.civo.m.upbound.io", @@ -4816,6 +5698,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l40s-civo": _input_usage(kind="InferenceClass", name="gpu-l40s-civo"), "activation": _activation_policy( activate=[ "clusters.kubernetes.civo.m.upbound.io", @@ -5005,6 +5888,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-h100-civo": _input_usage(kind="InferenceClass", name="gpu-h100-civo"), "activation": _activation_policy( activate=[ "clusters.kubernetes.civo.m.upbound.io", @@ -5202,6 +6086,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "usage-replicas": _guard_cluster_usage( reason="ModelReplicas, ModelRoutes and ModelCaches use this InferenceCluster" ), @@ -5326,6 +6211,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "usage-replicas": _guard_cluster_usage(reason="ModelRoutes use this InferenceCluster"), "namespace-team-c": _namespace_object(team="team-c", name="mp-team-c-d79d9"), "cluster-provider-config-kubernetes": _cluster_provider_config( @@ -5431,6 +6317,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "usage-replicas": _guard_cluster_usage(reason="ModelCaches use this InferenceCluster"), "namespace-team-d": _namespace_object(team="team-d", name="mp-team-d-c2383"), "cluster-provider-config-kubernetes": _cluster_provider_config( @@ -5537,6 +6424,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "usage-replicas": _guard_cluster_usage(reason="InferenceGateways use this InferenceCluster"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None @@ -5647,6 +6535,7 @@ def _to_dict(msg: message.Message) -> dict: gateway=None, ), resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), "cluster-provider-config-kubernetes": _cluster_provider_config( kubeconfig="my-kubeconfig", identity=None ), @@ -5686,17 +6575,12 @@ def _to_dict(msg: message.Message) -> dict: ], ), ), - # resolve_classes() returns False whenever a referenced InferenceClass isn't - # observed yet - a routine transient. The guard runs first, so a - # referencing replica still blocks deletion. This is the case that - # regresses if the guard is gated behind class resolution or cluster source. - # - # Only the guard and namespace are composed, both ready, so the function - # marks the XR not ready itself. Otherwise it would read ready while it - # waits for its classes. + # A missing InferenceClass fails the reconcile, so Crossplane applies none + # of its desired state and an existing cluster XR isn't deleted for being + # missing from it. ComposeCase( name="ClassUnresolved", - reason="With its class unresolved and a replica on the cluster, the function composes the guard and the replica's namespace before returning early.", + reason="With its class unresolved, the function fails the reconcile, reporting the missing class on ClusterReady.", req=fnv1.RunFunctionRequest( observed=fnv1.State( composite=_inference_cluster( @@ -5727,18 +6611,19 @@ def _to_dict(msg: message.Message) -> dict: ) ] ), + # Crossplane fetched the class, and it doesn't exist. + "class-gpu-l4": fnv1.Resources(), }, ), want=fnv1.RunFunctionResponse( meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), desired=fnv1.State( - composite=fnv1.Resource(ready=fnv1.READY_FALSE), resources={ "usage-replicas": _guard_cluster_usage(reason="ModelReplicas use this InferenceCluster"), "namespace-team-a": _namespace_object(team="team-a", name="mp-team-a-bd964"), }, ), - results=[fnv1.Result(severity=fnv1.SEVERITY_NORMAL, message="Waiting for InferenceClasses: gpu-l4")], + results=[fnv1.Result(severity=fnv1.SEVERITY_FATAL, message="Waiting for InferenceClasses: gpu-l4")], context=structpb.Struct(), requirements=fnv1.Requirements( resources={ @@ -5769,6 +6654,81 @@ def _to_dict(msg: message.Message) -> dict: ], ), ), + # The hostname gate, which is what keeps a cluster off the schedule until + # traffic to it is mutually authenticated in both directions. These cases + # observe an existing cluster's ServingStack ready, with whatever it and the + # fleet's InferenceGateways have published so far. The gateway name is + # Modelplane's own, so nothing configures it. An InferenceGateway running on + # test-cluster also composes the deletion guard. + # + # This cluster's CA lets an InferenceGateway tell it reached the right + # cluster, and an InferenceGateway CA makes the cluster gateway demand a + # client certificate. + ComposeCase( + name="ClassNotYetFetched", + reason="Before Crossplane has fetched its class, the function asks for it and waits, without failing the reconcile.", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_inference_cluster( + cluster=v1alpha1.Cluster( + source="Existing", + existing=v1alpha1.Existing(secretRef=v1alpha1.SecretRef(name="my-kubeconfig")), + ), + node_pools=[ + v1alpha1.NodePool(name="l4-pool", className="gpu-l4", nodeCount=2, maxNodeCount=4), + ], + ), + ), + required_resources={ + "model-replicas": fnv1.Resources( + items=[ + fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "modelplane.ai/v1alpha1", + "kind": "ModelReplica", + "metadata": { + "name": "deploy-test-cluster-0", + "namespace": "team-a", + "labels": {"modelplane.ai/cluster": "test-cluster"}, + }, + } + ) + ) + ] + ), + }, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "usage-replicas": _guard_cluster_usage(reason="ModelReplicas use this InferenceCluster"), + "namespace-team-a": _namespace_object(team="team-a", name="mp-team-a-bd964"), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "model-replicas": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelReplica", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-routes": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="ModelRoute", + match_labels=fnv1.MatchLabels(labels={"modelplane.ai/cluster": "test-cluster"}), + ), + "model-caches": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="ModelCache"), + "gateways": fnv1.ResourceSelector(api_version="modelplane.ai/v1alpha1", kind="InferenceGateway"), + "class-gpu-l4": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", kind="InferenceClass", match_name="gpu-l4" + ), + }, + ), + ), + ), # The devices gate fires on a cluster that is already running - a shared # class edited to claim a driver the cloud can't serve. Everything after # the gate composes the cluster, so a bare return would empty the desired @@ -5848,7 +6808,11 @@ def _to_dict(msg: message.Message) -> dict: ), want=fnv1.RunFunctionResponse( meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), - desired=fnv1.State(), + desired=fnv1.State( + resources={ + "usage-inferenceclass-gpu-l4": _input_usage(kind="InferenceClass", name="gpu-l4"), + }, + ), results=[ fnv1.Result( severity=fnv1.SEVERITY_FATAL, diff --git a/schemas/.lock.json b/schemas/.lock.json index 1f7f6524b..88400268f 100644 --- a/schemas/.lock.json +++ b/schemas/.lock.json @@ -1,6 +1,6 @@ { "packages": { - "fs://apis": "a214ca0e4ca2772e256e3da42cb23ba4c79d25a648e3eee69acc5c87b9b3dab2", + "fs://apis": "2b830096fb311172c8a7a9e71b26455c7babbb8ced021817c1d7515b380c60e9", "git://https://github.com/crossplane/crossplane/cluster/crds": "90d8b72ad8b829f0bcd7d7d5a98eaa0d579f244a", "git://https://github.com/modelplaneai/cluster-agent/cluster/charts/cluster-agent/files/crds": "ca7631d72ef8550a2d2f415f33db1ec52b20c65c", "xpkg://xpkg.upbound.io/upbound/provider-aws-ec2:v2.9.4": "sha256:7a0febdfb8da2d59a9fdcdbc2d476811f00566fcb941880313fc2034b0ca6177", diff --git a/schemas/python/models/ai/modelplane/capacityreservation/__init__.py b/schemas/python/models/ai/modelplane/capacityreservation/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/schemas/python/models/ai/modelplane/capacityreservation/v1alpha1.py b/schemas/python/models/ai/modelplane/capacityreservation/v1alpha1.py new file mode 100644 index 000000000..46c3c10dd --- /dev/null +++ b/schemas/python/models/ai/modelplane/capacityreservation/v1alpha1.py @@ -0,0 +1,145 @@ +# generated by datamodel-codegen: +# filename: workdir/modelplane_ai_v1alpha1_capacityreservation.yaml + +from __future__ import annotations + +from typing import Literal + +from pydantic import AwareDatetime, BaseModel, constr + +from ....io.k8s.apimachinery.pkg.apis.meta import v1 + + +class CompositionRef(BaseModel): + name: str + + +class CompositionRevisionRef(BaseModel): + name: str + + +class CompositionRevisionSelector(BaseModel): + matchLabels: dict[str, str] + + +class CompositionSelector(BaseModel): + matchLabels: dict[str, str] + + +class ResourceRef(BaseModel): + apiVersion: str + kind: str + name: str | None = None + namespace: str | None = None + + +class Crossplane(BaseModel): + compositionRef: CompositionRef | None = None + compositionRevisionRef: CompositionRevisionRef | None = None + compositionRevisionSelector: CompositionRevisionSelector | None = None + compositionSelector: CompositionSelector | None = None + compositionUpdatePolicy: Literal['Automatic', 'Manual'] | None = None + resourceRefs: list[ResourceRef] | None = None + + +class Eks(BaseModel): + id: constr(pattern=r'^cr-[0-9a-f]+$', min_length=4, max_length=64) + """ + ID of the Capacity Reservation (e.g. cr-0123456789abcdef0). Purchasing a Capacity Block yields this ID. + """ + region: constr(min_length=1, max_length=32) + """ + AWS region the reservation is in (e.g. us-east-2). + """ + type: Literal['CapacityBlock', 'OnDemand'] + """ + The kind of Capacity Reservation. CapacityBlock is a Capacity Block for ML. OnDemand is an On-Demand Capacity Reservation. + """ + + +class Gke(BaseModel): + name: constr(pattern=r'^[a-z]([-a-z0-9]{0,61}[a-z0-9])?$', max_length=63) + """ + Name of the reservation. + """ + project: constr(pattern=r'^[a-z][-a-z0-9]{4,28}[a-z0-9]$', max_length=30) | None = ( + None + ) + """ + Project that owns the reservation, for a reservation shared from another project. The reservation must be shared with the cluster's project, and the cluster project's GKE service agent must be allowed to list reservations in this project. Omit for a reservation in the cluster's own project. + """ + zone: constr(min_length=1, max_length=63) + """ + Zone the reservation is in (e.g. us-central1-a). A node pool backed by the reservation must be restricted to exactly this zone. + """ + + +class Spec(BaseModel): + crossplane: Crossplane | None = None + """ + Configures how Crossplane will reconcile this composite resource + """ + eks: Eks | None = None + """ + An EC2 Capacity Reservation. Required when provider is EKS. Immutable, because node pools that reference this CapacityReservation can't move to another reservation in place. + """ + gke: Gke | None = None + """ + A Compute Engine reservation, consumed as a specific reservation. Any reservation type works, including future reservations in calendar mode once their reservation period starts. Required when provider is GKE. Immutable, because node pools that reference this CapacityReservation can't move to another reservation in place. + """ + provider: Literal['GKE', 'EKS'] + """ + The cloud the reservation belongs to. A node pool can only reference a reservation whose provider matches its cluster's source. + """ + + +class Condition(BaseModel): + lastTransitionTime: AwareDatetime + message: str | None = None + observedGeneration: int | None = None + reason: str + status: str + type: str + + +class Status(BaseModel): + conditions: list[Condition] | None = None + """ + Conditions of the resource. + """ + + +class CapacityReservation(BaseModel): + apiVersion: Literal['modelplane.ai/v1alpha1'] | None = 'modelplane.ai/v1alpha1' + """ + APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + """ + kind: Literal['CapacityReservation'] | None = 'CapacityReservation' + """ + Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + """ + metadata: v1.ObjectMeta | None = None + """ + Standard object's metadata. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#metadata + """ + spec: Spec + status: Status | None = None + + +class CapacityReservationList(BaseModel): + apiVersion: str | None = None + """ + APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + """ + items: list[CapacityReservation] + """ + List of capacityreservations. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md + """ + kind: str | None = None + """ + Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + """ + metadata: v1.ListMeta | None = None + """ + Standard list metadata. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + """ \ No newline at end of file diff --git a/schemas/python/models/ai/modelplane/inferencecluster/v1alpha1.py b/schemas/python/models/ai/modelplane/inferencecluster/v1alpha1.py index d6a4a53a6..cfd9cac37 100644 --- a/schemas/python/models/ai/modelplane/inferencecluster/v1alpha1.py +++ b/schemas/python/models/ai/modelplane/inferencecluster/v1alpha1.py @@ -210,15 +210,6 @@ class Crossplane(BaseModel): resourceRefs: list[ResourceRef] | None = None -class CapacityBlock(BaseModel): - capacityReservationId: constr( - pattern=r'^cr-[0-9a-f]+$', min_length=4, max_length=64 - ) - """ - The ID of the Capacity Reservation backing the Capacity Block (e.g. cr-0123456789abcdef0). Purchasing a Capacity Block yields this ID. - """ - - class Infiniband(BaseModel): fabric: constr(min_length=1, max_length=63) """ @@ -237,11 +228,14 @@ class Fabric(BaseModel): """ -class NodePool(BaseModel): - capacityBlock: CapacityBlock | None = None +class ReservationRef(BaseModel): + name: constr(min_length=1, max_length=253) """ - Capacity Block reservation backing this node pool. EKS only. Large GPU instances (e.g. p5en.48xlarge) are rarely available on demand; AWS allocates them via Capacity Blocks for ML. Set this to back the pool with a Capacity Block you have purchased. The pool's zones must match the reservation's Availability Zone, and nodeCount must not exceed the reserved instance count. Omit for on-demand pools. + Name of the CapacityReservation. """ + + +class NodePool(BaseModel): className: constr(min_length=1, max_length=253) """ Name of the InferenceClass describing this pool's hardware. @@ -260,6 +254,10 @@ class NodePool(BaseModel): Name of the pool. Its nodes carry it as the value of the modelplane.ai/pool label, and the cluster agent observes an NVIDIA GPU pool through an object of this name. """ nodeCount: conint(ge=0) | None = 1 + reservationRef: ReservationRef | None = None + """ + CapacityReservation backing this node pool. GKE and EKS only. The pool only uses the referenced reservation, and never falls back to on-demand capacity, so nodeCount must not exceed the reserved instance count. The reservation's provider must match the cluster's source. On GKE the pool's zones must be exactly the reservation's zone; on EKS they must be the reservation's Availability Zone. Immutable, because neither cloud can move an existing node pool onto or off a reservation. Omit for on-demand pools. + """ zones: list[str] | None = None """ Zones to restrict this node pool to. Required for provisioned pools because not all zones in a region have every GPU type. diff --git a/schemas/python/models/ai/modelplane/infrastructure/ekscluster/v1alpha1.py b/schemas/python/models/ai/modelplane/infrastructure/ekscluster/v1alpha1.py index 192f72dea..37ce4eefa 100644 --- a/schemas/python/models/ai/modelplane/infrastructure/ekscluster/v1alpha1.py +++ b/schemas/python/models/ai/modelplane/infrastructure/ekscluster/v1alpha1.py @@ -68,12 +68,14 @@ class Networking(BaseModel): """ -class CapacityBlock(BaseModel): - capacityReservationId: constr( - pattern=r'^cr-[0-9a-f]+$', min_length=4, max_length=64 - ) +class CapacityReservation(BaseModel): + id: constr(pattern=r'^cr-[0-9a-f]+$', min_length=4, max_length=64) + """ + The ID of the Capacity Reservation (e.g. cr-0123456789abcdef0). + """ + type: Literal['CapacityBlock', 'OnDemand'] """ - The ID of the Capacity Reservation backing the Capacity Block (e.g. cr-0123456789abcdef0). Purchasing a Capacity Block yields this ID. + The kind of Capacity Reservation. CapacityBlock is a Capacity Block for ML. OnDemand is an On-Demand Capacity Reservation. """ @@ -89,9 +91,9 @@ class Zone(RootModel[constr(min_length=1, max_length=63)]): class NodePool(BaseModel): - capacityBlock: CapacityBlock | None = None + capacityReservation: CapacityReservation | None = None """ - Capacity Block reservation backing this node group. Large GPU instances (e.g. p5en.48xlarge) are rarely available on demand; AWS allocates them via Capacity Blocks for ML. Set this to back the node group with a Capacity Block you have purchased. When set, Modelplane composes a launch template targeting the reservation and creates the node group with CAPACITY_BLOCK capacity type. The node group's zones must match the reservation's Availability Zone, and nodeCount must not exceed the reserved instance count. Omit for on-demand node groups. + Capacity Reservation backing this node group. When set, Modelplane composes a launch template that launches instances only into the reservation. A Capacity Block node group also uses the CAPACITY_BLOCK capacity type. The node group's zones must match the reservation's Availability Zone, and nodeCount must not exceed the reserved instance count. Omit for on-demand node groups. """ diskSizeGb: conint(ge=10, le=65536) | None = 100 """ diff --git a/schemas/python/models/ai/modelplane/infrastructure/gkecluster/v1alpha1.py b/schemas/python/models/ai/modelplane/infrastructure/gkecluster/v1alpha1.py index 7a408a0d7..18144b5a5 100644 --- a/schemas/python/models/ai/modelplane/infrastructure/gkecluster/v1alpha1.py +++ b/schemas/python/models/ai/modelplane/infrastructure/gkecluster/v1alpha1.py @@ -74,6 +74,17 @@ class Gpu(BaseModel): """ +class Reservation(BaseModel): + name: constr(min_length=1, max_length=63) + """ + Name of the reservation. + """ + project: constr(min_length=1, max_length=30) | None = None + """ + Project that owns the reservation, for a reservation shared from another project. Omit for a reservation in the cluster's project. + """ + + class Zone(RootModel[constr(min_length=1, max_length=63)]): root: constr(min_length=1, max_length=63) @@ -107,6 +118,10 @@ class NodePool(BaseModel): """ Initial number of nodes. """ + reservation: Reservation | None = None + """ + Compute Engine reservation backing this node pool. The pool consumes only this reservation, as a specific reservation, so its size is limited to the reservation's capacity. The pool's zones must be the reservation's zone. Omit for on-demand pools. + """ role: Literal['System', 'GPU', 'CPU'] """ Determines what workloads this pool runs. System pools host controllers, gateways, and infrastructure. GPU pools host inference workloads and are tainted to exclude non-GPU pods. CPU pools host CPU inference workloads: a standard node image, no GPU, and the modelplane.ai/cpu label and taint. diff --git a/uv.lock b/uv.lock index ca57061c9..b29e54b2a 100644 --- a/uv.lock +++ b/uv.lock @@ -9,6 +9,7 @@ resolution-markers = [ [manifest] members = [ "compose-aks-cluster", + "compose-capacity-reservation", "compose-civo-cluster", "compose-eks-cluster", "compose-gke-cluster", @@ -292,6 +293,25 @@ requires-dist = [ { name = "grpcio", specifier = ">=1.73.1" }, ] +[[package]] +name = "compose-capacity-reservation" +version = "0.0.0" +source = { editable = "functions/compose-capacity-reservation" } +dependencies = [ + { name = "click" }, + { name = "crossplane-function-sdk-python" }, + { name = "crossplane-models" }, + { name = "grpcio" }, +] + +[package.metadata] +requires-dist = [ + { name = "click", specifier = ">=8.1.0" }, + { name = "crossplane-function-sdk-python", specifier = ">=0.14.0" }, + { name = "crossplane-models", editable = "schemas/python" }, + { name = "grpcio", specifier = ">=1.73.1" }, +] + [[package]] name = "compose-civo-cluster" version = "0.0.0"