diff --git a/apis/metricmappings/composition.yaml b/apis/metricmappings/composition.yaml new file mode 100644 index 000000000..8359dc414 --- /dev/null +++ b/apis/metricmappings/composition.yaml @@ -0,0 +1,13 @@ +apiVersion: apiextensions.crossplane.io/v1 +kind: Composition +metadata: + name: metricmappings.modelplane.ai +spec: + compositeTypeRef: + apiVersion: modelplane.ai/v1alpha1 + kind: MetricMapping + mode: Pipeline + pipeline: + - functionRef: + name: modelplane-modelplanecompose-metric-mapping + step: compose-metric-mapping diff --git a/apis/metricmappings/definition.yaml b/apis/metricmappings/definition.yaml new file mode 100644 index 000000000..77965ebd7 --- /dev/null +++ b/apis/metricmappings/definition.yaml @@ -0,0 +1,67 @@ +apiVersion: apiextensions.crossplane.io/v2 +kind: CompositeResourceDefinition +metadata: + name: metricmappings.modelplane.ai +spec: + group: modelplane.ai + names: + categories: [crossplane, modelplane] + kind: MetricMapping + plural: metricmappings + shortNames: [mm] + scope: Cluster + versions: + - name: v1alpha1 + served: true + referenceable: true + schema: + openAPIV3Schema: + type: object + required: [spec] + properties: + spec: + type: object + description: >- + How to normalize one component's Prometheus metrics onto the + modelplane_* surface. The metrics collector reads every + MetricMapping and renders the matching rename and label rules + into its config. Modelplane ships a MetricMapping per common + engine; a platform team applies one more for a new or forked + engine. + properties: + selector: + type: object + description: >- + Selects the pods this mapping applies to, by label. An + engine mapping matches the engine-type label + (modelplane.ai/engine); a scheduler mapping matches the + scheduler's pods. + properties: + matchLabels: + type: object + additionalProperties: + type: string + rename: + type: object + description: >- + Source metric name to its modelplane_* name, e.g. + vllm:time_to_first_token_seconds becomes + modelplane_time_to_first_token. + additionalProperties: + type: string + labels: + type: object + description: Label rewrites applied to the matched series. + properties: + add: + type: object + description: Labels to add to every matched series. + additionalProperties: + type: string + status: + type: object + properties: + conditions: + type: array + items: + type: object diff --git a/apis/modeldeployments/definition.yaml b/apis/modeldeployments/definition.yaml index 36661c051..cd75478f2 100644 --- a/apis/modeldeployments/definition.yaml +++ b/apis/modeldeployments/definition.yaml @@ -242,6 +242,22 @@ spec: default: 1 minimum: 1 maximum: 64 + type: + type: string + description: >- + Which engine this runs, e.g. vllm or sglang. + Stamped onto the serving pods as + modelplane.ai/engine, where a MetricMapping + selects on it to normalize that engine's metrics + onto the modelplane_* surface. Free-form rather + than an enum: a platform team adds a + MetricMapping for a forked or new engine without + a Modelplane release, and an engine with no + mapping is still scraped under its native names. + Omit it and the engine's metrics are collected + unnormalized. + minLength: 1 + maxLength: 63 phase: type: string enum: [Prefill, Decode] diff --git a/apis/modelreplicas/definition.yaml b/apis/modelreplicas/definition.yaml index 20931ebd4..adb4c7da7 100644 --- a/apis/modelreplicas/definition.yaml +++ b/apis/modelreplicas/definition.yaml @@ -137,6 +137,20 @@ spec: phase: type: string enum: [Prefill, Decode] + type: + type: string + description: >- + Which engine this runs, e.g. vllm or sglang. Stamped + onto the serving pods as modelplane.ai/engine, where a + MetricMapping selects on it to normalize that engine's + metrics onto the modelplane_* surface. Free-form rather + than an enum: a platform team adds a MetricMapping for a + forked or new engine without a Modelplane release, and an + engine with no mapping is still scraped under its native + names. Omit it and the engine's metrics are collected + unnormalized. + minLength: 1 + maxLength: 63 members: type: array minItems: 1 diff --git a/apis/servingstacks/definition.yaml b/apis/servingstacks/definition.yaml index 89841ef46..801e1cc9b 100644 --- a/apis/servingstacks/definition.yaml +++ b/apis/servingstacks/definition.yaml @@ -137,6 +137,15 @@ spec: description: kube-prometheus-stack chart version. minLength: 1 maxLength: 32 + otelCollector: + type: string + default: "0.116.0" + description: >- + OpenTelemetry Collector chart version. The collector + scrapes each engine's metrics and renames them onto the + modelplane_* surface from the cluster's MetricMappings. + minLength: 1 + maxLength: 32 leaderWorkerSet: type: string default: "v0.8.0" diff --git a/crossplane-project.yaml b/crossplane-project.yaml index 494c3bdb4..b49348460 100644 --- a/crossplane-project.yaml +++ b/crossplane-project.yaml @@ -67,6 +67,10 @@ spec: tarball: name: compose-model-service pathPrefix: _output/functions/compose-model-service + - source: Tarball + tarball: + name: compose-metric-mapping + pathPrefix: _output/functions/compose-metric-mapping - source: Tarball tarball: name: compose-usages diff --git a/docs/content/examples/collecting-engine-metrics.md b/docs/content/examples/collecting-engine-metrics.md index 2936132dc..8cb1cfeec 100644 --- a/docs/content/examples/collecting-engine-metrics.md +++ b/docs/content/examples/collecting-engine-metrics.md @@ -1,20 +1,20 @@ --- title: Collecting engine metrics weight: 50 -description: Scrape a vLLM engine's Prometheus metrics through the in-cluster Prometheus. +description: Read a vLLM engine's metrics, collected and normalized onto the modelplane_* surface. --- -Scraping an inference engine's Prometheus metrics, shown on the smallest serving -shape: a 0.5B Qwen chat model on one NVIDIA L4. vLLM publishes metrics at -`/metrics` on its serving port with no extra flag, and Modelplane runs a -Prometheus on every workload cluster with `PodMonitor` discovery open across -namespaces, so scraping the engine is a `PodMonitor` plus a `port-forward`. The -model is only the subject; the same wiring fits any engine, with the SGLang, -leader/worker, and prefill/decode differences noted at the end. +Reading an inference engine's metrics, shown on the smallest serving shape: a +0.5B Qwen chat model on one NVIDIA L4. Modelplane collects from every engine it +runs, so there is nothing to wire up: a collector on each workload cluster +discovers serving pods by label, scrapes the engine port by name, and renames the +engine's metrics onto a common `modelplane_*` surface. The model is only the +subject; the same applies to any engine, with the SGLang, leader/worker, and +prefill/decode differences noted at the end. This was run end to end on GKE. The `InferenceClass` and `ModelDeployment` are the -exact manifests from that run, and the `PodMonitor` below scraped this deployment. -Apply the platform side first, then the ML side. +exact manifests from that run, and the metric names below are the ones that run +produced. Apply the platform side first, then the ML side. ## Platform @@ -28,38 +28,83 @@ Apply the platform side first, then the ML side. {{< manifests "examples/collecting-engine-metrics/model-service.yaml" >}} -## Scraping the metrics +## Reading the metrics -The `PodMonitor` selects engine pods by the `modelplane.ai/serving` label -Modelplane stamps on them, and the `monitoring` namespace Prometheus discovers any -`PodMonitor`, so this is the whole config. The engine container port is unnamed, -so reference it by number with `targetPort`: +Nothing needs applying for collection. Every serving pod carries +`modelplane.ai/serving`, and its engine container's port is named `http`, which is +what the collector's scrape config matches — so an engine is collected from as +soon as it serves. -{{< manifests "examples/collecting-engine-metrics/podmonitor.yaml" >}} - -The engine pods and the `PodMonitor` CRD live on the workload cluster, not the -control plane, so apply it there. Then read the metrics from the in-cluster -Prometheus over a `port-forward`: +The collector re-exposes what it collected on the workload cluster, so read it +over a `port-forward`: ```bash -kubectl -n monitoring port-forward svc/prometheus-prometheus 9090:9090 # workload cluster -# open http://localhost:9090, Status > Targets to confirm the scrape, then query -# e.g. vllm:num_requests_running or vllm:gpu_cache_usage_perc +kubectl -n monitoring port-forward svc/otel-collector 8889:8889 # workload cluster +curl -s localhost:8889/metrics | grep '^modelplane_' +``` + +For the deployment above that returns the normalized names, each labelled with the +engine that produced it: + ``` +modelplane_requests_running{engine="vllm",model_name="qwen2.5-0.5b",...} +modelplane_requests_waiting{engine="vllm",model_name="qwen2.5-0.5b",...} +modelplane_time_to_first_token_sum{engine="vllm",model_name="qwen2.5-0.5b",...} +modelplane_request_latency_sum{engine="vllm",model_name="qwen2.5-0.5b",...} +``` + +### Which names get renamed + +A `MetricMapping` decides. Modelplane ships one per common engine, matched to +serving pods by the `modelplane.ai/engine` label that `engines[].type` sets, so an +engine that declares `type: vllm` gets the vLLM mapping and no detection is +involved. A metric with no mapping entry is not dropped — it passes through under +its own name. + +Two things to know about the names that pass through. The collector's exporter +replaces `:` with `_`, so vLLM's `vllm:gpu_cache_usage_perc` is published as +`vllm_gpu_cache_usage_perc`. And an engine that declares no `type` matches no +mapping, so all of its metrics pass through rather than being renamed by a guess. + +To normalize a new or forked engine, apply another `MetricMapping` — no Modelplane +release is needed: + +```yaml +apiVersion: modelplane.ai/v1alpha1 +kind: MetricMapping +metadata: + name: my-fork + namespace: ml-team +spec: + selector: + matchLabels: + modelplane.ai/engine: my-fork + rename: + myfork:queue_depth: modelplane_requests_waiting + labels: + add: + engine: my-fork +``` + +### Upgrading from a hand-written PodMonitor + +Earlier versions of this example applied a `PodMonitor` to the workload cluster by +hand. Delete it. Collection is composed now, and leaving it in place scrapes every +engine twice. ### Other engine shapes -The `PodMonitor` above fits a single-pod vLLM engine. The selector and port shift -by shape: +The example above is a single-pod vLLM engine. Collection needs no changes for the +other shapes, but what gets collected differs: - **SGLang**: exposes `/metrics` only when the engine runs with - `--enable-metrics`; otherwise it's identical (same selector, `targetPort: 8000`). + `--enable-metrics`; otherwise it is collected from the same way. - **Leader/worker**: only the leader serves the API and carries - `modelplane.ai/serving`, so the selector above already scrapes the leader alone; - the workers expose nothing. + `modelplane.ai/serving`, so only the leader is collected from; the workers serve + nothing and expose no metrics. - **prefill/decode**: two engines, labelled `llm-d.ai/role: prefill` and - `llm-d.ai/role: decode`. The prefill engine serves on `8000`; the decode engine - sits behind the routing sidecar that takes `8000` and listens on `8001`, so - scrape decode with `targetPort: 8001`. Select each by its role label to keep - them apart. + `llm-d.ai/role: decode`. Both are collected from without special casing, because + the scrape matches the engine port by name rather than by number: the decode + engine serves on `8001` since the routing sidecar takes `8000`, and a config + matching `8000` would report the sidecar's metrics as the engine's. diff --git a/docs/manifests/examples/collecting-engine-metrics/model-deployment.yaml b/docs/manifests/examples/collecting-engine-metrics/model-deployment.yaml index 4cd06648d..0fed68f92 100644 --- a/docs/manifests/examples/collecting-engine-metrics/model-deployment.yaml +++ b/docs/manifests/examples/collecting-engine-metrics/model-deployment.yaml @@ -10,7 +10,7 @@ # --served-model-name the id clients pass as "model" in OpenAI requests. # # vLLM exposes Prometheus metrics at /metrics on its serving port (:8000) with no -# extra flag, which is what the example's PodMonitor scrapes. +# extra flag, which is what Modelplane's collector scrapes. # # No --port or --host: Modelplane's routing expects the engine on its default # :8000 with a /health probe, and passes args through verbatim. @@ -26,6 +26,10 @@ spec: spec: engines: - name: qwen2-5-0-5b + # Names the engine so its serving pods carry modelplane.ai/engine, which + # is how the vLLM MetricMapping selects them. Without it the engine is + # still collected from, under vLLM's own metric names. + type: vllm members: # A single self-contained vLLM pod. The container named "engine" is the # inference server; its image and args pass through verbatim. diff --git a/docs/manifests/examples/collecting-engine-metrics/podmonitor.yaml b/docs/manifests/examples/collecting-engine-metrics/podmonitor.yaml deleted file mode 100644 index 00db036a7..000000000 --- a/docs/manifests/examples/collecting-engine-metrics/podmonitor.yaml +++ /dev/null @@ -1,14 +0,0 @@ -apiVersion: monitoring.coreos.com/v1 -kind: PodMonitor -metadata: - name: qwen2-5-0-5b-metrics - namespace: default -spec: - selector: - matchExpressions: - - key: modelplane.ai/serving # carried by every serving pod - operator: Exists - podMetricsEndpoints: - - targetPort: 8000 - path: /metrics - interval: 30s diff --git a/e2e/manifests/25-metric-mapping.yaml b/e2e/manifests/25-metric-mapping.yaml new file mode 100644 index 000000000..101d69289 --- /dev/null +++ b/e2e/manifests/25-metric-mapping.yaml @@ -0,0 +1,20 @@ +--- +# The vLLM mapping. The mock engine publishes vLLM's own metric names on +# /metrics, and the ModelDeployment declares `type: vllm`, so this selects that +# engine's pods by label and the collector renames only their series. An engine +# with no matching mapping keeps its native names. +apiVersion: modelplane.ai/v1alpha1 +kind: MetricMapping +metadata: + name: vllm + namespace: ml-team +spec: + selector: + matchLabels: + modelplane.ai/engine: vllm + rename: + vllm:num_requests_waiting: modelplane_requests_waiting + vllm:time_to_first_token_seconds: modelplane_time_to_first_token + labels: + add: + engine: vllm diff --git a/e2e/manifests/40-model-deployment.yaml b/e2e/manifests/40-model-deployment.yaml index f0d76ef14..bd9ba9358 100644 --- a/e2e/manifests/40-model-deployment.yaml +++ b/e2e/manifests/40-model-deployment.yaml @@ -15,6 +15,9 @@ spec: spec: engines: - name: mock + # Names the engine so its serving pods carry modelplane.ai/engine, which + # is what the vllm MetricMapping selects on. + type: vllm members: - role: Standalone nodeSelector: @@ -44,9 +47,29 @@ spec: self.send_header("content-length", str(len(b))) self.end_headers() self.wfile.write(b) + def _text(self, body, c=200): + b = body.encode() + self.send_response(c) + self.send_header("content-type", "text/plain; version=0.0.4") + self.send_header("content-length", str(len(b))) + self.end_headers() + self.wfile.write(b) def do_GET(self): if self.path == "/health": self._s({"status": "ok"}) + elif self.path == "/metrics": + # Prometheus exposition under vLLM's own names, so the + # MetricMapping has something real to rename. A real + # engine publishes these on the serving port too. + self._text( + "# TYPE vllm:num_requests_waiting gauge\n" + "vllm:num_requests_waiting 3\n" + "# TYPE vllm:time_to_first_token_seconds histogram\n" + 'vllm:time_to_first_token_seconds_bucket{le="0.1"} 1\n' + 'vllm:time_to_first_token_seconds_bucket{le="+Inf"} 2\n' + "vllm:time_to_first_token_seconds_sum 0.25\n" + "vllm:time_to_first_token_seconds_count 2\n" + ) elif self.path.startswith("/v1/models"): self._s({"object": "list", "data": [{"id": "mock", "object": "model"}]}) else: diff --git a/e2e/run.sh b/e2e/run.sh index 82eddaef1..c9a8f2d87 100644 --- a/e2e/run.sh +++ b/e2e/run.sh @@ -248,3 +248,39 @@ kubectl --context "$cpctx" -n "$ns" delete pod -l app.kubernetes.io/name=e2e-ver exit 1 } log "End to end OK: $addr serves OpenAI (/v1/chat/completions) and Anthropic (/v1/messages)" + +# Metrics normalization. The requests above gave the engine something to count, +# so its /metrics now has non-zero vLLM series. Assert the collector on the +# workload cluster re-exposes them under modelplane_* names: that is the whole +# MetricMapping path -- engine label stamped, port scraped by name, rename +# applied, gated on the engine attribute. Grepping the collector's exporter +# directly keeps this a test of normalization and not of Prometheus. +log "Verifying metrics normalization" +mns=monitoring +mpod=e2e-verify-metrics +otel="http://otel-collector.$mns.svc:8889/metrics" + +# The collector has to start, discover the engine pod and complete one scrape +# interval, so this polls rather than asserting on the first try. +found="" +for attempt in $(seq 1 20); do + kubectl --context "$WLCTX" -n "$mns" delete pod "$mpod" --now >/dev/null 2>&1 || true + kubectl --context "$WLCTX" -n "$mns" run "$mpod" --restart=Never --image="$CURL_IMAGE" \ + --command -- curl -sS --max-time 15 "$otel" >/dev/null 2>&1 || true + body="" + for _ in $(seq 1 15); do + body="$(kubectl --context "$WLCTX" -n "$mns" logs "$mpod" 2>/dev/null || true)" + [ -n "$body" ] && break + sleep 2 + done + found="$(printf '%s' "$body" | grep -c '^modelplane_' || true)" + log "metrics attempt $attempt: ${found:-0} modelplane_* series" + [ "${found:-0}" -gt 0 ] && { printf '%s' "$body" | grep '^modelplane_' | head -5 | sed 's/^/ /'; break; } + sleep 15 +done +kubectl --context "$WLCTX" -n "$mns" delete pod "$mpod" --now >/dev/null 2>&1 || true +[ "${found:-0}" -gt 0 ] || { + echo "verify: the collector at $otel exposed no modelplane_* series; the MetricMapping rename did not apply" >&2 + exit 1 +} +log "Metrics OK: the collector re-exposes the engine's vLLM series as modelplane_*" diff --git a/flake.nix b/flake.nix index 01e40170e..8ea0ac75c 100644 --- a/flake.nix +++ b/flake.nix @@ -59,6 +59,7 @@ "compose-inference-class" "compose-inference-cluster" "compose-inference-gateway" + "compose-metric-mapping" "compose-nebius-cluster" "compose-serving-stack" "compose-model-cache" diff --git a/functions/compose-metric-mapping/function/__init__.py b/functions/compose-metric-mapping/function/__init__.py new file mode 100644 index 000000000..ebf4b2ad4 --- /dev/null +++ b/functions/compose-metric-mapping/function/__init__.py @@ -0,0 +1,13 @@ +# 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-metric-mapping/function/fn.py b/functions/compose-metric-mapping/function/fn.py new file mode 100644 index 000000000..47e45b85b --- /dev/null +++ b/functions/compose-metric-mapping/function/fn.py @@ -0,0 +1,50 @@ +# 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 MetricMapping. + +MetricMapping is a data resource: the metrics collector reads every +MetricMapping to normalize a component's Prometheus metrics onto the +modelplane_* surface. 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.metricmapping 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-metric-mapping/function/main.py b/functions/compose-metric-mapping/function/main.py new file mode 100644 index 000000000..2e8441dac --- /dev/null +++ b/functions/compose-metric-mapping/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-metric-mapping/pyproject.toml b/functions/compose-metric-mapping/pyproject.toml new file mode 100644 index 000000000..526492339 --- /dev/null +++ b/functions/compose-metric-mapping/pyproject.toml @@ -0,0 +1,26 @@ +[build-system] +requires = ["uv_build>=0.11.0,<0.12"] +build-backend = "uv_build" + +[project] +name = "compose-metric-mapping" +version = "0.0.0" +description = "Mark a MetricMapping 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-metric-mapping/tests/__init__.py b/functions/compose-metric-mapping/tests/__init__.py new file mode 100644 index 000000000..ebf4b2ad4 --- /dev/null +++ b/functions/compose-metric-mapping/tests/__init__.py @@ -0,0 +1,13 @@ +# 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-metric-mapping/tests/test_fn.py b/functions/compose-metric-mapping/tests/test_fn.py new file mode 100644 index 000000000..4113fcbae --- /dev/null +++ b/functions/compose-metric-mapping/tests/test_fn.py @@ -0,0 +1,100 @@ +# 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-metric-mapping function.""" + +import dataclasses +import unittest + +from crossplane.function import logging, 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 +from google.protobuf import struct_pb2 as structpb + + +@dataclasses.dataclass +class Case: + """A test case for compose-metric-mapping.""" + + name: str + req: fnv1.RunFunctionRequest + want: fnv1.RunFunctionResponse + + +def setUpModule() -> None: + logging.configure(level=logging.Level.DISABLED) + + +class TestFunctionRunner(unittest.IsolatedAsyncioTestCase): + """Tests for FunctionRunner.RunFunction.""" + + @classmethod + def setUpClass(cls) -> None: + cls.runner = fn.FunctionRunner() + + async def test_compose(self) -> None: + """The function marks the MetricMapping as ready.""" + cases = [ + Case( + name="marks XR ready with Accepted condition and empty status", + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "modelplane.ai/v1alpha1", + "kind": "MetricMapping", + "metadata": {"name": "vllm"}, + "spec": { + "selector": {"matchLabels": {"modelplane.ai/engine": "vllm"}}, + "rename": { + "vllm:time_to_first_token_seconds": "modelplane_time_to_first_token", + }, + "labels": {"add": {"engine": "vllm"}}, + }, + } + ), + ), + ), + ), + 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, + ), + ), + conditions=[ + fnv1.Condition( + type="Accepted", + status=fnv1.STATUS_CONDITION_TRUE, + reason="Available", + ), + ], + context=structpb.Struct(), + ), + ), + ] + + for case in cases: + with self.subTest(case.name): + got = await self.runner.RunFunction(case.req, None) + self.assertEqual( + json_format.MessageToDict(case.want), + json_format.MessageToDict(got), + "-want, +got", + ) diff --git a/functions/compose-model-deployment/function/fn.py b/functions/compose-model-deployment/function/fn.py index 8771626ae..5556bc946 100644 --- a/functions/compose-model-deployment/function/fn.py +++ b/functions/compose-model-deployment/function/fn.py @@ -406,7 +406,8 @@ def compose_replicas(self, matched: list[scheduling.Candidate]) -> None: def _replica_engine(self, engine: v1alpha1.Engine, placement: scheduling.EnginePlacement) -> mrv1alpha1.Engine: """Build a ModelReplica engine from a deployment engine + placement. - The engine keeps its name, copies, phase, and member templates verbatim; + The engine keeps its name, copies, phase, type, and member templates + verbatim; the scheduler supplies each member's pool (nodePoolName) and resolved claim: DRA device requests. A member that claims nothing - no nodeSelector, or only synthetic devices matched - carries only its pool @@ -448,6 +449,11 @@ def _replica_engine(self, engine: v1alpha1.Engine, placement: scheduling.EngineP # doesn't serialize a null into the composed ModelReplica. if engine.phase is not None: replica_engine.phase = engine.phase + # type reaches the replica backend, which stamps it as the engine label a + # MetricMapping selects on. Optional, and omitted when unset for the same + # reason as phase. + if engine.type is not None: + replica_engine.type = engine.type return replica_engine def compose_endpoints(self, matched: list[scheduling.Candidate]) -> None: diff --git a/functions/compose-model-replica/function/backends/base.py b/functions/compose-model-replica/function/backends/base.py index 152aa91b5..d44c0ac1e 100644 --- a/functions/compose-model-replica/function/backends/base.py +++ b/functions/compose-model-replica/function/backends/base.py @@ -131,6 +131,14 @@ def apply_cache_args(args: list[str], replica: v1alpha1.ModelReplica, engine: v1 # the ModelEndpoint URLs, so it must not diverge between backends. ENGINE_PORT = 8000 +# Name for the engine's serving port. Named so a scrape can follow the engine by +# name rather than by number, which matters under prefill/decode: there the +# pd-sidecar takes ENGINE_PORT and the engine moves to its own --port, so a +# target matching 8000 by number hits the sidecar. The sidecar's port is left +# unnamed - a pod's named ports must be unique, and the engine is the one worth +# following. +ENGINE_PORT_NAME = "http" + # Pod label carrying the serving identity (the replica name). The replica's one # shared Service selects on it, so every engine's serving pods - a Standalone pod # or an LWS gang leader - carry it. A multi-node gang's worker followers do NOT @@ -143,6 +151,19 @@ def apply_cache_args(args: list[str], replica: v1alpha1.ModelReplica, engine: v1 # Deployment selectors fighting over each other's pods. LABEL_WORKLOAD = "modelplane.ai/workload" +# Pod label naming which engine a serving pod runs, from engines[].type. A +# MetricMapping selects on it to normalize that engine's metrics, so the label is +# what makes normalization label-driven rather than a guess from the image. Only +# set when the user declared a type: an engine with none is still scraped, under +# its native metric names. +LABEL_ENGINE = "modelplane.ai/engine" + + +def engine_labels(engine: v1alpha1.Engine) -> dict[str, str]: + """The engine-type label for a serving pod, or {} when none was declared.""" + return {LABEL_ENGINE: engine.type} if engine.type else {} + + # Backend-neutral env var carrying the gang leader's address, injected into # every engine container of a multi-node engine's gang. A member's command finds its # peers through this without hard-coding the underlying orchestrator's variable. diff --git a/functions/compose-model-replica/function/backends/llmd.py b/functions/compose-model-replica/function/backends/llmd.py index 8d5d9db8b..4f65264a2 100644 --- a/functions/compose-model-replica/function/backends/llmd.py +++ b/functions/compose-model-replica/function/backends/llmd.py @@ -103,7 +103,7 @@ def container(member: v1alpha1.Member, *, serving: bool) -> dict: env.extend(e.model_dump(exclude_none=True) for e in engine_container.env) c["env"] = env if serving: - c["ports"] = [{"containerPort": base.ENGINE_PORT}] + c["ports"] = [{"name": base.ENGINE_PORT_NAME, "containerPort": base.ENGINE_PORT}] c["readinessProbe"] = { "httpGet": {"path": "/health", "port": base.ENGINE_PORT}, "initialDelaySeconds": 30, @@ -135,7 +135,13 @@ def pod_spec(member: v1alpha1.Member, c: dict) -> dict: # the replica's shared Service selects on, plus the role label, the # serving port, and the readiness probe. leader_pod = { - "metadata": {"labels": {base.LABEL_SERVING: serving_label, _LABEL_ROLE: "leader"}}, + "metadata": { + "labels": { + base.LABEL_SERVING: serving_label, + _LABEL_ROLE: "leader", + **base.engine_labels(engine), + }, + }, "spec": pod_spec(leader, container(leader, serving=True)), } # The worker followers don't serve the OpenAI API, so they carry no diff --git a/functions/compose-model-replica/function/backends/native.py b/functions/compose-model-replica/function/backends/native.py index 7a29eeeb0..c9aa18708 100644 --- a/functions/compose-model-replica/function/backends/native.py +++ b/functions/compose-model-replica/function/backends/native.py @@ -48,7 +48,11 @@ def build( # per-workload label this Deployment selects on. The latter must be # engine-unique so two Standalone engines of one replica don't share a # selector and fight over each other's pods. - pod_labels = {base.LABEL_SERVING: serving_label, base.LABEL_WORKLOAD: name} + pod_labels = { + base.LABEL_SERVING: serving_label, + base.LABEL_WORKLOAD: name, + **base.engine_labels(engine), + } selector = {base.LABEL_WORKLOAD: name} cache_volumes, cache_volume_mounts = base.cache_mounts(replica) @@ -58,7 +62,7 @@ def build( "name": "engine", "image": engine_container.image, "args": args, - "ports": [{"containerPort": base.ENGINE_PORT}], + "ports": [{"name": base.ENGINE_PORT_NAME, "containerPort": base.ENGINE_PORT}], # vLLM tensor parallelism needs a large /dev/shm. "volumeMounts": [{"name": "dshm", "mountPath": "/dev/shm"}, *cache_volume_mounts], "readinessProbe": { diff --git a/functions/compose-model-replica/function/routing.py b/functions/compose-model-replica/function/routing.py index 9e01b6cff..ab0044bcd 100644 --- a/functions/compose-model-replica/function/routing.py +++ b/functions/compose-model-replica/function/routing.py @@ -375,7 +375,7 @@ def _add_sidecar_to_decode(obj: k8sobjv1alpha1.Object) -> None: containers = tmpl["spec"]["containers"] engine = next(c for c in containers if c["name"] == "engine") port = _decode_port(engine) - engine["ports"] = [{"containerPort": port}] + engine["ports"] = [{"name": base.ENGINE_PORT_NAME, "containerPort": port}] engine["readinessProbe"] = { "httpGet": {"path": "/health", "port": port}, "initialDelaySeconds": 30, diff --git a/functions/compose-model-replica/tests/test_backends.py b/functions/compose-model-replica/tests/test_backends.py index 892b97133..364036328 100644 --- a/functions/compose-model-replica/tests/test_backends.py +++ b/functions/compose-model-replica/tests/test_backends.py @@ -55,6 +55,7 @@ def _gpu_request(count: int) -> v1alpha1.DeviceRequest: def _standalone_engine( name: str = "main", *, + engine_type: str | None = None, copies: int = 1, args: list[str] | None = None, command: list[str] | None = None, @@ -69,6 +70,7 @@ def _standalone_engine( if command is not None: container.command = command return v1alpha1.Engine( + **({"type": engine_type} if engine_type else {}), name=name, copies=copies, members=[ @@ -208,7 +210,7 @@ def _claim_template(count: int, *, replica: str = "r", engine: str = "main", rol "name": "engine", "image": "vllm/vllm-openai:latest", "args": ["--model=Qwen/Qwen3-0.6B"], - "ports": [{"containerPort": 8000}], + "ports": [{"name": "http", "containerPort": 8000}], "resources": {"claims": [{"name": "devices"}]}, "volumeMounts": [{"name": "dshm", "mountPath": "/dev/shm"}], "readinessProbe": { @@ -297,7 +299,7 @@ def _engine( c["args"] = args c["env"] = env if env is not None else [_LEADER_ENV] if serving: - c["ports"] = [{"containerPort": 8000}] + c["ports"] = [{"name": "http", "containerPort": 8000}] c["readinessProbe"] = { "httpGet": {"path": "/health", "port": 8000}, "initialDelaySeconds": 30, @@ -800,6 +802,9 @@ def test_decode_port_follows_user_arg(self) -> None: self.assertEqual(engine["ports"][0]["containerPort"], 9000) self.assertIn("--vllm-port=9000", sidecar["args"]) self.assertEqual(sidecar["ports"][0]["containerPort"], 8000) + # Left unnamed: a pod's named ports must be unique, and the engine is the + # one worth following. + self.assertNotIn("name", sidecar["ports"][0]) def test_engines_role_labeled(self) -> None: out = self._apply() @@ -815,9 +820,15 @@ def test_decode_gets_sidecar_and_moves_engine_port(self) -> None: self.assertEqual(names, ["engine", "pd-sidecar"]) engine = next(c for c in containers if c["name"] == "engine") self.assertEqual(engine["ports"][0]["containerPort"], 8001) + # Named, so a scrape targeting "http" follows the engine to 8001 rather + # than matching 8000 by number and hitting the sidecar. + self.assertEqual(engine["ports"][0]["name"], "http") self.assertEqual(engine["readinessProbe"]["timeoutSeconds"], 5) sidecar = next(c for c in containers if c["name"] == "pd-sidecar") self.assertEqual(sidecar["ports"][0]["containerPort"], 8000) + # Left unnamed: a pod's named ports must be unique, and the engine is the + # one worth following. + self.assertNotIn("name", sidecar["ports"][0]) self.assertEqual(sidecar["readinessProbe"]["timeoutSeconds"], 5) self.assertIn("--secure-proxy=false", sidecar["args"]) @@ -959,3 +970,37 @@ def test_rendered_config_uses_block_size(self) -> None: if __name__ == "__main__": unittest.main() + + +class TestEngineLabel(unittest.TestCase): + """engines[].type reaches the serving pods as modelplane.ai/engine. + + It is what makes normalization label-driven: a MetricMapping selects on this + label, so a pod without it is scraped under the engine's native metric names + instead of being renamed by a guess. + """ + + def test_standalone_pod_carries_the_engine_label(self) -> None: + engine = _standalone_engine(engine_type="vllm") + replica = _replica(engines=[engine]) + out = native.NativeBackend().build(replica, engine, _PC, base.serving_label(replica)) + labels = out["model-serving-main"].spec.forProvider.manifest["spec"]["template"]["metadata"]["labels"] + self.assertEqual(labels[base.LABEL_ENGINE], "vllm") + + def test_no_type_means_no_engine_label(self) -> None: + engine = _standalone_engine() + replica = _replica(engines=[engine]) + out = native.NativeBackend().build(replica, engine, _PC, base.serving_label(replica)) + labels = out["model-serving-main"].spec.forProvider.manifest["spec"]["template"]["metadata"]["labels"] + self.assertNotIn(base.LABEL_ENGINE, labels) + + def test_gang_labels_the_leader_and_not_the_workers(self) -> None: + # The workers serve nothing, so they carry no serving label and no engine + # label either - there are no metrics on them to attribute. + engine = _gang_engine() + engine.type = "sglang" + replica = _replica(engines=[engine]) + out = llmd.LLMDBackend().build(replica, engine, _PC, base.serving_label(replica)) + tmpl = out["model-serving-main"].spec.forProvider.manifest["spec"]["leaderWorkerTemplate"] + self.assertEqual(tmpl["leaderTemplate"]["metadata"]["labels"][base.LABEL_ENGINE], "sglang") + self.assertNotIn(base.LABEL_ENGINE, tmpl["workerTemplate"].get("metadata", {}).get("labels", {})) diff --git a/functions/compose-model-replica/tests/test_fn.py b/functions/compose-model-replica/tests/test_fn.py index 32801c36c..f04ccffe4 100644 --- a/functions/compose-model-replica/tests/test_fn.py +++ b/functions/compose-model-replica/tests/test_fn.py @@ -186,7 +186,7 @@ async def test_compose(self) -> None: "name": "engine", "image": "vllm/vllm-openai:latest", "args": ["--model=Qwen/Qwen3-0.6B"], - "ports": [{"containerPort": 8000}], + "ports": [{"name": "http", "containerPort": 8000}], "resources": {"claims": [{"name": "devices"}]}, "volumeMounts": [ {"name": "dshm", "mountPath": "/dev/shm"}, diff --git a/functions/compose-serving-stack/function/fn.py b/functions/compose-serving-stack/function/fn.py index d153d52b1..87a2bf1ee 100644 --- a/functions/compose-serving-stack/function/fn.py +++ b/functions/compose-serving-stack/function/fn.py @@ -31,10 +31,11 @@ import grpc import yaml -from crossplane.function import logging, resource, response +from crossplane.function import logging, request, 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.infrastructure.servingstack import v1alpha1 +from models.ai.modelplane.metricmapping import v1alpha1 as mmv1alpha1 from models.io.crossplane.m.helm.providerconfig import v1beta1 as helmpcv1beta1 from models.io.crossplane.m.helm.release import v1beta1 as helmv1beta1 from models.io.crossplane.m.kubernetes.object import v1alpha1 as k8sobjv1alpha1 @@ -81,6 +82,35 @@ _PROMETHEUS_CHART = "kube-prometheus-stack" _PROMETHEUS_REPO = "https://prometheus-community.github.io/helm-charts" +# OpenTelemetry collector. It lands in the Prometheus namespace so the existing +# Prometheus discovers it with no extra wiring: the collector scrapes each engine +# under its native names, renames them onto modelplane_*, and re-exposes the +# result on _OTEL_EXPORT_PORT for Prometheus to scrape. Sitting alongside rather +# than replacing kube-prometheus-stack keeps this additive - the native series +# stay reachable, so a mapping that is wrong or missing costs nothing. +_OTEL_CHART = "opentelemetry-collector" +_OTEL_REPO = "https://open-telemetry.github.io/opentelemetry-helm-charts" +_OTEL_EXPORT_PORT = 8889 +# Pin the chart's resource names the way the Prometheus release does, so the +# collector's Service has a name a scrape config (and a test) can rely on rather +# than one derived from the release name. +_OTEL_FULLNAME_OVERRIDE = "otel-collector" + +# The pod label a MetricMapping selects an engine by, surfaced to OTTL as a +# resource attribute under this name. Extracted explicitly rather than relying on +# the k8sattributes default naming, which has changed across collector releases. +_OTEL_ENGINE_ATTR = "modelplane.engine" + +# Pod labels this function has to know by value, because a composition function +# cannot import another's package. compose-model-replica owns both - base.py's +# LABEL_ENGINE and LABEL_SERVING, and the engine port's name - so these three are +# a cross-function contract and have to change together. The serving one is +# spelled the way Prometheus service discovery exposes it, with dots and slashes +# collapsed to underscores. +_LABEL_ENGINE = "modelplane.ai/engine" +_LABEL_SERVING_SD = "modelplane_ai_serving" +_ENGINE_PORT_NAME = "http" + _DRA_DRIVER_NAMESPACE = "dra-driver-nvidia-gpu" # Upstream default for the DRA driver's NVIDIA_DRIVER_ROOT. A ServingStack whose # nvidiaDriverRoot differs from this is on a platform (GKE) that relocates the @@ -221,6 +251,122 @@ def _k8s_object( return obj +def _otel_statements(mappings: list[mmv1alpha1.MetricMapping]) -> list[str]: + """Render every MetricMapping into OTTL statements for the transform processor. + + One statement per rename and per added label, each gated on the engine the + mapping selects. The gate is what makes this label-driven rather than + name-driven: two mappings can rename the same source name (a fork of an + engine emits its upstream names), and a series from an engine with no mapping + has to come through untouched rather than be renamed by someone else's rule. + + Sorted so the rendered config is stable: an unordered dict would reshuffle + the statements between reconciles and churn the Helm release for no reason. + """ + out: list[str] = [] + for mapping in sorted(mappings, key=lambda m: _name(m.metadata)): + spec = mapping.spec + if spec is None: + continue + labels = (spec.selector.matchLabels if spec.selector else None) or {} + engine = labels.get(_LABEL_ENGINE) + # A mapping that selects nothing we can key on would apply everywhere. + # Skip it rather than let it rewrite another engine's series. + if not engine: + continue + gate = f'resource.attributes["{_OTEL_ENGINE_ATTR}"] == "{engine}"' + for source, target in sorted((spec.rename or {}).items()): + out.append(f'set(metric.name, "{target}") where metric.name == "{source}" and {gate}') + add = (spec.labels.add if spec.labels else None) or {} + for key, value in sorted(add.items()): + out.append(f'set(datapoint.attributes["{key}"], "{value}") where {gate}') + return out + + +def _otel_values(mappings: list[mmv1alpha1.MetricMapping]) -> dict: + """Helm values for the collector: scrape engines, rename, re-expose. + + The scrape keeps only pods carrying the serving label and only their port + named `http`, which is the engine's. Matching the port by name rather than + number is what makes this correct under prefill/decode, where the pd-sidecar + holds 8000 and the engine has moved to its own port. + """ + return { + "fullnameOverride": _OTEL_FULLNAME_OVERRIDE, + "mode": "deployment", + "replicaCount": 1, + "image": {"repository": "otel/opentelemetry-collector-contrib"}, + # servicePort as well as containerPort: the chart builds the Service from + # servicePort, so a port declared only on the container is exposed on the + # pod and unreachable through the Service. + "ports": { + "prom-export": { + "enabled": True, + "containerPort": _OTEL_EXPORT_PORT, + "servicePort": _OTEL_EXPORT_PORT, + "protocol": "TCP", + } + }, + "presets": {"kubernetesAttributes": {"enabled": True}}, + "config": { + "receivers": { + "prometheus": { + "config": { + "scrape_configs": [ + { + "job_name": "modelplane-engines", + "kubernetes_sd_configs": [{"role": "pod"}], + "relabel_configs": [ + { + "source_labels": [f"__meta_kubernetes_pod_label_{_LABEL_SERVING_SD}"], + "action": "keep", + "regex": ".+", + }, + { + "source_labels": ["__meta_kubernetes_pod_container_port_name"], + "action": "keep", + "regex": _ENGINE_PORT_NAME, + }, + ], + }, + ], + }, + }, + }, + "processors": { + "k8sattributes": { + "extract": { + "labels": [{"tag_name": _OTEL_ENGINE_ATTR, "key": _LABEL_ENGINE, "from": "pod"}], + }, + }, + "transform": {"metric_statements": [{"statements": _otel_statements(mappings)}]}, + }, + "exporters": {"prometheus": {"endpoint": f"0.0.0.0:{_OTEL_EXPORT_PORT}"}}, + "service": { + "pipelines": { + "metrics": { + "receivers": ["prometheus"], + "processors": ["k8sattributes", "transform"], + "exporters": ["prometheus"], + }, + }, + }, + }, + } + + +def _otel_release(version: str, provider_config: str, mappings: list[mmv1alpha1.MetricMapping]) -> helmv1beta1.Release: + """Build the OpenTelemetry collector release for a workload cluster.""" + return _helm_release( + chart=_OTEL_CHART, + repo=_OTEL_REPO, + version=version, + namespace=_PROMETHEUS_NAMESPACE, + provider_config=provider_config, + values=_otel_values(mappings), + ) + + def _prometheus_release(version: str, provider_config: str) -> helmv1beta1.Release: """Build a kube-prometheus-stack Helm release for a backend cluster.""" return _helm_release( @@ -318,6 +464,7 @@ def compose(self) -> None: self.compose_ai_gateway() self.compose_gaie_crds() self.compose_prometheus() + self.compose_otel_collector() self.compose_leader_worker_set() self.compose_node_feature_discovery() self.compose_dra_driver() @@ -613,6 +760,43 @@ def compose_prometheus(self) -> None: _prometheus_release(v.prometheus, _pc_name(self.xr)), # ty: ignore[invalid-argument-type] # XRD defaults this version and forbids null ) + def compose_otel_collector(self) -> None: + """Compose the metrics collector. Gated on ProviderConfigs being observed, + like every other release here. + + MetricMappings are required cluster-wide rather than selected. A mapping + names the engines it applies to by label, so which mappings exist is a + cluster-level fact rather than something a ServingStack spec repeats. + + The collector is composed whether or not any mapping exists: collection + is always on, and a mapping only changes what a series is named. + """ + pc_observed = self.provider_configs_observed() + if not (pc_observed or "otel-collector" in self.req.observed.resources): + return + + response.require_resources( + self.rsp, + name="metric-mappings", + api_version="modelplane.ai/v1alpha1", + kind="MetricMapping", + ) + # Crossplane re-calls once the requirement resolves. Composing before it + # does would install a collector with no renames, then rewrite its values + # when the mappings arrive and churn the release. + if "metric-mappings" not in self.req.required_resources: + return + + mappings = [ + mmv1alpha1.MetricMapping.model_validate(m) + for m in request.get_required_resources(self.req, "metric-mappings") + ] + v = self.xr.spec.versions or v1alpha1.Versions() + resource.update( + self.rsp.desired.resources["otel-collector"], + _otel_release(v.otelCollector, _pc_name(self.xr), mappings), # ty: ignore[invalid-argument-type] # XRD defaults this version and forbids null + ) + def compose_leader_worker_set(self) -> None: """Compose LeaderWorkerSet. Gated on ProviderConfigs being observed.""" pc_observed = self.provider_configs_observed() @@ -908,6 +1092,7 @@ def mark_readiness(self) -> None: "ai-gateway-crds", "ai-gateway", "prometheus", + "otel-collector", "leader-worker-set", "node-feature-discovery", "dra-driver", diff --git a/functions/compose-serving-stack/tests/test_fn.py b/functions/compose-serving-stack/tests/test_fn.py index 7f071b44a..4578e8e16 100644 --- a/functions/compose-serving-stack/tests/test_fn.py +++ b/functions/compose-serving-stack/tests/test_fn.py @@ -723,6 +723,16 @@ async def test_second_pass(self) -> None: ) want = fnv1.RunFunctionResponse( + # The collector step requires every MetricMapping cluster-wide once the + # ProviderConfigs are observed, so it is part of the response here. + requirements=fnv1.Requirements( + resources={ + "metric-mappings": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="MetricMapping", + ), + }, + ), meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), desired=fnv1.State( composite=fnv1.Resource( @@ -955,6 +965,103 @@ async def test_gateway_proxy_marked_ready_when_observed_ready(self) -> None: "gateway-proxy must be ready once provider-kubernetes observes it Ready", ) + async def test_collector_requested_then_composed(self) -> None: + """The collector goes through the whole function, not just its builders. + + Two asserts, because the earlier version of this code passed every + builder test while composing no collector at all: the function has to + ask for the MetricMappings, and once they resolve it has to put a + Release in the desired resources. + """ + req = _base_request() + for pc, api in ( + ("provider-config-helm", "helm.m.crossplane.io/v1beta1"), + ("provider-config-kubernetes", "kubernetes.m.crossplane.io/v1alpha1"), + ): + req.observed.resources[pc].CopyFrom( + fnv1.Resource(resource=resource.dict_to_struct({"apiVersion": api, "kind": "ProviderConfig"})), + ) + + # Before the requirement resolves: asked for, nothing composed. Composing + # here would install a collector with no renames and churn it later. + got = await self.runner.RunFunction(req, None) + self.assertIn("metric-mappings", got.requirements.resources) + self.assertNotIn("otel-collector", got.desired.resources) + + # Resolved: the mapping's rename reaches the rendered config, gated on + # the engine the mapping selects. + req.required_resources["metric-mappings"].items.append( + fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "modelplane.ai/v1alpha1", + "kind": "MetricMapping", + "metadata": {"name": "vllm", "namespace": "ml-team"}, + "spec": { + "selector": {"matchLabels": {"modelplane.ai/engine": "vllm"}}, + "rename": {"vllm:num_requests_waiting": "modelplane_requests_waiting"}, + }, + } + ), + ), + ) + got = await self.runner.RunFunction(req, None) + self.assertIn("otel-collector", got.desired.resources) + rendered = str( + json_format.MessageToDict(got).get("desired", {}).get("resources", {}).get("otel-collector", {}), + ) + self.assertIn("modelplane_requests_waiting", rendered) + self.assertIn(fn._OTEL_ENGINE_ATTR, rendered) + + async def test_collector_marked_ready_when_release_ready(self) -> None: + """An observed-Ready collector Release marks the composed resource ready. + + Without this the collector is composed but never marked ready, so the + ServingStack reports it as an unready resource forever, BackendReady + never goes true, and the InferenceCluster never becomes Ready. The + collector working is not enough -- it has to be *reported* as working. + """ + req = _base_request() + for pc, api in ( + ("provider-config-helm", "helm.m.crossplane.io/v1beta1"), + ("provider-config-kubernetes", "kubernetes.m.crossplane.io/v1alpha1"), + ): + req.observed.resources[pc].CopyFrom( + fnv1.Resource(resource=resource.dict_to_struct({"apiVersion": api, "kind": "ProviderConfig"})), + ) + req.required_resources["metric-mappings"].items.append( + fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "modelplane.ai/v1alpha1", + "kind": "MetricMapping", + "metadata": {"name": "vllm", "namespace": "ml-team"}, + "spec": {"rename": {"vllm:num_requests_waiting": "modelplane_requests_waiting"}}, + } + ), + ), + ) + + # Not yet observed: composed, and not claimed ready. + got = await self.runner.RunFunction(req, None) + self.assertIn("otel-collector", got.desired.resources) + self.assertNotEqual(got.desired.resources["otel-collector"].ready, fnv1.READY_TRUE) + + # Observed Ready: the composed resource is marked ready too. + req.observed.resources["otel-collector"].CopyFrom( + fnv1.Resource( + resource=resource.dict_to_struct( + { + "apiVersion": "helm.m.crossplane.io/v1beta1", + "kind": "Release", + "status": {"conditions": [{"type": "Ready", "status": "True"}]}, + } + ), + ), + ) + got = await self.runner.RunFunction(req, None) + self.assertEqual(got.desired.resources["otel-collector"].ready, fnv1.READY_TRUE) + async def test_third_pass(self) -> None: """Steady state: composed releases report Ready, and the gateway address is surfaced from the observed Object's manifest. The observed gateway Object @@ -1005,6 +1112,16 @@ async def test_third_pass(self) -> None: req.observed.resources[key].CopyFrom(observed) want = fnv1.RunFunctionResponse( + # The collector step requires every MetricMapping cluster-wide once the + # ProviderConfigs are observed, so it is part of the response here. + requirements=fnv1.Requirements( + resources={ + "metric-mappings": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="MetricMapping", + ), + }, + ), meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), desired=fnv1.State( composite=fnv1.Resource( diff --git a/schemas/.lock.json b/schemas/.lock.json index 64706d0fc..ca94906d1 100644 --- a/schemas/.lock.json +++ b/schemas/.lock.json @@ -1 +1 @@ -{"packages":{"fs://apis":"bc166b0716c9e27d8cf98a00cde6c077b818ef5186f9047ad49676e5c1c372a3","git://https://github.com/crossplane/crossplane/cluster/crds":"90d8b72ad8b829f0bcd7d7d5a98eaa0d579f244a","xpkg://xpkg.upbound.io/modelplane/provider-helm:v1.3.0-9a6fb4b":"sha256:af1858ee7dcabc9149bf2f8005bffb298961519c066b8e7c21a244039ccd066b","xpkg://xpkg.upbound.io/modelplane/provider-kubernetes:v1.2.1-070dae7":"sha256:718b481f5b760f5436e162e997fedf74c3b5e950c4f9d824a990973df071676b","xpkg://xpkg.upbound.io/upbound/provider-aws-ec2:v2.6.0":"sha256:acc26c8d2710e0306185b6c626a2f8c8fe0fdf89874e85efe7944a3668322865","xpkg://xpkg.upbound.io/upbound/provider-aws-efs:v2.6.0":"sha256:00f1bbbb3c0f1948b6dd45c841a083d63e41850f5a559a15580915311f404911","xpkg://xpkg.upbound.io/upbound/provider-aws-eks:v2.6.0":"sha256:5d144b19e188cb96c918aa7e4ccbc6759b8733ff09bd8ce412723c669aa3f763","xpkg://xpkg.upbound.io/upbound/provider-aws-iam:v2.6.0":"sha256:dbc5288589ccb302d527565680477f08477c280fc5c616dda95dfd558108a038","xpkg://xpkg.upbound.io/upbound/provider-azure-containerservice:v2.6.0":"sha256:7d8a9bb3eb168e6eef0694253a23321fa98acad8b897ade168a4f0bb89985fed","xpkg://xpkg.upbound.io/upbound/provider-azure-network:v2.6.0":"sha256:0d83bc4964488e5602b56dd7680e4bae0fbd5fa94b3a9af403d8431ded45635f","xpkg://xpkg.upbound.io/upbound/provider-family-azure:v2.6.0":"sha256:1f2f597d5702ccb241429f0d8942f3c21e0ea55fffcef4ff9dca76b32bdc5dff","xpkg://xpkg.upbound.io/upbound/provider-family-gcp:v2.6.0":"sha256:2e33bfd0f501155e7be63f13471d2f87f00519f99b916b7cd63ff40faa517145","xpkg://xpkg.upbound.io/upbound/provider-gcp-cloudplatform:v2.6.0":"sha256:f1fe8bc55c474464642303e6fa8608c83e369b42ff12bb8a60a3e2d77339a52b","xpkg://xpkg.upbound.io/upbound/provider-gcp-compute:v2.6.0":"sha256:c7417c461d403f0d59a2dd83f242cebd5d738f9ad328e03ffe0c7b99ea251635","xpkg://xpkg.upbound.io/upbound/provider-gcp-container:v2.6.0":"sha256:b3f68d01ab2529026f1a5dd6d6215a2f499710fad4edd4850546bf6c756befb6","xpkg://xpkg.upbound.io/upbound/provider-nebius:v1.0.1":"sha256:f14e09e8e35c2d4bf2da527191065c3e457850c404238fa4619493e0c2bfd13f"}} \ No newline at end of file +{"packages":{"fs://apis":"bc3a0da4067a49eef3bd2ce33a9086f0f2505d212ef9d248b822b38c59692f76","git://https://github.com/crossplane/crossplane/cluster/crds":"90d8b72ad8b829f0bcd7d7d5a98eaa0d579f244a","xpkg://xpkg.upbound.io/modelplane/provider-helm:v1.3.0-9a6fb4b":"sha256:af1858ee7dcabc9149bf2f8005bffb298961519c066b8e7c21a244039ccd066b","xpkg://xpkg.upbound.io/modelplane/provider-kubernetes:v1.2.1-070dae7":"sha256:718b481f5b760f5436e162e997fedf74c3b5e950c4f9d824a990973df071676b","xpkg://xpkg.upbound.io/upbound/provider-aws-ec2:v2.6.0":"sha256:acc26c8d2710e0306185b6c626a2f8c8fe0fdf89874e85efe7944a3668322865","xpkg://xpkg.upbound.io/upbound/provider-aws-efs:v2.6.0":"sha256:00f1bbbb3c0f1948b6dd45c841a083d63e41850f5a559a15580915311f404911","xpkg://xpkg.upbound.io/upbound/provider-aws-eks:v2.6.0":"sha256:5d144b19e188cb96c918aa7e4ccbc6759b8733ff09bd8ce412723c669aa3f763","xpkg://xpkg.upbound.io/upbound/provider-aws-iam:v2.6.0":"sha256:dbc5288589ccb302d527565680477f08477c280fc5c616dda95dfd558108a038","xpkg://xpkg.upbound.io/upbound/provider-azure-containerservice:v2.6.0":"sha256:7d8a9bb3eb168e6eef0694253a23321fa98acad8b897ade168a4f0bb89985fed","xpkg://xpkg.upbound.io/upbound/provider-azure-network:v2.6.0":"sha256:0d83bc4964488e5602b56dd7680e4bae0fbd5fa94b3a9af403d8431ded45635f","xpkg://xpkg.upbound.io/upbound/provider-family-azure:v2.6.0":"sha256:1f2f597d5702ccb241429f0d8942f3c21e0ea55fffcef4ff9dca76b32bdc5dff","xpkg://xpkg.upbound.io/upbound/provider-family-gcp:v2.6.0":"sha256:2e33bfd0f501155e7be63f13471d2f87f00519f99b916b7cd63ff40faa517145","xpkg://xpkg.upbound.io/upbound/provider-gcp-cloudplatform:v2.6.0":"sha256:f1fe8bc55c474464642303e6fa8608c83e369b42ff12bb8a60a3e2d77339a52b","xpkg://xpkg.upbound.io/upbound/provider-gcp-compute:v2.6.0":"sha256:c7417c461d403f0d59a2dd83f242cebd5d738f9ad328e03ffe0c7b99ea251635","xpkg://xpkg.upbound.io/upbound/provider-gcp-container:v2.6.0":"sha256:b3f68d01ab2529026f1a5dd6d6215a2f499710fad4edd4850546bf6c756befb6","xpkg://xpkg.upbound.io/upbound/provider-nebius:v1.0.1":"sha256:f14e09e8e35c2d4bf2da527191065c3e457850c404238fa4619493e0c2bfd13f"}} \ No newline at end of file diff --git a/schemas/python/models/ai/modelplane/infrastructure/servingstack/v1alpha1.py b/schemas/python/models/ai/modelplane/infrastructure/servingstack/v1alpha1.py index d7ba1d2d4..31139286a 100644 --- a/schemas/python/models/ai/modelplane/infrastructure/servingstack/v1alpha1.py +++ b/schemas/python/models/ai/modelplane/infrastructure/servingstack/v1alpha1.py @@ -116,6 +116,10 @@ class Versions(BaseModel): """ NVIDIA DRA driver chart version. Publishes GPUs as DRA ResourceSlices and the gpu.nvidia.com DeviceClass that ModelReplica ResourceClaims bind through. """ + otelCollector: constr(min_length=1, max_length=32) | None = '0.116.0' + """ + OpenTelemetry Collector chart version. The collector scrapes each engine's metrics and renames them onto the modelplane_* surface from the cluster's MetricMappings. + """ prometheus: constr(min_length=1, max_length=32) | None = '72.6.2' """ kube-prometheus-stack chart version. diff --git a/schemas/python/models/ai/modelplane/metricmapping/__init__.py b/schemas/python/models/ai/modelplane/metricmapping/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/schemas/python/models/ai/modelplane/metricmapping/v1alpha1.py b/schemas/python/models/ai/modelplane/metricmapping/v1alpha1.py new file mode 100644 index 000000000..3bd0950ad --- /dev/null +++ b/schemas/python/models/ai/modelplane/metricmapping/v1alpha1.py @@ -0,0 +1,127 @@ +# generated by datamodel-codegen: +# filename: workdir/modelplane_ai_v1alpha1_metricmapping.yaml + +from __future__ import annotations + +from typing import Literal + +from pydantic import AwareDatetime, BaseModel + +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 Labels(BaseModel): + add: dict[str, str] | None = None + """ + Labels to add to every matched series. + """ + + +class Selector(BaseModel): + matchLabels: dict[str, str] | None = None + + +class Spec(BaseModel): + crossplane: Crossplane | None = None + """ + Configures how Crossplane will reconcile this composite resource + """ + labels: Labels | None = None + """ + Label rewrites applied to the matched series. + """ + rename: dict[str, str] | None = None + """ + Source metric name to its modelplane_* name, e.g. vllm:time_to_first_token_seconds becomes modelplane_time_to_first_token. + """ + selector: Selector | None = None + """ + Selects the pods this mapping applies to, by label. An engine mapping matches the engine-type label (modelplane.ai/engine); a scheduler mapping matches the scheduler's pods. + """ + + +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 MetricMapping(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['MetricMapping'] | None = 'MetricMapping' + """ + 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 + """ + How to normalize one component's Prometheus metrics onto the modelplane_* surface. The metrics collector reads every MetricMapping and renders the matching rename and label rules into its config. Modelplane ships a MetricMapping per common engine; a platform team applies one more for a new or forked engine. + """ + status: Status | None = None + + +class MetricMappingList(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[MetricMapping] + """ + List of metricmappings. 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/modeldeployment/v1alpha1.py b/schemas/python/models/ai/modelplane/modeldeployment/v1alpha1.py index 69a73bff1..6eb812678 100644 --- a/schemas/python/models/ai/modelplane/modeldeployment/v1alpha1.py +++ b/schemas/python/models/ai/modelplane/modeldeployment/v1alpha1.py @@ -210,6 +210,10 @@ class Engine(BaseModel): """ The engine's phase in a PrefillDecode deployment, Prefill or Decode. Set only when serving.mode is PrefillDecode, where exactly one engine takes each phase. """ + type: constr(min_length=1, max_length=63) | None = None + """ + Which engine this runs, e.g. vllm or sglang. Stamped onto the serving pods as modelplane.ai/engine, where a MetricMapping selects on it to normalize that engine's metrics onto the modelplane_* surface. Free-form rather than an enum: a platform team adds a MetricMapping for a forked or new engine without a Modelplane release, and an engine with no mapping is still scraped under its native names. Omit it and the engine's metrics are collected unnormalized. + """ class ModelCacheRef(BaseModel): diff --git a/schemas/python/models/ai/modelplane/modelreplica/v1alpha1.py b/schemas/python/models/ai/modelplane/modelreplica/v1alpha1.py index d178c542c..474eaf8a8 100644 --- a/schemas/python/models/ai/modelplane/modelreplica/v1alpha1.py +++ b/schemas/python/models/ai/modelplane/modelreplica/v1alpha1.py @@ -148,6 +148,10 @@ class Engine(BaseModel): members: list[Member] = Field(..., max_length=2, min_length=1) name: constr(min_length=1, max_length=63) phase: Literal['Prefill', 'Decode'] | None = None + type: constr(min_length=1, max_length=63) | None = None + """ + Which engine this runs, e.g. vllm or sglang. Stamped onto the serving pods as modelplane.ai/engine, where a MetricMapping selects on it to normalize that engine's metrics onto the modelplane_* surface. Free-form rather than an enum: a platform team adds a MetricMapping for a forked or new engine without a Modelplane release, and an engine with no mapping is still scraped under its native names. Omit it and the engine's metrics are collected unnormalized. + """ class ModelCacheRef(BaseModel): diff --git a/uv.lock b/uv.lock index 7f452b153..902bb6b78 100644 --- a/uv.lock +++ b/uv.lock @@ -14,6 +14,7 @@ members = [ "compose-inference-class", "compose-inference-cluster", "compose-inference-gateway", + "compose-metric-mapping", "compose-model-cache", "compose-model-deployment", "compose-model-endpoint", @@ -188,6 +189,25 @@ requires-dist = [ { name = "pyyaml", specifier = ">=6.0" }, ] +[[package]] +name = "compose-metric-mapping" +version = "0.0.0" +source = { editable = "functions/compose-metric-mapping" } +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-model-cache" version = "0.0.0"