From c2e4a21a6f9d822962a9df6fe6dfed9ba4690ac1 Mon Sep 17 00:00:00 2001 From: kerthcet Date: Sat, 15 Aug 2026 11:37:05 +0100 Subject: [PATCH 1/4] add metrics Signed-off-by: kerthcet --- README.md | 1 + config/manager/kustomization.yaml | 4 +- docs/add-a-provider.md | 27 ++ docs/architecture.md | 38 ++- docs/metrics.md | 208 ++++++++++++ go.mod | 7 +- internal/controller/placement_metrics_test.go | 318 ++++++++++++++++++ .../controller/pod_placement_controller.go | 13 +- internal/controller/pod_placement_helpers.go | 44 +++ pkg/metrics/helper_test.go | 32 ++ pkg/metrics/metrics.go | 269 +++++++++++++++ pkg/metrics/metrics_test.go | 141 ++++++++ pkg/metrics/placement.go | 145 ++++++++ pkg/provider/errors.go | 99 +++++- pkg/provider/errors_test.go | 64 ++++ pkg/vnode/handler.go | 184 ++++++++-- pkg/vnode/handler_test.go | 97 ++++++ pkg/vnode/metrics_test.go | 264 +++++++++++++++ 18 files changed, 1914 insertions(+), 41 deletions(-) create mode 100644 docs/metrics.md create mode 100644 internal/controller/placement_metrics_test.go create mode 100644 pkg/metrics/helper_test.go create mode 100644 pkg/metrics/metrics.go create mode 100644 pkg/metrics/metrics_test.go create mode 100644 pkg/metrics/placement.go create mode 100644 pkg/vnode/metrics_test.go diff --git a/README.md b/README.md index 6f97cdd..2b541ab 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,7 @@ placement controller owns those. - See [docs/architecture.md](docs/architecture.md) for design details. - See [docs/status.md](docs/status.md) for how instance lifecycle becomes Pod and NodeClaim status, per provider. +- See [docs/metrics.md](docs/metrics.md) for what is instrumented and how to query it. ## License diff --git a/config/manager/kustomization.yaml b/config/manager/kustomization.yaml index 086bc1e..547a445 100644 --- a/config/manager/kustomization.yaml +++ b/config/manager/kustomization.yaml @@ -4,5 +4,5 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization images: - name: controller - newName: inftyai/nebula-controller - newTag: latest + newName: example.com/nebula + newTag: v0.0.1 diff --git a/docs/add-a-provider.md b/docs/add-a-provider.md index 6552211..a1cfba0 100644 --- a/docs/add-a-provider.md +++ b/docs/add-a-provider.md @@ -32,6 +32,33 @@ The Pod is the single source of truth for the workload shape; `ProvisionRequest` carries only what the Pod cannot express (the optimizer's capacity tier and the claim identity). Do not duplicate Pod fields onto the request. +### Wrap the errors your `Provision` returns + +`ClassifyProvisionError` decides *how widely* to blocklist, but a separate predicate — +`provider.IsRejection` — decides *whether to blocklist at all*, and whether the Pod is +failed. It answers: did the provider make a decision about this request ("no capacity", +"over quota", "bad credentials"), or did we merely fail to find out what it would have +decided (a transport error, a timeout, a 503)? + +Only a **decision** is acted on. An unattributable failure leaves the Pod +non-terminal at `Provisioning` for the pod controller to retry, and records nothing — +because failing a Pod there would stamp a terminal verdict on a request the provider may +well have accepted, reaping the Pod out from under a paid instance whose id was never +returned. + +What this asks of an adapter: **wrap every error your `Provision` path returns with the +matching sentinel** (`fmt.Errorf("...: %w", provider.ErrNoCapacity)`). A wrapped sentinel +always outranks the message text, so it is the only way to be certain of the outcome. +Unwrapped errors fall back to a string heuristic that recognizes the obvious API +phrasings and treats transport markers (`rpc error`, `connection refused`, `503`, `EOF`) +as unattributable — a reasonable default, but not one to rely on for a condition you can +classify yourself. + +The metrics say when an adapter has skipped this: an unwrapped rejection lands on +`nebula_provision_failures_total{reason="other"}`, so a sustained rate on that series for +your provider is a to-do list of conditions still to wrap (see +[metrics.md](metrics.md)). + Most adapters embed `catalog.Base` for the generic `Name`, `Offerings`, and the identity `MapAccelerator`, overriding only what the provider does differently (see how `pkg/provider/modal` embeds it). diff --git a/docs/architecture.md b/docs/architecture.md index ac889a2..dd7d720 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -28,6 +28,7 @@ mapping), see [docs/status.md](status.md). - [Provider Abstraction](#6-provider-abstraction) - [Placement Optimizer and Poll Loop](#7-placement-optimizer-and-poll-loop) - [CRDs](#crds) +- [Observability](#observability) - [Failure Domains and HA](#failure-domains-and-ha) - [Current Implementation Status](#current-implementation-status) @@ -538,6 +539,29 @@ paths as direct Pod/Deployment workloads. --- +## Observability + +The instrumented surface is the path a Pod takes from admission to a running external +instance: **placement**, then **provisioning**. Those are the parts whose cost and +failure modes are otherwise invisible — placement can silently leave a Pod gated forever, +and provisioning runs against a third party, takes seconds to minutes, bills money, and +fails for reasons the Pod status flattens away. Everything else is covered elsewhere and +deliberately not duplicated: reconcile counts, queue depth and API latency by +controller-runtime's own collectors, Pod-population questions by kube-state-metrics. + +Two design rules are worth knowing here, because they constrain the code above: + +- Placement and provisioning metrics share **one label set** (`provider`, `region`, + `capacity_type`, `accelerator`, `accelerator_count`), so a placement and the + provisioning attempt it led to join in PromQL without label surgery. That is why + `Handler.metricLabels` and `placementLabels` are field-for-field mirrors. +- Every label is bounded by **configuration** (NodePools, provider catalogs), never by + workload. Nothing derived from a Pod name, UID, namespace or unresolved user-supplied + pool label is ever a label value. + +See [metrics.md](metrics.md) for the full series list, label semantics, example queries +and the two known gaps. + ## Failure Domains and HA Components on the fail-closed path: @@ -560,7 +584,19 @@ Components designed to degrade without leaks: manager restarts with credentials available. - **Provider List temporarily failing.** VK skips that poll tick and retries on the next cadence. The NodeClaim finalizer does not release on transient list - errors, so teardown retries instead of abandoning the instance. + errors, so teardown retries instead of abandoning the instance. A `List` failure + during post-restart re-adoption is likewise treated as *unknown*, not *absent*: + `Handler.GetPod` returns a non-nil Pod together with a non-NotFound error, which + suppresses both a duplicate `CreatePod` and a premature `DeletePod` until the + provider answers. Conflating the two would let one failed list mark a healthy Pod + `Failed` for reaping while the paid instance kept running behind a zero instance id. +- **Provider unreachable during `Provision`.** Distinguished from a *rejection* + (`provider.IsRejection`). Only a rejection — no capacity, quota, auth, unsupported + accelerator — fails the Pod and files a blocklist entry. A transport error, timeout + or 503 leaves the Pod non-terminal at `Provisioning` with the error as its message + and records nothing, because it is not evidence about the request: the provider may + have accepted it. `Provision` is idempotent on `ClaimName`, so the retry adopts + whatever the failed attempt created rather than doubling it. Leader election (`LeaderElectionID: nebula.inftyai.com`) keeps a single active manager reconciling controllers and owning virtual-node leases. diff --git a/docs/metrics.md b/docs/metrics.md new file mode 100644 index 0000000..201bf01 --- /dev/null +++ b/docs/metrics.md @@ -0,0 +1,208 @@ +# Metrics + +What Nebula instruments, and what each series is for. + +The instrumented surface is the path a Pod takes from admission to a running external +instance — **placement**, then **provisioning**: + +``` +Pod created (gated) + | nebula_placement_wait_duration_seconds + | nebula_placement_deferrals_total <- why it is still waiting + | nebula_placement_candidate_skips_total <- why a candidate was passed over + v +placed (gate removed) nebula_placement_decisions_total + | nebula_provision_duration_seconds <- the provider API call alone + | nebula_provision_attempts_total + | nebula_provision_failures_total + v +instance accepted + | nebula_instance_ready_duration_seconds + v +Running +``` + +Those are the parts whose cost and failure modes are otherwise invisible: placement can +silently leave a Pod gated forever, and provisioning runs against a third party, takes +seconds to minutes, bills money, and fails for reasons the Pod status flattens away. +Everything else is already covered elsewhere and deliberately not duplicated here — +reconcile counts, queue depth and API latency by controller-runtime's own collectors, and +Pod-population questions ("how many Pods are gated right now?") by kube-state-metrics. + +- [Where they are served](#where-they-are-served) +- [Placement](#placement) +- [Provisioning](#provisioning) +- [Label semantics](#label-semantics) +- [Example queries](#example-queries) +- [Known gaps](#known-gaps) + +## Where they are served + +Every collector registers into controller-runtime's registry (see `pkg/metrics`; the +`init` in each file is what registers them, so importing the package is the only wiring), +which means they are served on the manager's existing `--metrics-bind-address` endpoint +alongside the standard controller and workqueue metrics. In the default overlay that is +`:8443` with authn/authz, so a scrape needs a bearer token whose subject is bound to the +`nebula-metrics-reader` ClusterRole (`config/rbac/metrics_reader_role.yaml`, which grants +`get` on `/metrics`). + +## Placement + +Where Pods land, and why they don't. + +| Metric | Type | What it answers | +| --- | --- | --- | +| `nebula_placement_decisions_total` | counter | Where Pods actually land. The `capacity_type` breakdown is the cost question: a fleet sliding from Spot to OnDemand is a regression with no error anywhere. | +| `nebula_placement_wait_duration_seconds` | histogram | How long a Pod sat gated, from Pod creation to the gate being removed. | +| `nebula_placement_deferrals_total{pool,reason}` | counter | *Why* a reconcile placed nothing. | +| `nebula_placement_candidate_skips_total{provider,capacity_type,region,reason}` | counter | Why the walk passed over one candidate. The only view into failover actually working. | + +The deferral `reason` is a closed set, and each value points at a **different owner** — +which is the whole reason for splitting them: + +| `reason` | Means | Clears when | +| --- | --- | --- | +| `no_pool` | The Pod names a NodePool that does not exist, or carries no pool label. | A human fixes the Pod (or the workload generating it). | +| `invalid_request` | The accelerator request is malformed — e.g. `nvidia.com/gpu` with no accelerator-type label. It is *not* treated as CPU-only. | A human fixes the Pod spec. | +| `all_blocked` | A servable candidate exists, but failover is holding every one of them off. | By itself — the Pod is already requeued for the block's expiry. | +| `no_candidate` | No provider in the pool can serve this request at all. | An operator adds a provider, or a provider registers. | +| `stale_claim` | A NodeClaim from a prior same-named Pod has not been reaped yet. | By itself, in seconds. A sustained rate means the NodeClaim backstop is stuck. | + +The skip `reason` is likewise closed: `provider_unregistered`, +`capacity_type_unsupported`, `accelerator_unsupported`, `blocked`. Only `blocked` clears +on its own. One reconcile can file several skips — the walk visits every candidate before +giving up. + +`nebula_placement_deferrals_total` counts **deferrals, not Pods**. A gated Pod is +reconciled again on every requeue and resync, so one Pod stuck for an hour contributes +many increments. The rate is therefore a measure of placement pressure, not a population: +for "how many Pods are stuck right now" read the SchedulingGated Pod count from +kube-state-metrics, and use this series to explain *why*. + +## Provisioning + +What the external call cost, and how it failed. + +| Metric | Type | What it answers | +| --- | --- | --- | +| `nebula_provision_attempts_total{result}` | counter | Provisioning volume and error rate, per candidate. | +| `nebula_provision_failures_total{reason}` | counter | *Why* provisioning fails. | +| `nebula_provision_duration_seconds{result}` | histogram | Latency of the `Provision` call alone. AWS sweeps a region's availability zones inside it, so a capacity shortage shows up as latency *here*; Modal returns as soon as the sandbox is accepted and the wait moves to the next metric. | +| `nebula_instance_ready_duration_seconds` | histogram | The whole user-visible wait, from `CreatePod` to the first poll tick reporting `Running` — including provider-side queueing, image pull, GPU attach and up to one poll interval of detection lag. | + +`nebula_provision_failures_total` deliberately overlaps +`nebula_provision_attempts_total{result="failure"}` rather than adding a `reason` label +there: `reason` is only meaningful on failure, and carrying it on the attempts counter +would multiply the success series by a label that is constant for them. + +The failure `reason` is a coarse, closed set, mapped from the shared sentinels in +`pkg/provider` — *not* from message text, which is what keeps the label bounded: + +| `reason` | Means | +| --- | --- | +| `capacity` | `ErrNoCapacity` — the provider has no capacity for this shape. | +| `quota` | `ErrQuota` — our account limit, not the provider's supply. | +| `auth` | `ErrAuth` — credentials or permissions. | +| `unsupported_accelerator` | `ErrUnsupportedAccelerator` — the request cannot be honoured here at all. | +| `timeout` | The `Provision` call hit its own deadline without a capacity cause. | +| `unreachable` | The provider never told us what it decided. | +| `other` | The provider *did* reject the request, but the adapter returned a raw API error without wrapping a sentinel, so the category was unavailable. | + +Fine-grained detail is deliberately *not* here: it stays where it is already available +(the Pod's `Failed` status message and the `vnode-handler` error log). These labels exist +to answer "are we losing capacity, or are our credentials broken?" at a glance. + +## Label semantics + +Every series except the `{pool,reason}` and `{provider,capacity_type,region,reason}` +diagnostics carries the same label set: + +``` +provider region capacity_type accelerator accelerator_count +``` + +That is on purpose: a placement and the provisioning attempt it led to carry **identical +label values**, so the two join in PromQL without label surgery — "placed on Spot but +never provisioned" is one query. For the same reason the two duration histograms measure +adjacent legs of one journey: `placement_wait` ends exactly where `instance_ready` +begins, so together they cover `kubectl apply` to `Running`. + +Five label values are load-bearing: + +- **`none`** is the placeholder for a label the request genuinely did not carry: no region + (unconstrained), no capacity tier (the provider's default), no accelerator (a CPU-only + Pod). An explicit token beats an empty string, which in PromQL is indistinguishable from + a label that was never set and silently matches `{region=""}` selectors nobody meant to + write. +- **`pool`** on the deferral counter is only ever the name of a NodePool that *exists*, or + `none`. The pool a Pod asks for is a Pod label — user-controlled and unbounded — so the + `no_pool` deferral files `none` rather than the unresolved string; a mislabeled workload + must not be able to mint a time series per typo. +- **`accelerator` and `accelerator_count`** are two labels, not the joined `H100:8` pool + identity the failover blocklist uses as its key. The two want opposite things from the + same pair: a blocklist needs one opaque key so an `H100:8` shortage never excludes + `H100:1`, while a metric needs two dimensions so `sum by (accelerator)` spans every size + and "all 8-GPU requests, whatever the type" is expressible at all. The pool key stays + recoverable as `accelerator + ":" + accelerator_count` when correlating a counted + failure with an excluded candidate. A CPU-only Pod renders both as `none` — not `0`, + which would land it in the numeric series read as real counts. +- **`region`** is the provider's own token, not necessarily one region. For a provider that + collapses every declared region into a single candidate (Modal) it is the joined form — + the same value `NodeClaimSpec.Region` carries. +- **`reason="unreachable"`** means the provider never told us what it decided: a transport + failure, a 503, an unparseable response. It is the one failure reason for which Nebula + deliberately does *not* fail the Pod or blocklist the candidate (see + `provider.IsRejection`), so a spike here alongside flat `capacity`/`auth` series is a + network or provider-outage problem, not a placement one. + +Cardinality is bounded by *configuration*, not by workload: providers x regions x tiers x +accelerator pools, all of which come from NodePools and provider catalogs. Nothing +derived from a Pod name, UID or namespace is ever a label. + +## Example queries + +```promql +# Provisioning error rate, by candidate. +sum by (provider, region, capacity_type) (rate(nebula_provision_attempts_total{result="failure"}[15m])) + / sum by (provider, region, capacity_type) (rate(nebula_provision_attempts_total[15m])) + +# Cost regression: what fraction of placements are falling back off Spot? +sum(rate(nebula_placement_decisions_total{capacity_type="OnDemand"}[1h])) + / sum(rate(nebula_placement_decisions_total[1h])) + +# ...and whether failover explains it. +sum by (provider, region) (rate(nebula_placement_candidate_skips_total{capacity_type="Spot",reason="blocked"}[1h])) + +# Placements that never led to a provisioning attempt (stuck between the two halves). +sum by (provider, region, capacity_type) (increase(nebula_placement_decisions_total[1h])) + - sum by (provider, region, capacity_type) (increase(nebula_provision_attempts_total[1h])) + +# End-to-end p95, apply to Running: the two legs summed. +histogram_quantile(0.95, sum by (le) (rate(nebula_placement_wait_duration_seconds_bucket[1h]))) + + histogram_quantile(0.95, sum by (le) (rate(nebula_instance_ready_duration_seconds_bucket[1h]))) + +# Adapters that are not wrapping their errors (a to-do, not an incident). +sum by (provider) (rate(nebula_provision_failures_total{reason="other"}[6h])) + +# Requests nobody is retrying their way out of: a human has to fix these. +sum by (reason) (rate(nebula_placement_deferrals_total{reason=~"no_pool|invalid_request"}[1h])) +``` + +## Known gaps + +Both are deliberate, and both bias toward looking *better* than reality — worth knowing +before trusting a dashboard. + +- **Counters reset on restart.** Every collector is in-process. This is ordinary Prometheus + semantics — `rate()` and `increase()` detect resets — but no cumulative history survives + a redeploy; the scrape backend owns durability. +- **`instance_ready_duration` under-samples slow boots.** The start timestamp lives only in + the virtual node's in-memory tracking map, so a provision still in flight when the + manager restarts is re-adopted without one and is never observed. A missing sample beats + a wrong one — measuring from re-adoption would report a wait of minutes as microseconds — + but the bias runs toward looking fast. Closing it needs the start time persisted durably + (a Pod annotation or NodeClaim status field), which is a write on the provisioning path. + +`placement_wait_duration` has no equivalent gap: its start timestamp is the Pod's own +`creationTimestamp`, so a placement that happens after a manager restart still reports the +true total wait. diff --git a/go.mod b/go.mod index 77c3fb7..95873e5 100644 --- a/go.mod +++ b/go.mod @@ -11,13 +11,16 @@ require ( github.com/onsi/ginkgo/v2 v2.27.2 github.com/onsi/gomega v1.38.2 github.com/open-policy-agent/cert-controller v0.14.0 + github.com/prometheus/client_golang v1.22.0 github.com/prometheus/client_model v0.6.1 github.com/virtual-kubelet/virtual-kubelet v1.11.0 + google.golang.org/grpc v1.78.0 k8s.io/api v0.33.4 k8s.io/apimachinery v0.33.4 k8s.io/client-go v0.33.4 k8s.io/utils v0.0.0-20241104100929-3ea5e8cea738 sigs.k8s.io/controller-runtime v0.21.0 + sigs.k8s.io/yaml v1.4.0 ) require ( @@ -70,6 +73,7 @@ require ( github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kisielk/og-rek v1.3.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/mailru/easyjson v0.7.7 // indirect github.com/moby/spdystream v0.5.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect @@ -78,7 +82,6 @@ require ( github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/pkg/errors v0.9.1 // indirect - github.com/prometheus/client_golang v1.22.0 // indirect github.com/prometheus/common v0.62.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect github.com/spf13/cobra v1.8.1 // indirect @@ -112,7 +115,6 @@ require ( gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20251029180050-ab9386a59fda // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 // indirect - google.golang.org/grpc v1.78.0 // indirect google.golang.org/protobuf v1.36.11 // indirect gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect @@ -126,5 +128,4 @@ require ( sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect sigs.k8s.io/randfill v1.0.0 // indirect sigs.k8s.io/structured-merge-diff/v4 v4.6.0 // indirect - sigs.k8s.io/yaml v1.4.0 // indirect ) diff --git a/internal/controller/placement_metrics_test.go b/internal/controller/placement_metrics_test.go new file mode 100644 index 0000000..0bd1326 --- /dev/null +++ b/internal/controller/placement_metrics_test.go @@ -0,0 +1,318 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +package controller + +import ( + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + dto "github.com/prometheus/client_model/go" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" + "github.com/InftyAI/Nebula/pkg/failover" + "github.com/InftyAI/Nebula/pkg/metrics" + "github.com/InftyAI/Nebula/pkg/util" +) + +// The collectors are package-level and registered once, so every test in this package +// shares them. Assertions are on the DELTA a reconcile produced, never on an absolute +// value — otherwise they would pass or fail depending on run order. + +func counterVal(t *testing.T, vec *prometheus.CounterVec, l prometheus.Labels) float64 { + t.Helper() + c, err := vec.GetMetricWith(l) + if err != nil { + t.Fatalf("GetMetricWith(%v): %v", l, err) + } + return testutil.ToFloat64(c) +} + +// histStats reads a histogram series' sample count and sum. GetMetricWith creates the +// series when absent, so an unobserved label set reads as (0, 0) rather than erroring. +func histStats(t *testing.T, vec *prometheus.HistogramVec, l prometheus.Labels) (uint64, float64) { + t.Helper() + obs, err := vec.GetMetricWith(l) + if err != nil { + t.Fatalf("GetMetricWith(%v): %v", l, err) + } + pb := &dto.Metric{} + if err := obs.(prometheus.Metric).Write(pb); err != nil { + t.Fatalf("write metric %v: %v", l, err) + } + return pb.GetHistogram().GetSampleCount(), pb.GetHistogram().GetSampleSum() +} + +// decisionLabels is the candidate label set for a Pod built by gatedPod: the count is 1 +// because an accelerator-type label with no explicit nvidia.com/gpu limit means one GPU, +// and the region is the placeholder because poolWith declares none. +func decisionLabels(prov string, tier nebulav1alpha1.CapacityType, region, accel string) prometheus.Labels { + return prometheus.Labels{ + "provider": prov, + "region": orNoneLabel(region), + "capacity_type": orNoneLabel(string(tier)), + "accelerator": orNoneLabel(accel), + "accelerator_count": "1", + } +} + +// skipLabels is the candidate-skip label set. A skip decided before the walk reaches +// the region axis carries no region, which renders as the placeholder. +func skipLabels(prov string, tier nebulav1alpha1.CapacityType, region, reason string) prometheus.Labels { + return prometheus.Labels{ + "provider": orNoneLabel(prov), + "capacity_type": orNoneLabel(string(tier)), + "region": orNoneLabel(region), + "reason": reason, + } +} + +func orNoneLabel(s string) string { + if s == "" { + return "none" + } + return s +} + +// Every reason a Pod goes unplaced must land on its OWN series, because each points at a +// different owner: no_pool/invalid_request need a human to fix the request, all_blocked +// clears itself, and no_candidate needs an operator to widen the pool. Collapsing any two +// of them would make the metric unactionable. +func TestPlacement_DeferralReasons(t *testing.T) { + tests := []struct { + name string + pool string // the pool label expected on the metric + want string + // build returns the objects to seed and the providers to register. + build func() ([]client.Object, []*fakeProvider, Blocklister) + }{{ + // The Pod names a pool that does not exist. The pool label is deliberately the + // placeholder, NOT the unresolved name — that string is a user-controlled Pod + // label, so filing it would mint a series per typo. + name: "missing pool", + pool: "none", + want: metrics.DeferNoPool, + build: func() ([]client.Object, []*fakeProvider, Blocklister) { + return []client.Object{gatedPod("d1", "default", "uid-d1", "ghost-pool", "H100")}, + []*fakeProvider{{name: "p1"}}, nil + }, + }, { + // nvidia.com/gpu with no accelerator-type label: malformed, not CPU-only. + name: "invalid accelerator request", + pool: "pool", + want: metrics.DeferInvalidRequest, + build: func() ([]client.Object, []*fakeProvider, Blocklister) { + pod := gatedPod("d1", "default", "uid-d1", "pool", "") + pod.Spec.Containers[0].Resources.Limits = corev1.ResourceList{ + util.NvidiaGPUResource: resource.MustParse("1"), + } + pool := poolWith("pool", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, "p1") + return []client.Object{pod, pool}, []*fakeProvider{{name: "p1"}}, nil + }, + }, { + // Servable, but every candidate is blocked: self-clearing, and the caller + // requeues for the expiry. + name: "all candidates blocked", + pool: "pool", + want: metrics.DeferAllBlocked, + build: func() ([]client.Object, []*fakeProvider, Blocklister) { + pod := gatedPod("d1", "default", "uid-d1", "pool", "H100") + pool := poolWith("pool", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, "p1") + return []client.Object{pod, pool}, []*fakeProvider{{name: "p1"}}, + &fakeBlocklist{blocked: []failover.Candidate{{Provider: "p1"}}} + }, + }, { + // No provider in the pool offers the accelerator: no TTL will ever fix this. + name: "no servable candidate", + pool: "pool", + want: metrics.DeferNoCandidate, + build: func() ([]client.Object, []*fakeProvider, Blocklister) { + pod := gatedPod("d1", "default", "uid-d1", "pool", "H100") + pool := poolWith("pool", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, "p1") + return []client.Object{pod, pool}, []*fakeProvider{{name: "p1", gpus: []string{"A100"}}}, nil + }, + }, { + // A claim from a prior same-named Pod has not been reaped yet. + name: "stale claim", + pool: "pool", + want: metrics.DeferStaleClaim, + build: func() ([]client.Object, []*fakeProvider, Blocklister) { + pod := gatedPod("d1", "default", "uid-new", "pool", "H100") + pool := poolWith("pool", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, "p1") + stale := &nebulav1alpha1.NodeClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "default-d1"}, + Spec: nebulav1alpha1.NodeClaimSpec{ + PodRef: nebulav1alpha1.PodReference{Namespace: "default", Name: "d1", UID: "uid-old"}, + }, + } + return []client.Object{pod, pool, stale}, []*fakeProvider{{name: "p1"}}, nil + }, + }} + + // Every reason value, so each case can assert the others did NOT move. + all := []string{ + metrics.DeferNoPool, metrics.DeferInvalidRequest, metrics.DeferAllBlocked, + metrics.DeferNoCandidate, metrics.DeferStaleClaim, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + objs, provs, bl := tt.build() + before := map[string]float64{} + for _, reason := range all { + before[reason] = counterVal(t, metrics.PlacementDeferrals, + prometheus.Labels{"pool": tt.pool, "reason": reason}) + } + + r, _ := newPlacementReconciler(t, objs, provs...) + r.Blocklist = bl + reconcilePod(t, r, "default", "d1") + + for _, reason := range all { + want := 0.0 + if reason == tt.want { + want = 1 + } + got := counterVal(t, metrics.PlacementDeferrals, + prometheus.Labels{"pool": tt.pool, "reason": reason}) - before[reason] + if got != want { + t.Fatalf("deferrals{pool=%q,reason=%q} delta = %v, want %v", tt.pool, reason, got, want) + } + } + }) + } +} + +// The skip counter is the only view into WHY the walk passed over a candidate, and each +// reason is a different fix: register the provider, drop the tier from the pool, or pick +// an accelerator the provider offers. One reconcile can file several — the walk visits +// every candidate before giving up. +func TestPlacement_CandidateSkipReasons(t *testing.T) { + unregistered := skipLabels("ghost", nebulav1alpha1.CapacitySpot, "", metrics.SkipProviderUnregistered) + tierUnsupported := skipLabels("p1", nebulav1alpha1.CapacitySpot, "", metrics.SkipCapacityUnsupported) + accelUnsupported := skipLabels("p2", nebulav1alpha1.CapacitySpot, "", metrics.SkipAcceleratorUnsupported) + before := map[string]float64{ + "unregistered": counterVal(t, metrics.CandidateSkips, unregistered), + "tier": counterVal(t, metrics.CandidateSkips, tierUnsupported), + "accel": counterVal(t, metrics.CandidateSkips, accelUnsupported), + } + + // A Spot-only pool over three providers: one unregistered, one with no Spot tier, + // one that has Spot but not this accelerator. Nothing is placeable, and each + // candidate is skipped for its own reason. + pod := gatedPod("s1", "default", "uid-s1", "pool", "H100") + pool := poolWith("pool", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacitySpot}, "ghost", "p1", "p2") + r, _ := newPlacementReconciler(t, []client.Object{pod, pool}, + &fakeProvider{name: "p1", spot: false}, + &fakeProvider{name: "p2", spot: true, gpus: []string{"A100"}}, + ) + reconcilePod(t, r, "default", "s1") + + if got := counterVal(t, metrics.CandidateSkips, unregistered) - before["unregistered"]; got != 1 { + t.Fatalf("provider_unregistered skips delta = %v, want 1", got) + } + if got := counterVal(t, metrics.CandidateSkips, tierUnsupported) - before["tier"]; got != 1 { + t.Fatalf("capacity_type_unsupported skips delta = %v, want 1", got) + } + if got := counterVal(t, metrics.CandidateSkips, accelUnsupported) - before["accel"]; got != 1 { + t.Fatalf("accelerator_unsupported skips delta = %v, want 1", got) + } +} + +// The pool a Pod asks for is a Pod LABEL: user-controlled and unbounded. An unresolvable +// one must never reach a metric label, or a mislabeled workload could mint a time series +// per typo and blow up the registry. +func TestPlacement_UnresolvedPoolNameNeverBecomesALabel(t *testing.T) { + ghost := prometheus.Labels{"pool": "ghost-pool", "reason": metrics.DeferNoPool} + before := counterVal(t, metrics.PlacementDeferrals, ghost) + + pod := gatedPod("d2", "default", "uid-d2", "ghost-pool", "H100") + r, _ := newPlacementReconciler(t, []client.Object{pod}, &fakeProvider{name: "p1"}) + reconcilePod(t, r, "default", "d2") + + if got := counterVal(t, metrics.PlacementDeferrals, ghost) - before; got != 0 { + t.Fatalf("deferrals{pool=%q} delta = %v, want 0 (the name must not be filed)", "ghost-pool", got) + } +} + +// A placed Pod is counted on the candidate it actually landed on, and its wait is +// measured from POD CREATION — not from the reconcile that placed it. The distinction is +// the whole value of the metric: a Pod that sat gated for two minutes waiting out a +// failover block must report two minutes, not the microseconds the final pass took. +func TestPlacement_RecordsDecisionAndWaitFromPodCreation(t *testing.T) { + landed := decisionLabels("p1", nebulav1alpha1.CapacityOnDemand, "", "H100") + beforeCount := counterVal(t, metrics.PlacementDecisions, landed) + beforeSamples, beforeSum := histStats(t, metrics.PlacementWaitDuration, landed) + + pod := gatedPod("m1", "default", "uid-m1", "pool", "H100") + pod.CreationTimestamp = metav1.NewTime(time.Now().Add(-2 * time.Minute)) + pool := poolWith("pool", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, "p1") + r, _ := newPlacementReconciler(t, []client.Object{pod, pool}, &fakeProvider{name: "p1"}) + reconcilePod(t, r, "default", "m1") + + if got := counterVal(t, metrics.PlacementDecisions, landed) - beforeCount; got != 1 { + t.Fatalf("decisions delta = %v, want 1", got) + } + samples, sum := histStats(t, metrics.PlacementWaitDuration, landed) + if got := samples - beforeSamples; got != 1 { + t.Fatalf("wait observations delta = %d, want 1", got) + } + // Two minutes of gated time, minus nothing: anything near zero means the clock + // started at the reconcile instead of at the Pod. + if got := sum - beforeSum; got < 100 { + t.Fatalf("observed wait = %vs, want >= 100s (measured from Pod creation)", got) + } +} + +// The tier a Pod LANDS on is the cost question, so the decision must be filed against +// the fallback actually used — not the tier the pool asked for first. A fleet sliding +// from Spot to OnDemand is a cost regression with no error anywhere; this series and the +// skip counter that explains it are the only signals. +func TestPlacement_DecisionRecordsTheFallbackTierActuallyUsed(t *testing.T) { + spot := decisionLabels("p1", nebulav1alpha1.CapacitySpot, "", "H100") + onDemand := decisionLabels("p1", nebulav1alpha1.CapacityOnDemand, "", "H100") + blockedSkip := skipLabels("p1", nebulav1alpha1.CapacitySpot, "", metrics.SkipBlocked) + beforeSpot := counterVal(t, metrics.PlacementDecisions, spot) + beforeOnDemand := counterVal(t, metrics.PlacementDecisions, onDemand) + beforeSkip := counterVal(t, metrics.CandidateSkips, blockedSkip) + + pod := gatedPod("m2", "default", "uid-m2", "pool", "H100") + pool := poolWith("pool", []nebulav1alpha1.CapacityType{ + nebulav1alpha1.CapacitySpot, nebulav1alpha1.CapacityOnDemand, + }, "p1") + r, _ := newPlacementReconciler(t, []client.Object{pod, pool}, &fakeProvider{name: "p1", spot: true}) + r.Blocklist = &fakeBlocklist{blocked: []failover.Candidate{ + {Provider: "p1", CapacityType: nebulav1alpha1.CapacitySpot}, + }} + reconcilePod(t, r, "default", "m2") + + if got := counterVal(t, metrics.PlacementDecisions, onDemand) - beforeOnDemand; got != 1 { + t.Fatalf("OnDemand decisions delta = %v, want 1", got) + } + if got := counterVal(t, metrics.PlacementDecisions, spot) - beforeSpot; got != 0 { + t.Fatalf("Spot decisions delta = %v, want 0 (Spot was blocked)", got) + } + // And the skip counter is what explains the fallback after the fact. + if got := counterVal(t, metrics.CandidateSkips, blockedSkip) - beforeSkip; got != 1 { + t.Fatalf("blocked Spot skips delta = %v, want 1", got) + } +} diff --git a/internal/controller/pod_placement_controller.go b/internal/controller/pod_placement_controller.go index 6ca65ca..fd00ccb 100644 --- a/internal/controller/pod_placement_controller.go +++ b/internal/controller/pod_placement_controller.go @@ -28,6 +28,7 @@ import ( nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" "github.com/InftyAI/Nebula/pkg/failover" + "github.com/InftyAI/Nebula/pkg/metrics" "github.com/InftyAI/Nebula/pkg/provider" "github.com/InftyAI/Nebula/pkg/util" ) @@ -125,11 +126,15 @@ func (r *PodPlacementReconciler) Reconcile(ctx context.Context, req ctrl.Request } if pool == nil { // No pool to place against. Leave the Pod gated rather than guessing; an - // operator sees a SchedulingGated Pod and a missing/mislabeled pool. + // operator sees a SchedulingGated Pod and a missing/mislabeled pool. The pool + // label is deliberately NOT filed on the metric — see metrics.RecordDeferral. + metrics.RecordDeferral("", metrics.DeferNoPool) log.Info("no NodePool resolved for Pod; leaving it gated", "pod", pod.Name) return ctrl.Result{}, nil } + // selectPlacement files its own deferral reason on the !ok paths — it is the only + // place that knows which of the three applies (see its doc). placement, ok, retryAfter := r.selectPlacement(ctx, &pod, pool) if !ok { // No provider in the pool can serve this Pod's GPU type right now. Leave it @@ -164,6 +169,7 @@ func (r *PodPlacementReconciler) Reconcile(ctx context.Context, req ctrl.Request // A stale claim for a prior same-named Pod still exists. Do NOT ungate // against the wrong ledger; wait for the NodeClaim backstop to reap it, // then a requeue re-creates the claim with this Pod's UID. + metrics.RecordDeferral(pool.Name, metrics.DeferStaleClaim) log.Info("waiting for a stale NodeClaim to be reclaimed before placing", "pod", pod.Name, "claim", util.ClaimName(pod.Namespace, pod.Name)) return ctrl.Result{RequeueAfter: staleClaimRequeue}, nil @@ -173,6 +179,11 @@ func (r *PodPlacementReconciler) Reconcile(ctx context.Context, req ctrl.Request if err := r.place(ctx, &pod, pool, placement); err != nil { return ctrl.Result{}, err } + // Counted only after the Update lands, which is also what keeps it from + // double-counting: a reconcile driven by a stale cache would see the gate still + // present, redo the walk, and then fail this Update on a resourceVersion conflict — + // returning above rather than reaching here. + metrics.ObservePlacement(placementLabels(&pod, placement), time.Since(pod.CreationTimestamp.Time)) log.Info("placed Pod", "pod", pod.Name, "provider", placement.provider, "capacityType", placement.capacityType, "region", placement.region) return ctrl.Result{}, nil diff --git a/internal/controller/pod_placement_helpers.go b/internal/controller/pod_placement_helpers.go index 8684d9d..d85dbfa 100644 --- a/internal/controller/pod_placement_helpers.go +++ b/internal/controller/pod_placement_helpers.go @@ -33,6 +33,7 @@ import ( nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" "github.com/InftyAI/Nebula/pkg/failover" + "github.com/InftyAI/Nebula/pkg/metrics" "github.com/InftyAI/Nebula/pkg/provider" "github.com/InftyAI/Nebula/pkg/util" ) @@ -86,6 +87,12 @@ func (r *PodPlacementReconciler) poolFor(ctx context.Context, pod *corev1.Pod) ( // expires, rather than idling until the periodic resync — blocklist TTL expiry // emits no event of its own. // +// It also owns the placement metrics for everything it decides: a skip counter per +// candidate passed over, and — on ok=false — the deferral reason, because this is the +// only scope that can tell an invalid request from an all-blocked pool from a pool that +// simply cannot serve the request. The caller files the two deferrals it owns instead +// (no resolvable pool, and a stale NodeClaim). +// // Strategy (LowestPrice/Weighted price-ranking) is a later swap-in for the inner // ordering; today the inner walk is listed order (Ordered), which the caller's // flow does not depend on — only that a (provider, capacityType, region) comes @@ -102,6 +109,7 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev // source object) before placement can proceed. accel, count, err := util.AcceleratorRequest(pod) if err != nil { + metrics.RecordDeferral(pool.Name, metrics.DeferInvalidRequest) log.Info("invalid accelerator request; leaving Pod gated", "error", err.Error()) return placement{}, false, 0 } @@ -111,11 +119,15 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev for _, ref := range pool.Spec.Providers { // provider (Ordered = listed order) prov, ok := r.provider(ref.Name) if !ok { + // No region on this skip and the two below: they are decided before the + // walk reaches the region axis, so they rule out every region at once. + metrics.RecordCandidateSkip(ref.Name, tier, "", metrics.SkipProviderUnregistered) log.V(1).Info("skipping candidate: provider not registered", "provider", ref.Name, "capacityType", tier) continue // unregistered; NodePool status surfaces this separately } if !servesCapacity(prov, tier) { + metrics.RecordCandidateSkip(ref.Name, tier, "", metrics.SkipCapacityUnsupported) log.V(1).Info("skipping candidate: provider does not offer the capacity tier", "provider", ref.Name, "capacityType", tier) continue @@ -129,6 +141,7 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev accelerator := util.AcceleratorPool(accel, count) if accel != "" { if _, offered := prov.MapAccelerator(accel, count); !offered { + metrics.RecordCandidateSkip(ref.Name, tier, "", metrics.SkipAcceleratorUnsupported) log.V(1).Info("skipping candidate: provider does not offer the accelerator", "provider", ref.Name, "accelerator", accel, "count", count) continue @@ -138,6 +151,7 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev if until, blocked := r.blockedUntil(ref.Name, accelerator, tier, region); blocked { // Servable but failed recently; try the next region, then the next // tier, and remember when this one frees so we can requeue for it. + metrics.RecordCandidateSkip(ref.Name, tier, region, metrics.SkipBlocked) log.Info("skipping candidate: blocked by failover blocklist", "provider", ref.Name, "accelerator", accelerator, "capacityType", tier, "region", region, "freesIn", until.String()) @@ -157,9 +171,39 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev } } } + // The walk is exhausted. Which deferral this is turns on whether anything was + // merely blocked: a positive soonest means a servable candidate exists and failover + // is holding it off (self-clearing, and the caller requeues for it), while zero + // means nothing in the pool can serve this request at all. + if soonest > 0 { + metrics.RecordDeferral(pool.Name, metrics.DeferAllBlocked) + } else { + metrics.RecordDeferral(pool.Name, metrics.DeferNoCandidate) + } return placement{}, false, soonest } +// placementLabels renders the metric label set for a completed placement. It mirrors +// the virtual kubelet's Handler.metricLabels field for field — that symmetry is the +// point, not a coincidence: the handler reads region and capacity type back off the +// annotations that place() stamps from this very decision, so both halves of a Pod's +// journey report identical label values and their series join cleanly. +// +// The accelerator type and count are passed apart, not as the joined pool identity +// placement.accelerator carries, because a metric label must be aggregatable (see +// metrics.provisionLabels). The parse cannot fail here: selectPlacement already +// rejected a malformed request before any placement was returned. +func placementLabels(pod *corev1.Pod, p placement) metrics.Labels { + accel, count, _ := util.AcceleratorRequest(pod) + return metrics.Labels{ + Provider: p.provider, + Region: p.region, + CapacityType: string(p.capacityType), + Accelerator: accel, + AcceleratorCount: count, + } +} + // capacityTiers is the outer axis to walk: the pool's CapacityTypes in fallback // order. An empty list means "the provider default tier" — a single unnamed // candidate ("") so the walk still runs once. (Admission defaults the field, so diff --git a/pkg/metrics/helper_test.go b/pkg/metrics/helper_test.go new file mode 100644 index 0000000..5d40a59 --- /dev/null +++ b/pkg/metrics/helper_test.go @@ -0,0 +1,32 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +package metrics + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// counterValue reads one counter series. The collectors are package-level and shared by +// every test here, so assertions are on the DELTA a call produced rather than an +// absolute value — otherwise specs would pass or fail by run order. +func counterValue(t *testing.T, c prometheus.Counter) float64 { + t.Helper() + return testutil.ToFloat64(c) +} diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go new file mode 100644 index 0000000..ff06f77 --- /dev/null +++ b/pkg/metrics/metrics.go @@ -0,0 +1,269 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +// Package metrics holds Nebula's Prometheus instrumentation. +// +// Everything here registers into controller-runtime's registry, so it is served on +// the manager's existing --metrics-bind-address endpoint alongside the standard +// controller/workqueue metrics. Importing this package is what registers the +// collectors (see init); no wiring is needed in main. +// +// The instrumented surface is the path a Pod takes from admission to a running +// external instance: PLACEMENT (this file's siblings in placement.go) and +// PROVISIONING (here). Those are the parts whose cost and failure modes are +// otherwise invisible — placement can silently leave a Pod gated forever, and +// provisioning runs against a third party, takes seconds to minutes, bills money, +// and fails for reasons the Pod status flattens away. Everything else (reconcile +// counts, queue depth, API latency) is already covered by controller-runtime's own +// metrics, and Pod-population questions by kube-state-metrics. +// +// The two halves deliberately share one label set (Labels, provisionLabels) so a +// placement and the provisioning attempt it led to carry identical label values and +// can be joined in PromQL without label surgery. +package metrics + +import ( + "context" + "errors" + "strconv" + "time" + + "github.com/prometheus/client_golang/prometheus" + ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + + "github.com/InftyAI/Nebula/pkg/provider" +) + +// Result values for the result label. A provisioning attempt either returned an +// instance id or an error; there is no third outcome. +const ( + ResultSuccess = "success" + ResultFailure = "failure" +) + +// Reason values for the failure label. This is a deliberately COARSE, closed set: +// it is a metric label, so it must stay bounded no matter what text a provider API +// returns. The fine-grained detail stays where it is already available (the Pod's +// Failed status message and the vnode-handler error log); this exists to answer +// "are we losing capacity, or are our credentials broken?" at a glance. +const ( + ReasonCapacity = "capacity" + ReasonQuota = "quota" + ReasonAuth = "auth" + ReasonUnsupported = "unsupported_accelerator" + ReasonTimeout = "timeout" + // ReasonUnreachable: the provider never told us what it decided — a transport + // failure, a 503, an unparseable response. Kept separate from every other reason + // because it is the one that is NOT about capacity, credentials or the request: it + // says the integration itself is unhealthy, and it is the only reason for which + // Nebula deliberately does not fail the Pod or blocklist the candidate (see + // provider.IsRejection). A spike here alongside flat capacity/auth series is a + // network or provider-outage signal, not a placement one. + ReasonUnreachable = "unreachable" + // ReasonOther: the provider DID reject the request, but not through a sentinel, so + // the category is unavailable. In practice that means the adapter returned a raw API + // error without wrapping it, which makes a sustained rate on this series a to-do + // rather than an incident: wrap the condition in the adapter (see + // docs/add-a-provider.md) and the failure moves onto its real category. + ReasonOther = "other" +) + +// none is the placeholder for a label the request genuinely did not carry: no +// region (unconstrained), no capacity tier (the provider's default), or no +// accelerator (a CPU-only Pod). An explicit token beats an empty string, which in +// PromQL is indistinguishable from a label that was never set and silently matches +// `{region=""}` selectors an operator did not mean to write. +const none = "none" + +// provisionLabels is the label set every provisioning metric carries, in order. +// Region is included because a capacity shortfall is region-local and comparing +// regions is the whole point of collecting this; note that for a provider which +// collapses several declared regions into one candidate (Modal) the value is that +// provider's joined token, not a single region name — see NodeClaimSpec.Region. +// +// The accelerator TYPE and COUNT are two labels, not the joined "H100:8" pool identity +// used as the blocklist key. A metric label set is meant to be aggregated over, and a +// joined string cannot be: `sum by (accelerator)` over every size of H100 requires +// splitting the value in PromQL, and selecting all 8-GPU requests across types is not +// expressible at all. Two labels give both for free, and the pool key is still +// recoverable as accelerator + ":" + accelerator_count when correlating with a +// blocklist entry. +var provisionLabels = []string{"provider", "region", "capacity_type", "accelerator", "accelerator_count"} + +var ( + // ProvisionAttempts counts provisioning attempts by outcome. Rate of the + // result="failure" series over the total is the provisioning error rate; the + // per-region/accelerator breakdown is what tells you WHERE it is failing. + ProvisionAttempts = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "nebula_provision_attempts_total", + Help: "Total external instance provisioning attempts, by provider, region, capacity type, accelerator type, accelerator count and outcome.", + }, append(append([]string{}, provisionLabels...), "result")) + + // ProvisionFailures breaks failures down by coarse cause. It deliberately + // overlaps ProvisionAttempts{result="failure"} rather than adding a reason label + // there: reason is only meaningful on failure, and carrying it on the attempts + // counter would multiply the success series by a label that is constant for + // them. + ProvisionFailures = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "nebula_provision_failures_total", + Help: "Failed provisioning attempts by coarse cause (capacity, quota, auth, unsupported_accelerator, timeout, other).", + }, append(append([]string{}, provisionLabels...), "reason")) + + // ProvisionDuration measures the provider's Provision call alone — not the + // wait for the instance to become usable. The two differ enormously and for + // different reasons: AWS sweeps a region's availability zones inside this call + // (so a capacity shortage shows up as latency HERE), while Modal returns as soon + // as the sandbox is accepted and the wait moves to InstanceReadyDuration. + // Bucketed out to 300s because the call is bounded by + // Capabilities.ProvisionTimeout, which AWS raises above the 90s default. + ProvisionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "nebula_provision_duration_seconds", + Help: "Latency of the provider's Provision call, by provider, region, capacity type, accelerator type, accelerator count and outcome.", + Buckets: []float64{0.5, 1, 2.5, 5, 10, 20, 30, 45, 60, 90, 120, 180, 300}, + }, append(append([]string{}, provisionLabels...), "result")) + + // InstanceReadyDuration measures the whole user-visible wait: from the moment + // CreatePod starts provisioning to the first poll tick that reports the instance + // Running. It therefore includes the Provision call, any provider-side queueing + // for capacity, image pull, GPU attach, container boot, and up to one poll + // interval of detection lag — which is the honest number, because that is what a + // user waits. + // + // Only observed ONCE per Pod, on the first transition to Running, and never for + // an instance re-adopted after a restart (the original start time is gone, and a + // duration measured from re-adoption would understate it wildly). Buckets run to + // 30min because a queueing provider on a large GPU shape genuinely takes that + // long. + InstanceReadyDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "nebula_instance_ready_duration_seconds", + Help: "Time from the start of provisioning to the instance first reporting Running, by provider, region, capacity type, accelerator type and accelerator count.", + Buckets: []float64{5, 10, 20, 30, 45, 60, 90, 120, 180, 300, 600, 900, 1800}, + }, provisionLabels) +) + +func init() { + ctrlmetrics.Registry.MustRegister( + ProvisionAttempts, + ProvisionFailures, + ProvisionDuration, + InstanceReadyDuration, + ) +} + +// Labels identifies the placement one provisioning attempt was made against. The +// zero value is valid: every field normalizes to "none". +type Labels struct { + Provider string + Region string + CapacityType string + // Accelerator is the accelerator TYPE alone (e.g. "H100"), and AcceleratorCount how + // many were requested. They are kept apart so both aggregations work — see + // provisionLabels. Empty/zero for a CPU-only Pod. + Accelerator string + AcceleratorCount int32 +} + +// values renders the label set in provisionLabels order, with extra appended for +// the metrics that carry a result/reason dimension. +func (l Labels) values(extra ...string) []string { + return append([]string{ + orNone(l.Provider), + orNone(l.Region), + orNone(l.CapacityType), + orNone(l.Accelerator), + countOrNone(l.AcceleratorCount), + }, extra...) +} + +func orNone(s string) string { + if s == "" { + return none + } + return s +} + +// countOrNone renders an accelerator count, or the placeholder when there is no +// accelerator to count. Zero is deliberately NOT rendered as "0": a CPU-only Pod did not +// request zero GPUs, it requested none at all, and "0" would put it in the same numeric +// series an operator reads as a real count. util.AcceleratorRequest never returns a +// positive count without a type, so this cannot mask a real request. +func countOrNone(n int32) string { + if n <= 0 { + return none + } + return strconv.FormatInt(int64(n), 10) +} + +// ObserveProvision records one completed provisioning attempt: its outcome, its +// latency, and — when it failed — the coarse cause. It is the single call the +// virtual kubelet makes on both the success and failure paths, so the attempt and +// failure counters cannot drift out of step. +// +// err nil means success. d is the duration of the Provision call itself. +func ObserveProvision(l Labels, d time.Duration, err error) { + result := ResultSuccess + if err != nil { + result = ResultFailure + } + ProvisionAttempts.WithLabelValues(l.values(result)...).Inc() + ProvisionDuration.WithLabelValues(l.values(result)...).Observe(d.Seconds()) + if err != nil { + ProvisionFailures.WithLabelValues(l.values(FailureReason(err))...).Inc() + } +} + +// ObserveReady records the end-to-end wait for an instance to reach Running. +func ObserveReady(l Labels, d time.Duration) { + InstanceReadyDuration.WithLabelValues(l.values()...).Observe(d.Seconds()) +} + +// FailureReason maps a Provision error onto the closed reason set. +// +// It matches on the shared sentinels in pkg/provider rather than on message text, +// which is what keeps the label bounded: an adapter that wraps its API errors with +// ErrNoCapacity/ErrQuota/ErrAuth gets an accurate reason, and one that does not +// falls through to "other" instead of inventing a new series per distinct provider +// message. This is intentionally NOT provider.ClassifyError: that answers a +// different question (what to blocklist, and how widely), and its "unrecognized +// errors are scoped like capacity" default would be an outright lie as a metric — +// it would report unknown failures as capacity shortfalls. +func FailureReason(err error) string { + switch { + case err == nil: + return "" + case errors.Is(err, provider.ErrAuth): + return ReasonAuth + case errors.Is(err, provider.ErrQuota): + return ReasonQuota + case errors.Is(err, provider.ErrUnsupportedAccelerator): + return ReasonUnsupported + case errors.Is(err, provider.ErrNoCapacity): + return ReasonCapacity + // Checked after the sentinels: a provider that hits its own deadline while + // sweeping for capacity may wrap both, and the capacity cause is the more useful + // of the two. + case errors.Is(err, context.DeadlineExceeded): + return ReasonTimeout + // Everything left is either a rejection whose category we could not name, or a + // failure to reach the provider at all. Splitting them is the whole point of having + // this label: the first is a placement problem, the second an integration one, and + // they are fixed by completely different people. + case !provider.IsRejection(err): + return ReasonUnreachable + default: + return ReasonOther + } +} diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go new file mode 100644 index 0000000..38b865e --- /dev/null +++ b/pkg/metrics/metrics_test.go @@ -0,0 +1,141 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +package metrics + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/InftyAI/Nebula/pkg/provider" +) + +// The reason label is a CLOSED set, so this pins every mapping into it. The label is +// what an operator reads to decide whether a provisioning problem is theirs (quota, +// credentials), the provider's (capacity, unreachable) or neither. +func TestFailureReason(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + {"nil has no reason", nil, ""}, + {"auth", provider.ErrAuth, ReasonAuth}, + {"quota", provider.ErrQuota, ReasonQuota}, + {"unsupported", provider.ErrUnsupportedAccelerator, ReasonUnsupported}, + {"capacity", provider.ErrNoCapacity, ReasonCapacity}, + {"wrapped capacity", fmt.Errorf("create sandbox: %w", provider.ErrNoCapacity), ReasonCapacity}, + + // A provider that hits its own deadline WHILE sweeping for capacity wraps both; + // the capacity cause is the more useful of the two, so the sentinels come first. + {"capacity beats timeout", fmt.Errorf("%w: %w", provider.ErrNoCapacity, context.DeadlineExceeded), ReasonCapacity}, + {"bare timeout", context.DeadlineExceeded, ReasonTimeout}, + + // The split that matters operationally: an integration outage must not read as a + // capacity shortfall, or the failure series points at the wrong problem entirely. + {"grpc transport", errors.New("rpc error: code = Unavailable desc = transport is closing"), ReasonUnreachable}, + {"connection refused", errors.New("dial tcp: connect: connection refused"), ReasonUnreachable}, + {"unrecognized", errors.New("weird transient blip"), ReasonUnreachable}, + + // An unwrapped provider message is recognized as a REJECTION (so not + // "unreachable"), but its category is unavailable here: FailureReason matches + // sentinels only, on purpose, rather than re-running the string heuristics. So it + // lands on "other", which is precisely the signal that an adapter is not wrapping + // its errors — actionable, unlike a guess at the category. + {"unwrapped rejection", errors.New("InsufficientInstanceCapacity"), ReasonOther}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := FailureReason(tt.err); got != tt.want { + t.Fatalf("FailureReason(%v) = %q, want %q", tt.err, got, tt.want) + } + }) + } +} + +// ObserveProvision is the single call site for both outcomes, so the attempt and +// failure counters can never drift: a failure must increment both, a success only the +// attempt counter. +func TestObserveProvision_CountersStayInStep(t *testing.T) { + l := Labels{Provider: "p", Region: "r", CapacityType: "OnDemand", Accelerator: "H100", AcceleratorCount: 1} + + failureBefore := counterValue(t, ProvisionAttempts.WithLabelValues(l.values(ResultFailure)...)) + reasonBefore := counterValue(t, ProvisionFailures.WithLabelValues(l.values(ReasonCapacity)...)) + successBefore := counterValue(t, ProvisionAttempts.WithLabelValues(l.values(ResultSuccess)...)) + + ObserveProvision(l, 0, provider.ErrNoCapacity) + ObserveProvision(l, 0, nil) + + if got := counterValue(t, ProvisionAttempts.WithLabelValues(l.values(ResultFailure)...)) - failureBefore; got != 1 { + t.Fatalf("failure attempts delta = %v, want 1", got) + } + if got := counterValue(t, ProvisionFailures.WithLabelValues(l.values(ReasonCapacity)...)) - reasonBefore; got != 1 { + t.Fatalf("capacity failures delta = %v, want 1", got) + } + if got := counterValue(t, ProvisionAttempts.WithLabelValues(l.values(ResultSuccess)...)) - successBefore; got != 1 { + t.Fatalf("success attempts delta = %v, want 1", got) + } + // A success must never touch the failure-reason counter, whatever the reason. + for _, reason := range []string{ReasonCapacity, ReasonAuth, ReasonQuota, ReasonUnsupported, ReasonTimeout, ReasonUnreachable, ReasonOther} { + if reason == ReasonCapacity { + continue // asserted above + } + if got := counterValue(t, ProvisionFailures.WithLabelValues(l.values(reason)...)); got != 0 { + t.Fatalf("reason %q counted %v times, want 0", reason, got) + } + } +} + +// An unset field renders as the explicit "none" placeholder, never "" — which in PromQL +// is indistinguishable from a label that was never set and silently matches {region=""} +// selectors an operator did not mean to write. The accelerator COUNT is held to the same +// rule: a CPU-only Pod did not request zero GPUs, it requested none, and rendering "0" +// would drop it into the numeric series an operator reads as real counts. +func TestLabels_RendersInProvisionLabelOrder(t *testing.T) { + tests := []struct { + name string + in Labels + want []string + }{ + {"zero value is all placeholders", Labels{}, []string{none, none, none, none, none}}, + { + "cpu-only pod has no count, not a zero count", + Labels{Provider: "p", Region: "r", CapacityType: "OnDemand"}, + []string{"p", "r", "OnDemand", none, none}, + }, + { + "accelerator type and count are separate values", + Labels{Provider: "p", Region: "r", CapacityType: "Spot", Accelerator: "H100", AcceleratorCount: 8}, + []string{"p", "r", "Spot", "H100", "8"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := tt.in.values() + if len(got) != len(provisionLabels) { + t.Fatalf("values() = %v (%d values), want %d to match provisionLabels %v", + got, len(got), len(provisionLabels), provisionLabels) + } + for i := range tt.want { + if got[i] != tt.want[i] { + t.Fatalf("values()[%d] (%s) = %q, want %q", i, provisionLabels[i], got[i], tt.want[i]) + } + } + }) + } +} diff --git a/pkg/metrics/placement.go b/pkg/metrics/placement.go new file mode 100644 index 0000000..28b0560 --- /dev/null +++ b/pkg/metrics/placement.go @@ -0,0 +1,145 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +package metrics + +import ( + "time" + + "github.com/prometheus/client_golang/prometheus" + ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" + + nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" +) + +// Reason values for the placement deferral label: why one reconcile ended without +// placing the Pod. Closed set, and each value points at a DIFFERENT owner — which is +// the whole reason for splitting them: +// +// - no_pool / invalid_request: the request is wrong. Nobody is retrying their way out +// of these; a human must edit the Pod (or the workload that generates it). +// - all_blocked: the request is fine and a candidate exists, but failover is holding +// it off. Self-clearing, and the Pod is already requeued for the moment it frees. +// - no_candidate: the pool cannot serve this request at all today. Self-clearing only +// if an operator adds a provider or a provider registers. +// - stale_claim: a NodeClaim from a prior same-named Pod has not been reaped yet. +// Self-clearing in seconds; a sustained rate means the NodeClaim backstop is stuck. +const ( + DeferNoPool = "no_pool" + DeferInvalidRequest = "invalid_request" + DeferAllBlocked = "all_blocked" + DeferNoCandidate = "no_candidate" + DeferStaleClaim = "stale_claim" +) + +// Reason values for the candidate skip label: why the placement walk passed over one +// (tier, provider, region) candidate. Only "blocked" clears on its own. +const ( + SkipProviderUnregistered = "provider_unregistered" + SkipCapacityUnsupported = "capacity_type_unsupported" + SkipAcceleratorUnsupported = "accelerator_unsupported" + SkipBlocked = "blocked" +) + +var ( + // PlacementDecisions counts Pods actually placed, by the candidate they landed on. + // It carries the same labels as the provisioning metrics on purpose: the two are + // joinable without label surgery, so "placed on Spot but never provisioned" is one + // query rather than a correlation exercise. The tier breakdown is the money + // question — a fleet quietly sliding from Spot to OnDemand is a cost regression + // with no error anywhere. + PlacementDecisions = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "nebula_placement_decisions_total", + Help: "Pods placed, by provider, region, capacity type, accelerator type and accelerator count.", + }, provisionLabels) + + // PlacementWaitDuration measures from Pod creation to the gate being removed: the + // user-visible queue time BEFORE provisioning starts, which + // nebula_instance_ready_duration_seconds then continues from. Together they cover + // the whole path from `kubectl apply` to a Running instance. + // + // Unlike the ready duration, this one has no restart gap: the start timestamp is + // the Pod's own creationTimestamp, so a placement that happens after a manager + // restart still reports the true total wait. Buckets span three orders of + // magnitude because the honest range does: an unblocked Pod is placed in + // milliseconds, while one waiting out a failover block waits the blocklist TTL. + PlacementWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "nebula_placement_wait_duration_seconds", + Help: "Time from Pod creation to placement (scheduling gate removal), by provider, region, capacity type, accelerator type and accelerator count.", + Buckets: []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 1800}, + }, provisionLabels) + + // PlacementDeferrals counts reconciles that ended without placing the Pod, by + // reason. + // + // It counts DEFERRALS, not Pods: a gated Pod is reconciled again on every requeue + // and resync, so one Pod stuck for an hour contributes many increments. That makes + // the rate a measure of placement pressure, not a population count — for "how many + // Pods are stuck right now", read the SchedulingGated Pod count from + // kube-state-metrics and use this series to explain WHY. + PlacementDeferrals = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "nebula_placement_deferrals_total", + Help: "Reconciles that ended without placing a Pod, by pool and reason (no_pool, invalid_request, all_blocked, no_candidate, stale_claim).", + }, []string{"pool", "reason"}) + + // CandidateSkips counts individual (tier, provider, region) candidates passed over + // during the placement walk. This is the only view into failover actually working: + // when every Pod lands on OnDemand, a rate on {capacity_type="Spot", + // reason="blocked"} is the explanation, and one on + // reason="capacity_type_unsupported" says the pool is misconfigured instead. + // + // Cardinality is bounded by pool configuration (providers x tiers x regions x four + // reasons), not by Pod count. + CandidateSkips = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "nebula_placement_candidate_skips_total", + Help: "Placement candidates skipped, by provider, capacity type, region and reason.", + }, []string{"provider", "capacity_type", "region", "reason"}) +) + +func init() { + ctrlmetrics.Registry.MustRegister( + PlacementDecisions, + PlacementWaitDuration, + PlacementDeferrals, + CandidateSkips, + ) +} + +// ObservePlacement records one Pod placed onto the candidate l describes, having +// waited waited since it was created. +func ObservePlacement(l Labels, waited time.Duration) { + PlacementDecisions.WithLabelValues(l.values()...).Inc() + PlacementWaitDuration.WithLabelValues(l.values()...).Observe(waited.Seconds()) +} + +// RecordDeferral records one reconcile that placed nothing. +// +// pool MUST be the name of a NodePool that actually exists, or "" for the deferral +// where it does not (DeferNoPool). The pool a Pod asks for is a Pod LABEL — user +// controlled and unbounded — so filing the unresolved string here would let a +// mislabeled workload mint a new time series per typo. Once the pool has been +// resolved to a real object, its name is bounded by cluster resources and safe. +func RecordDeferral(pool, reason string) { + PlacementDeferrals.WithLabelValues(orNone(pool), reason).Inc() +} + +// RecordCandidateSkip records one candidate the placement walk passed over. region is +// empty for the skips decided before the walk reaches the region axis (an +// unregistered provider, an unservable tier, a missing accelerator), which is honest: +// those rule out every region at once. +func RecordCandidateSkip(prov string, tier nebulav1alpha1.CapacityType, region, reason string) { + CandidateSkips.WithLabelValues(orNone(prov), orNone(string(tier)), orNone(region), reason).Inc() +} diff --git a/pkg/provider/errors.go b/pkg/provider/errors.go index 38fa804..e944a66 100644 --- a/pkg/provider/errors.go +++ b/pkg/provider/errors.go @@ -17,6 +17,7 @@ limitations under the License. package provider import ( + "context" "errors" "strings" @@ -93,33 +94,105 @@ func ClassifyError(err error, capacityType nebulav1alpha1.CapacityType, accelera return s } + switch categorize(err) { + case catAuth: + return BlockScope{DenyAll: true} + case catCapacity: + return capacityScope() + default: + // An unrecognized error is scoped like capacity (this accelerator + tier, and + // per region once the adapter confines it), NOT DenyAll. A DenyAll on an + // unknown error fences off the WHOLE provider — every region and accelerator — + // which is far too broad a blast radius for a failure we can't even identify + // (e.g. a transient malformed-request blip in one region). Failover past the + // one failing candidate is the safer default; the TTL still bounds it. + // + // This answers "how widely, IF we block", not "should we block": a caller that + // cannot attribute the failure to the request at all should not be filing a block + // in the first place — see IsRejection. + return capacityScope() + } +} + +// failureCategory is the internal classification ClassifyError and IsRejection both +// drive off, so the two can never disagree about whether an error was recognized. +type failureCategory int + +const ( + // catUnattributable: nothing in the error says what the provider decided, because + // it may not have decided anything — a transport failure, a timeout, an + // unparseable API blip. Kept distinct from the scope ClassifyError ultimately + // returns for it. + catUnattributable failureCategory = iota + // catAuth: credentials or authorization failed, so nothing on the provider works. + catAuth + // catCapacity: the provider refused THIS request — no capacity, quota exhausted, + // or an accelerator it does not offer. + catCapacity +) + +// categorize buckets a provision error, sentinels first and string heuristics after. +func categorize(err error) failureCategory { + // Sentinels first. An adapter that wrapped one has made an explicit decision, and + // it outranks anything the raw message text happens to contain. switch { case errors.Is(err, ErrAuth): - return BlockScope{DenyAll: true} + return catAuth case errors.Is(err, ErrNoCapacity), errors.Is(err, ErrUnsupportedAccelerator), errors.Is(err, ErrQuota): - return capacityScope() + return catCapacity } - // Fall back to string heuristics for errors not wrapped with a sentinel. msg := strings.ToLower(err.Error()) + + // Transport and timeout markers are checked BEFORE the category heuristics, + // because they are the failures those heuristics most reliably MISREAD: a gRPC + // status renders as "rpc error: code = Unavailable desc = ...", whose + // "unavailable" would otherwise match the capacity bucket below and turn "we could + // not reach the provider" into "the provider has no capacity" — the exact + // misattribution IsRejection exists to prevent. A deadline is unattributable for a + // second reason too: our own ProvisionTimeout can fire on a call the provider went + // on to honour, so the instance may well exist. + switch { + case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled): + return catUnattributable + case containsAny(msg, + "rpc error", "connection refused", "connection reset", "broken pipe", + "no such host", "i/o timeout", "eof", "tls handshake", + "service unavailable", "bad gateway", "gateway timeout", "internal server error"): + return catUnattributable + } + switch { case containsAny(msg, "unauthorized", "forbidden", "authentication", "invalid token", "api key"): - return BlockScope{DenyAll: true} + return catAuth case containsAny(msg, "quota", "limit exceeded", "rate limit"): - return capacityScope() + return catCapacity case containsAny(msg, "no capacity", "capacity", "unavailable", "out of", "no gpu"): - return capacityScope() + return catCapacity default: - // An unrecognized error is scoped like capacity (this accelerator + tier, and - // per region once the adapter confines it), NOT DenyAll. A DenyAll on an - // unknown error fences off the WHOLE provider — every region and accelerator — - // which is far too broad a blast radius for a failure we can't even identify - // (e.g. a transient malformed-request blip in one region). Failover past the - // one failing candidate is the safer default; the TTL still bounds it. - return capacityScope() + return catUnattributable } } +// IsRejection reports whether err is a provider DECISION about this request — "no +// capacity", "over quota", "bad credentials", "I do not offer that accelerator" — as +// opposed to a failure to find out what the provider would have decided: a transport +// error, a timeout, a 503, an unparseable response. +// +// The distinction exists because the two call for opposite handling and the costs are +// asymmetric. A rejection is authoritative, so the Pod is failed and the candidate +// blocklisted, and failover routes around it — all correct, because the provider said +// no. An unattributable failure is authoritative about nothing, so the same writes +// stamp a terminal status onto a request the provider may have accepted (leaving a +// paid instance running behind a Pod that is about to be reaped) and fence off a +// candidate that never misbehaved. Retrying costs a request and Provision is +// idempotent on ClaimName; acting on a guess costs an instance. +// +// A nil error is not a rejection. +func IsRejection(err error) bool { + return err != nil && categorize(err) != catUnattributable +} + // containsAny reports whether s contains any of subs. func containsAny(s string, subs ...string) bool { for _, sub := range subs { diff --git a/pkg/provider/errors_test.go b/pkg/provider/errors_test.go index 44a993e..3727003 100644 --- a/pkg/provider/errors_test.go +++ b/pkg/provider/errors_test.go @@ -17,6 +17,8 @@ limitations under the License. package provider import ( + "context" + "errors" "fmt" "reflect" "testing" @@ -82,4 +84,66 @@ func TestClassifyError_EmptyAcceleratorStaysNil(t *testing.T) { } } +// IsRejection separates a provider DECISION about the request from a failure to +// learn what it would have decided. The vnode handler fails the Pod and blocklists +// the candidate only for the former, so a transport error misclassified as a +// rejection stamps a terminal status on a request that may have been accepted. +func TestIsRejection(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {"nil is not a rejection", nil, false}, + {"auth sentinel", ErrAuth, true}, + {"no-capacity sentinel", ErrNoCapacity, true}, + {"quota sentinel", ErrQuota, true}, + {"unsupported sentinel", ErrUnsupportedAccelerator, true}, + {"wrapped sentinel", fmt.Errorf("create sandbox: %w", ErrNoCapacity), true}, + {"string capacity", errors.New("InsufficientInstanceCapacity"), true}, + {"string auth", errors.New("HTTP 401 unauthorized"), true}, + + // The failures this predicate exists for. + {"deadline exceeded", context.DeadlineExceeded, false}, + {"wrapped deadline", fmt.Errorf("provision: %w", context.DeadlineExceeded), false}, + {"canceled", context.Canceled, false}, + {"connection refused", errors.New("dial tcp 10.0.0.1:443: connect: connection refused"), false}, + {"eof", errors.New("unexpected EOF"), false}, + {"http 503", errors.New("503 Service Unavailable"), false}, + {"unrecognized", errors.New("weird transient blip"), false}, + + // A gRPC status renders "Unavailable" in its text, which the capacity heuristic + // would otherwise match — this is the misread the transport check precedes it for. + {"grpc unavailable", errors.New("rpc error: code = Unavailable desc = transport is closing"), false}, + // ...but a sentinel the adapter wrapped still wins over the raw text, so an + // adapter that classified a gRPC error itself is not second-guessed. + {"grpc unavailable wrapping a sentinel", + fmt.Errorf("rpc error: code = Unavailable desc = no gpu: %w", ErrNoCapacity), true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := IsRejection(tt.err); got != tt.want { + t.Fatalf("IsRejection(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +// ClassifyError answers "how widely, IF we block" and keeps its capacity-shaped +// default for an unattributable error: a caller that has decided to block something +// still wants the narrow blast radius. IsRejection is the separate question of +// whether to block at all, so the two must not be collapsed. +func TestClassifyError_UnattributableStillScopesNarrow(t *testing.T) { + got := ClassifyError( + errors.New("rpc error: code = Unavailable desc = transport is closing"), + nebulav1alpha1.CapacitySpot, "H100:8") + if got.DenyAll { + t.Fatalf("an unattributable error must never widen to DenyAll, got %+v", got) + } + if got.Accelerator == nil || *got.Accelerator != "H100:8" || + got.CapacityType != nebulav1alpha1.CapacitySpot { + t.Fatalf("expected a Spot/H100:8-scoped block, got %+v", got) + } +} + func ptrStr(s string) *string { return &s } diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index b602c93..1a78f45 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -19,6 +19,7 @@ package vnode import ( "context" "encoding/json" + "fmt" "io" "math/rand/v2" "sync" @@ -37,6 +38,7 @@ import ( logf "sigs.k8s.io/controller-runtime/pkg/log" nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" + "github.com/InftyAI/Nebula/pkg/metrics" "github.com/InftyAI/Nebula/pkg/provider" "github.com/InftyAI/Nebula/pkg/util" ) @@ -156,6 +158,28 @@ type trackedPod struct { // never a lost value, because the annotation itself lives in etcd. No credential is // held here — see persistCredential. connectEndpoint string + + // provisionStart is when THIS process began provisioning for the pod. It arms the + // metrics.InstanceReadyDuration observation the poll loop makes on the first + // transition to Running, and is CONSUMED by it — a one-shot token, not a record, so + // zero carries both halves of "do not observe": + // + // - Never armed. A pod re-adopted after a restart (GetPod's cold-map path) has no + // recoverable start time — it died with the process — and measuring from + // re-adoption would report a multi-minute wait as milliseconds. Zero means "do + // not observe", NOT "observe as 0": a missing sample beats a wrong one. + // - Already spent. The poll loop walks every tracked pod every tick and a pod sits + // in Running for its whole life, so an observation that stayed armed would fire + // per tick with an ever-growing duration — a one-minute boot would contribute + // hundreds of samples stretching to the pod's full age, measuring longevity + // instead of readiness. + // + // The first case is a known bias worth naming: a provision still in flight when the + // manager restarts never contributes, so the histogram under-samples exactly the + // slowest boots. Closing it needs the start time persisted somewhere durable (a Pod + // annotation or NodeClaim status), which is a write on the provisioning path we have + // not taken. + provisionStart time.Time } // NewHandler builds a Handler for the given provider backend. The poll cadence @@ -223,15 +247,55 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { log.Info("provisioning external instance", "capacityType", req.CapacityType, "region", req.Region, "timeout", timeout.String()) + // Two clocks, because they measure two different waits. provisionStart anchors the + // end-to-end wait a user experiences (it runs until the instance reports Running, + // and is handed to store below), while callStart isolates the Provision call itself. + // They are NOT interchangeable: the emit between them is a synchronous notify that + // can issue an API write, so reusing one timestamp would silently charge that write + // to the provider's latency. + provisionStart := time.Now() + mlabels := h.metricLabels(pod) + // Report Provisioning BEFORE the call. It can run for minutes (AWS sweeps the // region's zones on a capacity error), and until it returns this is the only // explanation the Pod carries. Emit but do NOT store — see the tracked invariant. h.markStatus(pod, corev1.PodPending, reasonProvisioning, "allocating external instance") h.emit(pod) + callStart := time.Now() res, err := h.prov.Provision(provisionCtx, pod, req) + // One call site for both outcomes, so the attempt and failure counters cannot drift + // out of step. It counts the attempt regardless of how the error is handled below: + // an unreachable provider is a failed attempt even though nothing is blocklisted for + // it, and result="failure" with reason="unreachable" is precisely the series that + // says so. + metrics.ObserveProvision(mlabels, time.Since(callStart), err) if err != nil { - log.Error(err, "provision failed; Pod marked Failed for failover") + // An error the provider never attributed to this request — a transport failure, + // our own ProvisionTimeout, a 503 — is not a rejection, so it must not be acted + // on like one (see provider.IsRejection). Failing the Pod here would be a + // terminal verdict drawn from no evidence: the request may have been accepted, in + // which case the Pod is reaped out from under a paid instance whose id we never + // learned. Blocklisting would fence off a candidate that never misbehaved. + // + // So leave the Pod NON-terminal at the Provisioning it was already stamped with, + // carrying the error as its message so the wait is explained, and return the error + // for VK to retry with backoff. Provision is idempotent on ClaimName, so a retry + // adopts whatever the failed attempt may have created rather than doubling it. + // + // Deliberately NOT stored: the tracked invariant admits only acknowledged or + // terminal pods, and tracking this one with no instance id would have the poll + // loop find it absent from List and write the very Terminated status this branch + // exists to avoid. + if !provider.IsRejection(err) { + log.Error(err, "provision failed with an error the provider did not attribute "+ + "to this request; Pod left provisioning for retry, nothing blocklisted") + h.markStatus(pod, corev1.PodPending, reasonProvisioning, "retrying: "+err.Error()) + h.emit(pod) + return err + } + + log.Error(err, "provision rejected by the provider; Pod marked Failed for failover") // Record the failure on the shared blocklist so placement fails over to the // next candidate (zone → region → tier) instead of hot-looping here. The // provider classifies its own error into the precise BlockScope (a Spot @@ -242,7 +306,9 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // Surface the failure on the Pod so the placement controller can fail // over, and return the error so the pod controller retries with backoff. h.markStatus(pod, corev1.PodFailed, reasonProvisionFailed, err.Error()) - h.store(pod, claim, "") + // A zero start: this pod is terminal and will never reach Running, so there is no + // ready-duration to observe. + h.store(pod, claim, "", time.Time{}) h.emit(pod) return err } @@ -267,7 +333,7 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // loop uses, and every later tick re-offers it until a write lands. No lock: this // Pod is not shared until store, VK having handed us a copy of its own. setEndpoint(pod, res.ConnectURL) - h.store(pod, claim, res.InstanceID) + h.store(pod, claim, res.InstanceID, provisionStart) // The TOKEN, unlike the address, cannot ride the Pod (it would be readable by // anyone with `get pod` and sit unencrypted in etcd), so it gets its own write @@ -385,6 +451,18 @@ func (h *Handler) DeletePod(ctx context.Context, pod *corev1.Pod) error { // claim is still live, and if so rebuild the tracking entry from it. VK then sees // a non-nil pod and takes the UpdatePod (adopt) branch, and the re-tracked pod is // advanced to its true state by the next poll tick. +// +// The three outcomes are kept DISTINCT, because "the provider has no such instance" +// and "we could not ask the provider" have opposite consequences here. Reporting the +// latter as NotFound is what let a single failed List destroy a healthy workload: VK +// discards this function's error and branches on nil-ness alone (createOrUpdatePod in +// node/pod.go), so a nil pod re-issues CreatePod against an instance that is already +// running — and if the provider is still unreachable, that Provision fails too, marking +// the Pod Failed for a reap while the real instance keeps billing behind a zero +// instance id. So an unreachable provider returns a non-nil pod WITH an error: the +// non-nil pod suppresses the create, and the non-NotFound error makes VK's other +// caller (the pod controller's delete path) requeue rather than terminate. Both then +// wait for the next sync, which is the only correct move while ownership is unknown. func (h *Handler) GetPod(ctx context.Context, namespace, name string) (*corev1.Pod, error) { h.mu.Lock() tp, ok := h.tracked[key(namespace, name)] @@ -393,37 +471,54 @@ func (h *Handler) GetPod(ctx context.Context, namespace, name string) (*corev1.P return tp.pod.DeepCopy(), nil } + log := logf.FromContext(ctx).WithName("vnode-handler").WithValues( + "provider", h.prov.Name(), "pod", key(namespace, name)) + // Cold map: re-adopt from the live provider if the instance still exists. claim := util.ClaimName(namespace, name) - inst, found := h.instanceByClaim(ctx, claim) + inst, found, err := h.instanceByClaim(ctx, claim) + if err != nil { + // Loud, and never swallowed: this is the only signal that a Pod is being held + // back, and every alternative to holding back is a guess that costs either a + // duplicate paid instance or a reaped live one. + log.Error(err, "cannot determine whether an instance exists for this Pod; "+ + "holding off create and delete until the provider answers", "claim", claim) + stub := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}} + return stub, fmt.Errorf("provider %q unreachable, ownership of pod %s/%s unknown: %w", + h.prov.Name(), namespace, name, err) + } if !found { return nil, errdefs.NotFoundf("pod %s/%s not found on virtual node", namespace, name) } pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}} applyState(pod, inst.State, inst.Endpoint, h.nowFn()) - h.store(pod, claim, inst.ID) - logf.FromContext(ctx).WithName("vnode-handler").Info( - "re-adopted live instance after cold tracking map (VK restart)", - "provider", h.prov.Name(), "pod", key(namespace, name), "claim", claim, - "instanceID", inst.ID, "state", inst.State) + // A zero start: this process never provisioned the instance, so its real start time + // is gone and its ready-duration is not observable (see trackedPod.provisionStart). + h.store(pod, claim, inst.ID, time.Time{}) + log.Info("re-adopted live instance after cold tracking map (VK restart)", + "claim", claim, "instanceID", inst.ID, "state", inst.State) return pod.DeepCopy(), nil } -// instanceByClaim returns the live provider instance whose ClaimName matches, and -// whether one was found. A List error yields (zero,false): re-adoption then falls -// through to NotFound and VK retries on its next sync rather than acting on a -// half-known fleet. It backs GetPod's post-restart re-adoption. -func (h *Handler) instanceByClaim(ctx context.Context, claim string) (provider.Instance, bool) { +// instanceByClaim returns the live provider instance whose ClaimName matches, whether +// one was found, and any List error. +// +// The error is returned rather than folded into found=false because the two are not +// the same answer: found=false ASSERTS the instance does not exist, while an error +// means we do not know. GetPod acts very differently on each, and conflating them +// turned "unknown" into "absent" — the strongest possible claim from the least +// information. It backs GetPod's post-restart re-adoption. +func (h *Handler) instanceByClaim(ctx context.Context, claim string) (provider.Instance, bool, error) { instances, err := h.prov.List(ctx) if err != nil { - return provider.Instance{}, false + return provider.Instance{}, false, err } for _, inst := range instances { if inst.ClaimName == claim { - return inst, true + return inst, true, nil } } - return provider.Instance{}, false + return provider.Instance{}, false, nil } // GetPodStatus returns the tracked pod's status. @@ -524,6 +619,7 @@ func (h *Handler) reconcileOnce(ctx context.Context) { // The observed address, for a provider that cannot know it before boot. // Empty for one that published at create, which must not clear it. setEndpoint(tp.pod, inst.Endpoint) + h.observeReady(tp, inst.State) } // Log the before -> after status every tick. This is the "how does the system // look" signal an operator watches: the lifecycle progression (Provisioning -> @@ -563,6 +659,28 @@ func (h *Handler) reconcileOnce(ctx context.Context) { } } +// observeReady records the end-to-end provisioning wait the first time a tracked pod's +// instance reports Running. This is where the whole user-visible number is available and +// nowhere else: the poll loop is the only place InstanceRunning is ever reached (see +// applyState), because a provider's create returns long before the instance is usable. +// +// It measures with time.Since rather than h.nowFn, deliberately. nowFn is the seam for +// the STATUS clock, which a test may pin to a fixed instant; subtracting a real start +// time from a pinned now would yield a nonsense (possibly negative) duration, so the +// measurement is kept independent of it. +// +// The observation is ONE-SHOT: it consumes provisionStart, so a zero value covers both +// "never armed" (a re-adopted instance, whose real start died with the previous process) +// and "already recorded" — see trackedPod.provisionStart for why leaving it armed would +// turn one boot into a sample per poll tick. Callers must hold h.mu. +func (h *Handler) observeReady(tp *trackedPod, state provider.InstanceState) { + if state != provider.InstanceRunning || tp.provisionStart.IsZero() { + return + } + metrics.ObserveReady(h.metricLabels(tp.pod), time.Since(tp.provisionStart)) + tp.provisionStart = time.Time{} // spent; never observe this pod again +} + // setEndpoint stamps a reachable address onto a Pod's annotation — the single stamp // every path uses, so the annotation has one assignment site regardless of where the // address came from. The notify wrapper then patches it to the API server (PodIP cannot @@ -613,13 +731,37 @@ func statusSignature(pod *corev1.Pod) string { } // store records/updates the tracked pod under lock. -func (h *Handler) store(pod *corev1.Pod, claim, instance string) { +// +// provisionStart is when provisioning began, and arms the ready-duration observation +// (see trackedPod.provisionStart). Pass the zero Time from any path that cannot know +// it — a re-adoption after a restart, or a pod already terminal. +func (h *Handler) store(pod *corev1.Pod, claim, instance string, provisionStart time.Time) { h.mu.Lock() defer h.mu.Unlock() h.tracked[key(pod.Namespace, pod.Name)] = &trackedPod{ - pod: pod.DeepCopy(), - claimName: claim, - instance: instance, + pod: pod.DeepCopy(), + claimName: claim, + instance: instance, + provisionStart: provisionStart, + } +} + +// metricLabels renders the provisioning metric label set for a Pod, read off what +// placement stamped on it. +// +// The accelerator type and count are passed SEPARATELY, unlike recordBlock, which files +// the joined pool identity (util.AcceleratorPool). The two want opposite things from the +// same pair: a blocklist needs one opaque key so an H100:8 shortage never excludes +// H100:1, while a metric needs two dimensions so it can be aggregated either way. The +// pool key is still recoverable from the labels when correlating the two. +func (h *Handler) metricLabels(pod *corev1.Pod) metrics.Labels { + accel, count, _ := util.AcceleratorRequest(pod) + return metrics.Labels{ + Provider: h.prov.Name(), + Region: pod.Annotations[nebulav1alpha1.RegionAnnotation], + CapacityType: pod.Annotations[nebulav1alpha1.CapacityTypeAnnotation], + Accelerator: accel, + AcceleratorCount: count, } } diff --git a/pkg/vnode/handler_test.go b/pkg/vnode/handler_test.go index ba54ce2..cde152e 100644 --- a/pkg/vnode/handler_test.go +++ b/pkg/vnode/handler_test.go @@ -337,6 +337,61 @@ func TestCreatePod_ProvisionErrorSurfaces(t *testing.T) { } } +// An error the provider never attributed to the request is not a rejection, so it +// must not be acted on like one: no terminal status (the request may have been +// accepted, and a Failed Pod is reaped out from under a paid instance), no blocklist +// entry against a candidate that never misbehaved, and no tracking (a tracked pod with +// no instance id is written Terminated by the very next poll tick). The Pod stays +// Provisioning with the error as its message and VK retries with backoff. +func TestCreatePod_UnattributableErrorLeavesPodProvisioning(t *testing.T) { + provErr := errors.New("rpc error: code = Unavailable desc = transport is closing") + // A non-empty classifyScope proves the guard runs BEFORE classification: a provider + // willing to hand back a blockable scope still must not have one recorded. + accel := "H100:1" + fp := &fakeProvider{ + provisionErr: provErr, + classifyScope: provider.BlockScope{Accelerator: &accel}, + } + bl := &recordingBlocklist{} + h := NewHandler(fp, nil, bl) + + var mu sync.Mutex + var emitted []string + h.NotifyPods(context.Background(), func(p *corev1.Pod) { + mu.Lock() + emitted = append(emitted, string(p.Status.Phase)+"/"+p.Status.Reason) + mu.Unlock() + }) + + pod := testPod("default", "p1") + err := h.CreatePod(context.Background(), pod) + if !errors.Is(err, provErr) { + t.Fatalf("CreatePod must return the provision error for VK to back off, got %v", err) + } + if pod.Status.Phase != corev1.PodPending || pod.Status.Reason != reasonProvisioning { + t.Fatalf("status = %s/%s, want the non-terminal %s/%s", + pod.Status.Phase, pod.Status.Reason, corev1.PodPending, reasonProvisioning) + } + if !strings.Contains(pod.Status.Message, provErr.Error()) { + t.Fatalf("expected the error surfaced as the Pod message, got %q", pod.Status.Message) + } + if bl.calls != 0 { + t.Fatalf("expected no blocklist entry for an unattributable error, got %d", bl.calls) + } + if len(h.tracked) != 0 { + t.Fatalf("expected the Pod left untracked, got %d tracked", len(h.tracked)) + } + mu.Lock() + defer mu.Unlock() + // Both emits are the same non-terminal status: the pre-call stamp and the retry + // message. Neither may be Failed. + for _, e := range emitted { + if strings.HasPrefix(e, string(corev1.PodFailed)) { + t.Fatalf("emitted a terminal status %q for an unattributable error: %v", e, emitted) + } + } +} + func TestCreatePod_ProvisionFailureRecordsBlock(t *testing.T) { accel, region := "H100", "us-east-1" scope := provider.BlockScope{ @@ -1147,6 +1202,48 @@ func TestGetPod_ReAdoptsLiveInstanceAfterRestart(t *testing.T) { } } +// A List error means "we do not know whether an instance exists", which must NOT be +// reported as NotFound. VK discards GetPod's error and branches on nil-ness alone, so +// a nil pod re-issues CreatePod against an instance that may already be running — the +// path that reaps a live workload after a restart. A non-nil pod suppresses the create; +// the non-NotFound error makes VK's delete path requeue instead of terminating. +func TestGetPod_ListErrorIsUnknownNotAbsent(t *testing.T) { + fp := &fakeProvider{listErr: errors.New("rpc error: code = Unavailable")} + h := NewHandler(fp, nil, nil) + + got, err := h.GetPod(context.Background(), "default", "p1") + if err == nil { + t.Fatal("expected an error when the provider cannot be listed") + } + if errdefs.IsNotFound(err) { + t.Fatalf("a List failure must not read as NotFound (VK would then CreatePod): %v", err) + } + if got == nil { + t.Fatal("expected a non-nil Pod so VK takes the adopt branch instead of creating") + } + if got.Namespace != "default" || got.Name != "p1" { + t.Fatalf("stub Pod identity = %s/%s, want default/p1", got.Namespace, got.Name) + } + // Nothing may be tracked off a failed List: a tracked pod with no instance id would + // be found absent from the next List and written Terminated. + if len(h.tracked) != 0 { + t.Fatalf("expected nothing tracked after a failed List, got %d", len(h.tracked)) + } + + // Once the provider answers, the same call re-adopts normally. + fp.listErr = nil + fp.list = []provider.Instance{{ + ID: "inst-9", ClaimName: "default-p1", State: provider.InstanceRunning, + }} + got, err = h.GetPod(context.Background(), "default", "p1") + if err != nil { + t.Fatalf("GetPod after the provider recovered: %v", err) + } + if got.Status.Phase != corev1.PodRunning { + t.Fatalf("expected re-adoption once List succeeds, got %q", got.Status.Phase) + } +} + func TestGetPod_UnknownClaimStaysNotFound(t *testing.T) { // No tracking and no live instance for this claim => genuinely absent. GetPod // must report NotFound so VK creates it, not silently adopt a phantom. diff --git a/pkg/vnode/metrics_test.go b/pkg/vnode/metrics_test.go new file mode 100644 index 0000000..0aecd37 --- /dev/null +++ b/pkg/vnode/metrics_test.go @@ -0,0 +1,264 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +package vnode + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + dto "github.com/prometheus/client_model/go" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" + "github.com/InftyAI/Nebula/pkg/metrics" + "github.com/InftyAI/Nebula/pkg/provider" +) + +// The collectors are package-level and registered once, so every test in this package +// shares them. Assertions are therefore on the DELTA a call produced, never on an +// absolute value — otherwise the specs would pass or fail depending on run order. + +// histCount reads a histogram series' sample count. GetMetricWith creates the series +// when absent, so an unobserved label set reads as 0 rather than erroring. +func histCount(t *testing.T, h *prometheus.HistogramVec, l prometheus.Labels) uint64 { + t.Helper() + obs, err := h.GetMetricWith(l) + if err != nil { + t.Fatalf("GetMetricWith(%v): %v", l, err) + } + m, ok := obs.(prometheus.Metric) + if !ok { + t.Fatalf("observer for %v is not a prometheus.Metric", l) + } + pb := &dto.Metric{} + if err := m.Write(pb); err != nil { + t.Fatalf("write metric %v: %v", l, err) + } + return pb.GetHistogram().GetSampleCount() +} + +// metricPod is a Pod carrying everything the label set is read off: the tier and region +// placement stamped, plus the accelerator label the pool identity is derived from. +func metricPod(ns, name string) *corev1.Pod { + pod := testPod(ns, name) + pod.Annotations = map[string]string{ + nebulav1alpha1.CapacityTypeAnnotation: string(nebulav1alpha1.CapacitySpot), + nebulav1alpha1.RegionAnnotation: "us-east-1", + } + pod.Labels = map[string]string{nebulav1alpha1.AcceleratorTypeLabel: "H100"} + return pod +} + +// labelsFor is the label set metricPod produces. The accelerator TYPE and COUNT are +// separate labels so either aggregation works; the count is 1 because a type with no +// explicit nvidia.com/gpu limit means one GPU (see util.AcceleratorRequest). +func labelsFor(extraKey, extraVal string) prometheus.Labels { + l := prometheus.Labels{ + "provider": "fake", + "region": "us-east-1", + "capacity_type": string(nebulav1alpha1.CapacitySpot), + "accelerator": "H100", + "accelerator_count": "1", + } + if extraKey != "" { + l[extraKey] = extraVal + } + return l +} + +func TestCreatePod_RecordsSuccessfulProvisionAttempt(t *testing.T) { + success := labelsFor("result", metrics.ResultSuccess) + beforeAttempts := testutil.ToFloat64(metrics.ProvisionAttempts.With(success)) + beforeDuration := histCount(t, metrics.ProvisionDuration, success) + + h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil) + if err := h.CreatePod(context.Background(), metricPod("default", "m1")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + if got := testutil.ToFloat64(metrics.ProvisionAttempts.With(success)) - beforeAttempts; got != 1 { + t.Fatalf("success attempts delta = %v, want 1", got) + } + if got := histCount(t, metrics.ProvisionDuration, success) - beforeDuration; got != 1 { + t.Fatalf("duration observations delta = %d, want 1", got) + } +} + +// A rejection is counted with the reason the sentinel names, so the failure series can +// answer "are we out of capacity, or are our credentials broken?" without log grepping. +func TestCreatePod_RecordsRejectionReason(t *testing.T) { + failure := labelsFor("result", metrics.ResultFailure) + capacity := labelsFor("reason", metrics.ReasonCapacity) + beforeAttempts := testutil.ToFloat64(metrics.ProvisionAttempts.With(failure)) + beforeFailures := testutil.ToFloat64(metrics.ProvisionFailures.With(capacity)) + + fp := &fakeProvider{provisionErr: provider.ErrNoCapacity} + h := NewHandler(fp, nil, nil) + if err := h.CreatePod(context.Background(), metricPod("default", "m2")); err == nil { + t.Fatal("expected the provision error") + } + + if got := testutil.ToFloat64(metrics.ProvisionAttempts.With(failure)) - beforeAttempts; got != 1 { + t.Fatalf("failure attempts delta = %v, want 1", got) + } + if got := testutil.ToFloat64(metrics.ProvisionFailures.With(capacity)) - beforeFailures; got != 1 { + t.Fatalf("capacity failures delta = %v, want 1", got) + } +} + +// An unreachable provider is still a counted ATTEMPT — the call happened and failed — +// even though the handler deliberately does not fail the Pod or blocklist anything for +// it. reason="unreachable" is what distinguishes an integration outage from a capacity +// shortfall, which is the whole reason the two are not both "other". +func TestCreatePod_UnreachableProviderCountedSeparately(t *testing.T) { + failure := labelsFor("result", metrics.ResultFailure) + unreachable := labelsFor("reason", metrics.ReasonUnreachable) + capacity := labelsFor("reason", metrics.ReasonCapacity) + beforeAttempts := testutil.ToFloat64(metrics.ProvisionAttempts.With(failure)) + beforeUnreachable := testutil.ToFloat64(metrics.ProvisionFailures.With(unreachable)) + beforeCapacity := testutil.ToFloat64(metrics.ProvisionFailures.With(capacity)) + + fp := &fakeProvider{provisionErr: errors.New("rpc error: code = Unavailable desc = transport is closing")} + h := NewHandler(fp, nil, nil) + if err := h.CreatePod(context.Background(), metricPod("default", "m3")); err == nil { + t.Fatal("expected the provision error") + } + + if got := testutil.ToFloat64(metrics.ProvisionAttempts.With(failure)) - beforeAttempts; got != 1 { + t.Fatalf("failure attempts delta = %v, want 1", got) + } + if got := testutil.ToFloat64(metrics.ProvisionFailures.With(unreachable)) - beforeUnreachable; got != 1 { + t.Fatalf("unreachable failures delta = %v, want 1", got) + } + // "Unavailable" in the gRPC status text must not be read as a capacity shortfall. + if got := testutil.ToFloat64(metrics.ProvisionFailures.With(capacity)) - beforeCapacity; got != 0 { + t.Fatalf("capacity failures delta = %v, want 0 for a transport error", got) + } +} + +// The ready duration is observed on the FIRST tick that reports Running and never +// again, because provisionStart is consumed. Without that, every subsequent tick would +// add a sample with an ever-growing value — measuring the pod's age, not its boot. +func TestReconcileOnce_ObservesReadyDurationExactlyOnce(t *testing.T) { + ready := labelsFor("", "") + before := histCount(t, metrics.InstanceReadyDuration, ready) + + fp := &fakeProvider{provisionID: "inst-1"} + h := NewHandler(fp, nil, nil) + if err := h.CreatePod(context.Background(), metricPod("default", "m4")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + // Still initializing: nothing to observe yet. + fp.list = []provider.Instance{{ID: "inst-1", ClaimName: "default-m4", State: provider.InstancePending}} + h.reconcileOnce(context.Background()) + if got := histCount(t, metrics.InstanceReadyDuration, ready) - before; got != 0 { + t.Fatalf("observations while Pending = %d, want 0", got) + } + + // First Running tick observes. + fp.list = []provider.Instance{{ID: "inst-1", ClaimName: "default-m4", State: provider.InstanceRunning}} + h.reconcileOnce(context.Background()) + if got := histCount(t, metrics.InstanceReadyDuration, ready) - before; got != 1 { + t.Fatalf("observations after the first Running tick = %d, want 1", got) + } + + // Every later tick must add nothing, however long the pod runs. + h.reconcileOnce(context.Background()) + h.reconcileOnce(context.Background()) + if got := histCount(t, metrics.InstanceReadyDuration, ready) - before; got != 1 { + t.Fatalf("observations after three Running ticks = %d, want 1 (the token is one-shot)", got) + } +} + +// A pod re-adopted after a restart has no recoverable start time, so its ready duration +// is SKIPPED rather than measured from re-adoption — which would report a wait of +// minutes as microseconds and bias the histogram fast. A missing sample beats a wrong +// one. +func TestGetPod_ReAdoptedPodIsNotReadyObserved(t *testing.T) { + // A re-adopted pod is a synthesized stub with no annotations or labels, so it renders + // to the all-"none" series — that, not the fully-labelled one, is where a wrongly + // taken observation would land, and it is the assertion that carries this test. Both + // are deltas because other specs in this package write to both series. + ready := labelsFor("", "") + none := prometheus.Labels{ + "provider": "fake", "region": "none", "capacity_type": "none", + "accelerator": "none", "accelerator_count": "none", + } + beforeReady := histCount(t, metrics.InstanceReadyDuration, ready) + beforeNone := histCount(t, metrics.InstanceReadyDuration, none) + + // Cold map, live instance: the re-adoption path. The instance is already Running, + // so a naive implementation would observe on the very next tick. + fp := &fakeProvider{list: []provider.Instance{{ + ID: "inst-9", ClaimName: "default-m5", State: provider.InstanceRunning, + }}} + h := NewHandler(fp, nil, nil) + if _, err := h.GetPod(context.Background(), "default", "m5"); err != nil { + t.Fatalf("GetPod: %v", err) + } + h.reconcileOnce(context.Background()) + h.reconcileOnce(context.Background()) + + if got := histCount(t, metrics.InstanceReadyDuration, none) - beforeNone; got != 0 { + t.Fatalf("observations for a re-adopted pod = %d, want 0", got) + } + if got := histCount(t, metrics.InstanceReadyDuration, ready) - beforeReady; got != 0 { + t.Fatalf("observations on the labelled series = %d, want 0", got) + } +} + +// The provisioning clock is independent of nowFn, the STATUS clock a test may pin. A +// pinned status clock must not produce a nonsense (or negative) duration. +func TestObserveReady_IndependentOfPinnedStatusClock(t *testing.T) { + ready := labelsFor("", "") + before := histCount(t, metrics.InstanceReadyDuration, ready) + + fp := &fakeProvider{provisionID: "inst-1"} + h := NewHandler(fp, nil, nil) + // A status clock pinned far in the PAST: reusing it to measure would go negative. + pinned := metav1.NewTime(time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)) + h.nowFn = func() metav1.Time { return pinned } + + if err := h.CreatePod(context.Background(), metricPod("default", "m6")); err != nil { + t.Fatalf("CreatePod: %v", err) + } + fp.list = []provider.Instance{{ID: "inst-1", ClaimName: "default-m6", State: provider.InstanceRunning}} + h.reconcileOnce(context.Background()) + + if got := histCount(t, metrics.InstanceReadyDuration, ready) - before; got != 1 { + t.Fatalf("observations = %d, want 1", got) + } + // A negative duration lands in no bucket, so a positive count in the smallest one is + // the proof the measurement used the real clock. + obs, err := metrics.InstanceReadyDuration.GetMetricWith(ready) + if err != nil { + t.Fatalf("GetMetricWith: %v", err) + } + pb := &dto.Metric{} + if err := obs.(prometheus.Metric).Write(pb); err != nil { + t.Fatalf("write: %v", err) + } + if sum := pb.GetHistogram().GetSampleSum(); sum < 0 { + t.Fatalf("sample sum = %v, want a non-negative duration (the status clock leaked in)", sum) + } +} From 0e18dabcaaf027bbaff684b31716f435fd8dec4d Mon Sep 17 00:00:00 2001 From: kerthcet Date: Sat, 15 Aug 2026 12:15:24 +0100 Subject: [PATCH 2/4] update metrics Signed-off-by: kerthcet --- internal/controller/pod_placement_helpers.go | 2 +- pkg/metrics/doc.go | 41 +++++++ pkg/metrics/labels.go | 100 ++++++++++++++++++ pkg/metrics/labels_test.go | 58 ++++++++++ pkg/metrics/placement.go | 4 +- pkg/metrics/{metrics.go => provision.go} | 94 +--------------- .../{metrics_test.go => provision_test.go} | 39 ------- 7 files changed, 206 insertions(+), 132 deletions(-) create mode 100644 pkg/metrics/doc.go create mode 100644 pkg/metrics/labels.go create mode 100644 pkg/metrics/labels_test.go rename pkg/metrics/{metrics.go => provision.go} (66%) rename pkg/metrics/{metrics_test.go => provision_test.go} (75%) diff --git a/internal/controller/pod_placement_helpers.go b/internal/controller/pod_placement_helpers.go index d85dbfa..51cd65a 100644 --- a/internal/controller/pod_placement_helpers.go +++ b/internal/controller/pod_placement_helpers.go @@ -191,7 +191,7 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev // // The accelerator type and count are passed apart, not as the joined pool identity // placement.accelerator carries, because a metric label must be aggregatable (see -// metrics.provisionLabels). The parse cannot fail here: selectPlacement already +// metrics.candidateLabels). The parse cannot fail here: selectPlacement already // rejected a malformed request before any placement was returned. func placementLabels(pod *corev1.Pod, p placement) metrics.Labels { accel, count, _ := util.AcceleratorRequest(pod) diff --git a/pkg/metrics/doc.go b/pkg/metrics/doc.go new file mode 100644 index 0000000..6f9aa58 --- /dev/null +++ b/pkg/metrics/doc.go @@ -0,0 +1,41 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +// Package metrics holds Nebula's Prometheus instrumentation. +// +// Everything here registers into controller-runtime's registry, so it is served on +// the manager's existing --metrics-bind-address endpoint alongside the standard +// controller/workqueue metrics. Importing this package is what registers the +// collectors (see the init in each file); no wiring is needed in main. +// +// The instrumented surface is the path a Pod takes from admission to a running +// external instance, one file per leg: +// +// placement.go the Pod is gated -> a candidate is chosen -> the gate is removed +// provision.go the provider is called -> the instance reports Running +// +// Those are the parts whose cost and failure modes are otherwise invisible — +// placement can silently leave a Pod gated forever, and provisioning runs against a +// third party, takes seconds to minutes, bills money, and fails for reasons the Pod +// status flattens away. Everything else is already covered elsewhere and deliberately +// not duplicated: reconcile counts, queue depth and API latency by +// controller-runtime's own collectors, Pod-population questions ("how many Pods are +// gated right now?") by kube-state-metrics. +// +// The two legs deliberately share one label set (labels.go) so a placement and the +// provisioning attempt it led to carry identical label values and can be joined in +// PromQL without label surgery. See docs/metrics.md for the operator-facing view. +package metrics diff --git a/pkg/metrics/labels.go b/pkg/metrics/labels.go new file mode 100644 index 0000000..fd46714 --- /dev/null +++ b/pkg/metrics/labels.go @@ -0,0 +1,100 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +package metrics + +import "strconv" + +// none is the placeholder for a label the request genuinely did not carry: no +// region (unconstrained), no capacity tier (the provider's default), or no +// accelerator (a CPU-only Pod). An explicit token beats an empty string, which in +// PromQL is indistinguishable from a label that was never set and silently matches +// `{region=""}` selectors an operator did not mean to write. +const none = "none" + +// candidateLabels is the label set that identifies a PLACEMENT CANDIDATE — the +// (provider, region, capacity type, accelerator pool) tuple that placement selects and +// provisioning then acts on. Both legs carry it, in this order, which is what lets a +// placement and the provisioning attempt it led to be joined in PromQL without label +// surgery. Extending it therefore touches both legs at once; that is intended. +// +// It is deliberately the same dimensions as failover.Candidate, minus the joined +// accelerator (see below), so a counted failure and an excluded candidate line up. +// +// Region is included because a capacity shortfall is region-local and comparing regions +// is the whole point of collecting this; note that for a provider which collapses +// several declared regions into one candidate (Modal) the value is that provider's +// joined token, not a single region name — see NodeClaimSpec.Region. +// +// The accelerator TYPE and COUNT are two labels, not the joined "H100:8" pool identity +// used as the blocklist key. A metric label set is meant to be aggregated over, and a +// joined string cannot be: `sum by (accelerator)` over every size of H100 requires +// splitting the value in PromQL, and selecting all 8-GPU requests across types is not +// expressible at all. Two labels give both for free, and the pool key is still +// recoverable as accelerator + ":" + accelerator_count when correlating with a +// blocklist entry. +var candidateLabels = []string{"provider", "region", "capacity_type", "accelerator", "accelerator_count"} + +// withExtra returns candidateLabels plus one trailing dimension (result, reason), for +// the collectors that carry an outcome. It copies, because appending to a package-level +// slice would let two collectors share and overwrite one backing array. +func withExtra(name string) []string { + return append(append([]string{}, candidateLabels...), name) +} + +// Labels identifies the candidate one placement or provisioning attempt was made +// against. The zero value is valid: every field normalizes to "none". +type Labels struct { + Provider string + Region string + CapacityType string + // Accelerator is the accelerator TYPE alone (e.g. "H100"), and AcceleratorCount how + // many were requested. They are kept apart so both aggregations work — see + // candidateLabels. Empty/zero for a CPU-only Pod. + Accelerator string + AcceleratorCount int32 +} + +// values renders the label set in candidateLabels order, with extra appended for +// the metrics that carry a result/reason dimension. +func (l Labels) values(extra ...string) []string { + return append([]string{ + orNone(l.Provider), + orNone(l.Region), + orNone(l.CapacityType), + orNone(l.Accelerator), + countOrNone(l.AcceleratorCount), + }, extra...) +} + +func orNone(s string) string { + if s == "" { + return none + } + return s +} + +// countOrNone renders an accelerator count, or the placeholder when there is no +// accelerator to count. Zero is deliberately NOT rendered as "0": a CPU-only Pod did not +// request zero GPUs, it requested none at all, and "0" would put it in the same numeric +// series an operator reads as a real count. util.AcceleratorRequest never returns a +// positive count without a type, so this cannot mask a real request. +func countOrNone(n int32) string { + if n <= 0 { + return none + } + return strconv.FormatInt(int64(n), 10) +} diff --git a/pkg/metrics/labels_test.go b/pkg/metrics/labels_test.go new file mode 100644 index 0000000..e9eb9ce --- /dev/null +++ b/pkg/metrics/labels_test.go @@ -0,0 +1,58 @@ +/* +Copyright 2026 The InftyAI Team. + +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. +*/ + +package metrics + +import "testing" + +// An unset field renders as the explicit "none" placeholder, never "" — which in PromQL +// is indistinguishable from a label that was never set and silently matches {region=""} +// selectors an operator did not mean to write. The accelerator COUNT is held to the same +// rule: a CPU-only Pod did not request zero GPUs, it requested none, and rendering "0" +// would drop it into the numeric series an operator reads as real counts. +func TestLabels_RendersInCandidateLabelOrder(t *testing.T) { + tests := []struct { + name string + in Labels + want []string + }{ + {"zero value is all placeholders", Labels{}, []string{none, none, none, none, none}}, + { + "cpu-only pod has no count, not a zero count", + Labels{Provider: "p", Region: "r", CapacityType: "OnDemand"}, + []string{"p", "r", "OnDemand", none, none}, + }, + { + "accelerator type and count are separate values", + Labels{Provider: "p", Region: "r", CapacityType: "Spot", Accelerator: "H100", AcceleratorCount: 8}, + []string{"p", "r", "Spot", "H100", "8"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := tt.in.values() + if len(got) != len(candidateLabels) { + t.Fatalf("values() = %v (%d values), want %d to match candidateLabels %v", + got, len(got), len(candidateLabels), candidateLabels) + } + for i := range tt.want { + if got[i] != tt.want[i] { + t.Fatalf("values()[%d] (%s) = %q, want %q", i, candidateLabels[i], got[i], tt.want[i]) + } + } + }) + } +} diff --git a/pkg/metrics/placement.go b/pkg/metrics/placement.go index 28b0560..5334d8b 100644 --- a/pkg/metrics/placement.go +++ b/pkg/metrics/placement.go @@ -64,7 +64,7 @@ var ( PlacementDecisions = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "nebula_placement_decisions_total", Help: "Pods placed, by provider, region, capacity type, accelerator type and accelerator count.", - }, provisionLabels) + }, candidateLabels) // PlacementWaitDuration measures from Pod creation to the gate being removed: the // user-visible queue time BEFORE provisioning starts, which @@ -80,7 +80,7 @@ var ( Name: "nebula_placement_wait_duration_seconds", Help: "Time from Pod creation to placement (scheduling gate removal), by provider, region, capacity type, accelerator type and accelerator count.", Buckets: []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 1800}, - }, provisionLabels) + }, candidateLabels) // PlacementDeferrals counts reconciles that ended without placing the Pod, by // reason. diff --git a/pkg/metrics/metrics.go b/pkg/metrics/provision.go similarity index 66% rename from pkg/metrics/metrics.go rename to pkg/metrics/provision.go index ff06f77..befcab9 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/provision.go @@ -14,31 +14,11 @@ See the License for the specific language governing permissions and limitations under the License. */ -// Package metrics holds Nebula's Prometheus instrumentation. -// -// Everything here registers into controller-runtime's registry, so it is served on -// the manager's existing --metrics-bind-address endpoint alongside the standard -// controller/workqueue metrics. Importing this package is what registers the -// collectors (see init); no wiring is needed in main. -// -// The instrumented surface is the path a Pod takes from admission to a running -// external instance: PLACEMENT (this file's siblings in placement.go) and -// PROVISIONING (here). Those are the parts whose cost and failure modes are -// otherwise invisible — placement can silently leave a Pod gated forever, and -// provisioning runs against a third party, takes seconds to minutes, bills money, -// and fails for reasons the Pod status flattens away. Everything else (reconcile -// counts, queue depth, API latency) is already covered by controller-runtime's own -// metrics, and Pod-population questions by kube-state-metrics. -// -// The two halves deliberately share one label set (Labels, provisionLabels) so a -// placement and the provisioning attempt it led to carry identical label values and -// can be joined in PromQL without label surgery. package metrics import ( "context" "errors" - "strconv" "time" "github.com/prometheus/client_golang/prometheus" @@ -81,28 +61,6 @@ const ( ReasonOther = "other" ) -// none is the placeholder for a label the request genuinely did not carry: no -// region (unconstrained), no capacity tier (the provider's default), or no -// accelerator (a CPU-only Pod). An explicit token beats an empty string, which in -// PromQL is indistinguishable from a label that was never set and silently matches -// `{region=""}` selectors an operator did not mean to write. -const none = "none" - -// provisionLabels is the label set every provisioning metric carries, in order. -// Region is included because a capacity shortfall is region-local and comparing -// regions is the whole point of collecting this; note that for a provider which -// collapses several declared regions into one candidate (Modal) the value is that -// provider's joined token, not a single region name — see NodeClaimSpec.Region. -// -// The accelerator TYPE and COUNT are two labels, not the joined "H100:8" pool identity -// used as the blocklist key. A metric label set is meant to be aggregated over, and a -// joined string cannot be: `sum by (accelerator)` over every size of H100 requires -// splitting the value in PromQL, and selecting all 8-GPU requests across types is not -// expressible at all. Two labels give both for free, and the pool key is still -// recoverable as accelerator + ":" + accelerator_count when correlating with a -// blocklist entry. -var provisionLabels = []string{"provider", "region", "capacity_type", "accelerator", "accelerator_count"} - var ( // ProvisionAttempts counts provisioning attempts by outcome. Rate of the // result="failure" series over the total is the provisioning error rate; the @@ -110,7 +68,7 @@ var ( ProvisionAttempts = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "nebula_provision_attempts_total", Help: "Total external instance provisioning attempts, by provider, region, capacity type, accelerator type, accelerator count and outcome.", - }, append(append([]string{}, provisionLabels...), "result")) + }, withExtra("result")) // ProvisionFailures breaks failures down by coarse cause. It deliberately // overlaps ProvisionAttempts{result="failure"} rather than adding a reason label @@ -120,7 +78,7 @@ var ( ProvisionFailures = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "nebula_provision_failures_total", Help: "Failed provisioning attempts by coarse cause (capacity, quota, auth, unsupported_accelerator, timeout, other).", - }, append(append([]string{}, provisionLabels...), "reason")) + }, withExtra("reason")) // ProvisionDuration measures the provider's Provision call alone — not the // wait for the instance to become usable. The two differ enormously and for @@ -133,7 +91,7 @@ var ( Name: "nebula_provision_duration_seconds", Help: "Latency of the provider's Provision call, by provider, region, capacity type, accelerator type, accelerator count and outcome.", Buckets: []float64{0.5, 1, 2.5, 5, 10, 20, 30, 45, 60, 90, 120, 180, 300}, - }, append(append([]string{}, provisionLabels...), "result")) + }, withExtra("result")) // InstanceReadyDuration measures the whole user-visible wait: from the moment // CreatePod starts provisioning to the first poll tick that reports the instance @@ -151,7 +109,7 @@ var ( Name: "nebula_instance_ready_duration_seconds", Help: "Time from the start of provisioning to the instance first reporting Running, by provider, region, capacity type, accelerator type and accelerator count.", Buckets: []float64{5, 10, 20, 30, 45, 60, 90, 120, 180, 300, 600, 900, 1800}, - }, provisionLabels) + }, candidateLabels) ) func init() { @@ -163,50 +121,6 @@ func init() { ) } -// Labels identifies the placement one provisioning attempt was made against. The -// zero value is valid: every field normalizes to "none". -type Labels struct { - Provider string - Region string - CapacityType string - // Accelerator is the accelerator TYPE alone (e.g. "H100"), and AcceleratorCount how - // many were requested. They are kept apart so both aggregations work — see - // provisionLabels. Empty/zero for a CPU-only Pod. - Accelerator string - AcceleratorCount int32 -} - -// values renders the label set in provisionLabels order, with extra appended for -// the metrics that carry a result/reason dimension. -func (l Labels) values(extra ...string) []string { - return append([]string{ - orNone(l.Provider), - orNone(l.Region), - orNone(l.CapacityType), - orNone(l.Accelerator), - countOrNone(l.AcceleratorCount), - }, extra...) -} - -func orNone(s string) string { - if s == "" { - return none - } - return s -} - -// countOrNone renders an accelerator count, or the placeholder when there is no -// accelerator to count. Zero is deliberately NOT rendered as "0": a CPU-only Pod did not -// request zero GPUs, it requested none at all, and "0" would put it in the same numeric -// series an operator reads as a real count. util.AcceleratorRequest never returns a -// positive count without a type, so this cannot mask a real request. -func countOrNone(n int32) string { - if n <= 0 { - return none - } - return strconv.FormatInt(int64(n), 10) -} - // ObserveProvision records one completed provisioning attempt: its outcome, its // latency, and — when it failed — the coarse cause. It is the single call the // virtual kubelet makes on both the success and failure paths, so the attempt and diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/provision_test.go similarity index 75% rename from pkg/metrics/metrics_test.go rename to pkg/metrics/provision_test.go index 38b865e..33a2563 100644 --- a/pkg/metrics/metrics_test.go +++ b/pkg/metrics/provision_test.go @@ -100,42 +100,3 @@ func TestObserveProvision_CountersStayInStep(t *testing.T) { } } } - -// An unset field renders as the explicit "none" placeholder, never "" — which in PromQL -// is indistinguishable from a label that was never set and silently matches {region=""} -// selectors an operator did not mean to write. The accelerator COUNT is held to the same -// rule: a CPU-only Pod did not request zero GPUs, it requested none, and rendering "0" -// would drop it into the numeric series an operator reads as real counts. -func TestLabels_RendersInProvisionLabelOrder(t *testing.T) { - tests := []struct { - name string - in Labels - want []string - }{ - {"zero value is all placeholders", Labels{}, []string{none, none, none, none, none}}, - { - "cpu-only pod has no count, not a zero count", - Labels{Provider: "p", Region: "r", CapacityType: "OnDemand"}, - []string{"p", "r", "OnDemand", none, none}, - }, - { - "accelerator type and count are separate values", - Labels{Provider: "p", Region: "r", CapacityType: "Spot", Accelerator: "H100", AcceleratorCount: 8}, - []string{"p", "r", "Spot", "H100", "8"}, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - got := tt.in.values() - if len(got) != len(provisionLabels) { - t.Fatalf("values() = %v (%d values), want %d to match provisionLabels %v", - got, len(got), len(provisionLabels), provisionLabels) - } - for i := range tt.want { - if got[i] != tt.want[i] { - t.Fatalf("values()[%d] (%s) = %q, want %q", i, provisionLabels[i], got[i], tt.want[i]) - } - } - }) - } -} From a45cd73f4f2c4e6154773bb4d34c53b2389bab4d Mon Sep 17 00:00:00 2001 From: kerthcet Date: Sat, 15 Aug 2026 15:59:43 +0100 Subject: [PATCH 3/4] fix image tag Signed-off-by: kerthcet --- config/manager/kustomization.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/config/manager/kustomization.yaml b/config/manager/kustomization.yaml index 547a445..086bc1e 100644 --- a/config/manager/kustomization.yaml +++ b/config/manager/kustomization.yaml @@ -4,5 +4,5 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization images: - name: controller - newName: example.com/nebula - newTag: v0.0.1 + newName: inftyai/nebula-controller + newTag: latest From a788bbdd2b3542be856af0427e50cee954efa626 Mon Sep 17 00:00:00 2001 From: kerthcet Date: Sat, 15 Aug 2026 16:12:15 +0100 Subject: [PATCH 4/4] fix lint Signed-off-by: kerthcet --- pkg/metrics/placement.go | 8 +++++--- pkg/metrics/provision.go | 16 ++++++++++------ pkg/metrics/provision_test.go | 6 +++++- 3 files changed, 20 insertions(+), 10 deletions(-) diff --git a/pkg/metrics/placement.go b/pkg/metrics/placement.go index 5334d8b..6266186 100644 --- a/pkg/metrics/placement.go +++ b/pkg/metrics/placement.go @@ -77,8 +77,9 @@ var ( // magnitude because the honest range does: an unblocked Pod is placed in // milliseconds, while one waiting out a failover block waits the blocklist TTL. PlacementWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Name: "nebula_placement_wait_duration_seconds", - Help: "Time from Pod creation to placement (scheduling gate removal), by provider, region, capacity type, accelerator type and accelerator count.", + Name: "nebula_placement_wait_duration_seconds", + Help: "Time from Pod creation to placement (scheduling gate removal), by provider, region, " + + "capacity type, accelerator type and accelerator count.", Buckets: []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 1800}, }, candidateLabels) @@ -92,7 +93,8 @@ var ( // kube-state-metrics and use this series to explain WHY. PlacementDeferrals = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "nebula_placement_deferrals_total", - Help: "Reconciles that ended without placing a Pod, by pool and reason (no_pool, invalid_request, all_blocked, no_candidate, stale_claim).", + Help: "Reconciles that ended without placing a Pod, by pool and reason " + + "(no_pool, invalid_request, all_blocked, no_candidate, stale_claim).", }, []string{"pool", "reason"}) // CandidateSkips counts individual (tier, provider, region) candidates passed over diff --git a/pkg/metrics/provision.go b/pkg/metrics/provision.go index befcab9..7deb6f6 100644 --- a/pkg/metrics/provision.go +++ b/pkg/metrics/provision.go @@ -67,7 +67,8 @@ var ( // per-region/accelerator breakdown is what tells you WHERE it is failing. ProvisionAttempts = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "nebula_provision_attempts_total", - Help: "Total external instance provisioning attempts, by provider, region, capacity type, accelerator type, accelerator count and outcome.", + Help: "Total external instance provisioning attempts, by provider, region, capacity type, " + + "accelerator type, accelerator count and outcome.", }, withExtra("result")) // ProvisionFailures breaks failures down by coarse cause. It deliberately @@ -77,7 +78,8 @@ var ( // them. ProvisionFailures = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "nebula_provision_failures_total", - Help: "Failed provisioning attempts by coarse cause (capacity, quota, auth, unsupported_accelerator, timeout, other).", + Help: "Failed provisioning attempts by coarse cause " + + "(capacity, quota, auth, unsupported_accelerator, timeout, other).", }, withExtra("reason")) // ProvisionDuration measures the provider's Provision call alone — not the @@ -88,8 +90,9 @@ var ( // Bucketed out to 300s because the call is bounded by // Capabilities.ProvisionTimeout, which AWS raises above the 90s default. ProvisionDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Name: "nebula_provision_duration_seconds", - Help: "Latency of the provider's Provision call, by provider, region, capacity type, accelerator type, accelerator count and outcome.", + Name: "nebula_provision_duration_seconds", + Help: "Latency of the provider's Provision call, by provider, region, capacity type, " + + "accelerator type, accelerator count and outcome.", Buckets: []float64{0.5, 1, 2.5, 5, 10, 20, 30, 45, 60, 90, 120, 180, 300}, }, withExtra("result")) @@ -106,8 +109,9 @@ var ( // 30min because a queueing provider on a large GPU shape genuinely takes that // long. InstanceReadyDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Name: "nebula_instance_ready_duration_seconds", - Help: "Time from the start of provisioning to the instance first reporting Running, by provider, region, capacity type, accelerator type and accelerator count.", + Name: "nebula_instance_ready_duration_seconds", + Help: "Time from the start of provisioning to the instance first reporting Running, " + + "by provider, region, capacity type, accelerator type and accelerator count.", Buckets: []float64{5, 10, 20, 30, 45, 60, 90, 120, 180, 300, 600, 900, 1800}, }, candidateLabels) ) diff --git a/pkg/metrics/provision_test.go b/pkg/metrics/provision_test.go index 33a2563..4122d2e 100644 --- a/pkg/metrics/provision_test.go +++ b/pkg/metrics/provision_test.go @@ -91,7 +91,11 @@ func TestObserveProvision_CountersStayInStep(t *testing.T) { t.Fatalf("success attempts delta = %v, want 1", got) } // A success must never touch the failure-reason counter, whatever the reason. - for _, reason := range []string{ReasonCapacity, ReasonAuth, ReasonQuota, ReasonUnsupported, ReasonTimeout, ReasonUnreachable, ReasonOther} { + allReasons := []string{ + ReasonCapacity, ReasonAuth, ReasonQuota, + ReasonUnsupported, ReasonTimeout, ReasonUnreachable, ReasonOther, + } + for _, reason := range allReasons { if reason == ReasonCapacity { continue // asserted above }