From 515f59906fbd8bebe51f303745edf7be13014dd0 Mon Sep 17 00:00:00 2001 From: kerthcet Date: Sat, 22 Aug 2026 15:32:27 +0100 Subject: [PATCH 1/3] remove placement regions Signed-off-by: kerthcet --- api/v1alpha1/groupversion_info.go | 13 - cmd/main.go | 8 +- .../pod_placement_controller_test.go | 66 +++- internal/controller/pod_placement_helpers.go | 31 +- pkg/provider/modal/modal.go | 4 +- pkg/provider/provider.go | 2 +- pkg/vnode/cluster.go | 136 ++++++++ pkg/vnode/doc.go | 13 +- pkg/vnode/env_test.go | 4 +- pkg/vnode/exec_test.go | 16 +- pkg/vnode/handler.go | 140 +++++--- pkg/vnode/handler_test.go | 312 ++++++++++++------ pkg/vnode/kubelet_test.go | 10 +- pkg/vnode/logs_test.go | 14 +- pkg/vnode/metrics_test.go | 35 +- pkg/vnode/node.go | 21 +- pkg/vnode/pool.go | 99 ------ 17 files changed, 566 insertions(+), 358 deletions(-) create mode 100644 pkg/vnode/cluster.go delete mode 100644 pkg/vnode/pool.go diff --git a/api/v1alpha1/groupversion_info.go b/api/v1alpha1/groupversion_info.go index df9673d..4aece76 100644 --- a/api/v1alpha1/groupversion_info.go +++ b/api/v1alpha1/groupversion_info.go @@ -99,19 +99,6 @@ const ( // via util.AcceleratorRequest. AcceleratorTypeLabel = "nebula.inftyai.com/accelerator-type" - // CapacityTypeAnnotation carries the chosen purchase tier (Spot/OnDemand). It is a - // provisioning input the Pod spec cannot express, so the placement controller - // writes it when it ungates the Pod and the virtual kubelet — which provisions - // solely from the Pod — reads it back on CreatePod. Empty means "use the provider's - // default" (Modal is OnDemand-only and ignores it). - CapacityTypeAnnotation = "nebula.inftyai.com/capacity-type" - - // RegionAnnotation carries the chosen provider region. Same flow as - // CapacityTypeAnnotation: stamped at ungate, read on CreatePod into - // ProvisionRequest.Region. Absent means the provider's default region; - // region-simple providers (Modal, RunPod) ignore it. - RegionAnnotation = "nebula.inftyai.com/region" - // EndpointAnnotation carries the reachable address of the external instance (a DNS // name, an IP, or a URL, in the provider's own form). It is the only way to reach // the workload, and PodIP cannot hold it — the API server validates PodIP as a diff --git a/cmd/main.go b/cmd/main.go index 0a5b1d4..c1ddbfb 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -458,10 +458,10 @@ func setupVirtualNodes(mgr ctrl.Manager, blocklist vnode.Blocklister, kubeletSrv if !ok { continue } - // The manager's client, so the pool read on the provisioning path hits the shared - // cache the controllers already keep warm. - pools := vnode.NewCachedPoolReader(mgr.GetClient()) - if err := mgr.Add(vnode.NewRunner(prov, clientset, blocklist, kubeletSrv, pools)); err != nil { + // The manager's client, so the pool and claim reads on the provisioning path hit the + // shared cache the controllers already keep warm. + cluster := vnode.NewCachedClusterReader(mgr.GetClient()) + if err := mgr.Add(vnode.NewRunner(prov, clientset, blocklist, kubeletSrv, cluster)); err != nil { return err } setupLog.Info("registered virtual node", "provider", name, "node", vnode.NodeName(name)) diff --git a/internal/controller/pod_placement_controller_test.go b/internal/controller/pod_placement_controller_test.go index 83cc392..e65b679 100644 --- a/internal/controller/pod_placement_controller_test.go +++ b/internal/controller/pod_placement_controller_test.go @@ -174,10 +174,6 @@ func TestPlacement_UngatesAndRoutesAndCreatesClaim(t *testing.T) { if got.Spec.NodeSelector[nebulav1alpha1.ProviderLabel] != provider.ProviderModal { t.Fatalf("expected nodeSelector provider=modal, got %v", got.Spec.NodeSelector) } - // Capacity tier stamped for the VK handler. - if got.Annotations[nebulav1alpha1.CapacityTypeAnnotation] != "OnDemand" { - t.Fatalf("expected capacity-type OnDemand, got %q", got.Annotations[nebulav1alpha1.CapacityTypeAnnotation]) - } // NodeClaim created, pinned to the Pod, on the chosen provider. var nc nebulav1alpha1.NodeClaim if err := c.Get(context.Background(), types.NamespacedName{Name: "default-p1"}, &nc); err != nil { @@ -186,6 +182,11 @@ func TestPlacement_UngatesAndRoutesAndCreatesClaim(t *testing.T) { if nc.Spec.Provider != provider.ProviderModal || nc.Spec.PodRef.UID != "uid-1" || nc.Spec.PoolRef != "pool-a" { t.Fatalf("unexpected claim spec: %+v", nc.Spec) } + // Capacity tier recorded on the CLAIM, which is what the VK handler reads on + // CreatePod. Never on the Pod, where it would be patchable after the gate is gone. + if nc.Spec.CapacityType != nebulav1alpha1.CapacityOnDemand { + t.Fatalf("expected claim capacityType OnDemand, got %q", nc.Spec.CapacityType) + } // The request's POOL identity (type:count) is recorded for reporting: H100 with // no explicit count defaults to 1, so the pool is "H100:1". if nc.Spec.Accelerator != "H100:1" { @@ -442,9 +443,8 @@ func TestPlacement_FailsOverToNextRegionWhenBlocked(t *testing.T) { reconcilePod(t, r, "default", "p1") - got := getPod(t, c, "default", "p1") - if got.Annotations[nebulav1alpha1.RegionAnnotation] != "us-west-2" { - t.Fatalf("expected failover to us-west-2, got region %q", got.Annotations[nebulav1alpha1.RegionAnnotation]) + if got := getClaim(t, c, "default-p1").Spec.Region; got != "us-west-2" { + t.Fatalf("expected failover to us-west-2, got region %q", got) } } @@ -469,11 +469,11 @@ func TestPlacement_CapacityIsOuterAxis(t *testing.T) { reconcilePod(t, r, "default", "p1") - got := getPod(t, c, "default", "p1") - if got.Annotations[nebulav1alpha1.CapacityTypeAnnotation] != "OnDemand" { - t.Fatalf("expected the walk to drop to OnDemand, got %q", got.Annotations[nebulav1alpha1.CapacityTypeAnnotation]) + if got := getClaim(t, c, "default-p1").Spec.CapacityType; got != nebulav1alpha1.CapacityOnDemand { + t.Fatalf("expected the walk to drop to OnDemand, got %q", got) } // ...and to the FIRST provider (runpod), since OnDemand is walked provider-first. + got := getPod(t, c, "default", "p1") if got.Spec.NodeSelector[nebulav1alpha1.ProviderLabel] != "runpod" { t.Fatalf("expected first provider runpod at the OnDemand tier, got %v", got.Spec.NodeSelector) } @@ -481,9 +481,9 @@ func TestPlacement_CapacityIsOuterAxis(t *testing.T) { func TestPlacement_ExpandsRegionGroupIntoConcreteCandidates(t *testing.T) { // The pool declares a GROUP token, not a region. Placement must walk the concrete - // regions the provider expands it into — and must stamp a CONCRETE one on the Pod, - // never the token: RegionAnnotation feeds ProvisionRequest.Region, which the - // adapter turns into a regional API endpoint, and "us" is not one. + // regions the provider expands it into — and must record a CONCRETE one on the claim, + // never the token: the claim's region feeds ProvisionRequest.Region, which the adapter + // turns into a regional API endpoint, and "us" is not one. pod := gatedPod("p1", "default", "uid-1", "pool-a", "H100") pool := poolWithRegions("pool-a", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, provider.ProviderModal, "us") @@ -506,8 +506,7 @@ func TestPlacement_ExpandsRegionGroupIntoConcreteCandidates(t *testing.T) { reconcilePod(t, r, "default", "p1") - got := getPod(t, c, "default", "p1") - if region := got.Annotations[nebulav1alpha1.RegionAnnotation]; region != "us-west-2" { + if region := getClaim(t, c, "default-p1").Spec.Region; region != "us-west-2" { t.Fatalf("expected the group to expand and fail over to us-west-2, got %q", region) } } @@ -560,9 +559,8 @@ func TestPlacement_SkipsSpotWhenProviderHasNoSpotTier(t *testing.T) { if hasGateNamed(got) { t.Fatal("expected the Pod placed at the OnDemand tier") } - if got.Annotations[nebulav1alpha1.CapacityTypeAnnotation] != string(nebulav1alpha1.CapacityOnDemand) { - t.Fatalf("expected the Spot candidate skipped for OnDemand, got %q", - got.Annotations[nebulav1alpha1.CapacityTypeAnnotation]) + if tier := getClaim(t, c, "default-p1").Spec.CapacityType; tier != nebulav1alpha1.CapacityOnDemand { + t.Fatalf("expected the Spot candidate skipped for OnDemand, got %q", tier) } } @@ -629,6 +627,38 @@ func TestPlacement_CopiesNoEgressPolicyOntoThePod(t *testing.T) { } } +func TestPlacement_CopiesNothingNebulaOwnedOntoThePod(t *testing.T) { + pod := gatedPod("p1", "default", "uid-1", "pool-a", "H100") + pool := poolWithRegions("pool-a", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacitySpot}, + provider.ProviderModal, "us-east-1") + // A pool with something to say on every axis that used to ride the Pod. + pool.Spec.Egress = &nebulav1alpha1.EgressPolicy{Mode: nebulav1alpha1.EgressBlocked} + pool.Spec.Failover = &nebulav1alpha1.FailoverPolicy{BlocklistTTL: metav1.Duration{Duration: time.Hour}} + prov := &fakeProvider{name: provider.ProviderModal, gpus: []string{"H100"}, spot: true, egress: true} + r, c := newPlacementReconciler(t, []client.Object{pod, pool}, prov) + + reconcilePod(t, r, "default", "p1") + + got := getPod(t, c, "default", "p1") + if hasGateNamed(got) { + t.Fatal("expected the Pod placed") + } + // Swept by prefix rather than key by key: the claim is that NO Nebula-owned annotation is + // written, so a copy under a fresh spelling is the same bug and fails here too. + for k, v := range got.Annotations { + if strings.HasPrefix(k, nebulav1alpha1.GroupVersion.Group+"/") { + t.Errorf("annotation %s=%q was stamped on the Pod; a provisioning input on the Pod "+ + "is patchable between ungate and CreatePod", k, v) + } + } + // ...and the decision really is recorded, on the claim the handler reads. + nc := getClaim(t, c, "default-p1") + if nc.Spec.CapacityType != nebulav1alpha1.CapacitySpot || nc.Spec.Region != "us-east-1" { + t.Errorf("claim records tier %q region %q, want Spot/us-east-1", + nc.Spec.CapacityType, nc.Spec.Region) + } +} + func TestPlacement_RestrictedPoolStaysGatedWhenNoProviderEnforcesEgress(t *testing.T) { // The provider cannot enforce the policy, so there is no candidate and the Pod stays // visibly unplaceable. This is the whole point of the capability gate: placing it diff --git a/internal/controller/pod_placement_helpers.go b/internal/controller/pod_placement_helpers.go index ce68c36..3614bb1 100644 --- a/internal/controller/pod_placement_helpers.go +++ b/internal/controller/pod_placement_helpers.go @@ -354,14 +354,15 @@ func (r *PodPlacementReconciler) ensureClaim(ctx context.Context, pod *corev1.Po return false, nil // stale claim for a prior Pod; wait for the backstop } -// place stamps the routing decision onto the Pod and removes the gate, atomically -// from the Pod's perspective (one Update). After this, the scheduler is free to -// bind the Pod to the chosen provider's virtual node. +// place writes the one thing the Pod itself needs — the nodeSelector that routes it to the +// chosen provider's virtual node — and removes the gate, atomically from the Pod's +// perspective (one Update). After this, the scheduler is free to bind it. // -// It takes the DECISION, not the pool: nothing from the pool's own spec is copied onto the -// Pod any more. The egress policy and the failover TTL both used to be stamped here for the -// VK handler to read back, which made them patchable by whoever owns the Pod; the handler -// reads both from the NodePool at provision time instead. +// Nothing else about the decision goes on the Pod. The provisioning inputs (capacity tier, +// region) are already on the NodeClaim ensureClaim wrote a moment ago, and the pool's policy +// (egress, failover TTL) stays on the NodePool. All of it used to be stamped here for the VK +// handler to read back, which made every one of them patchable between ungate and CreatePod +// by whoever the policy constrains; the handler reads cluster state instead. func (r *PodPlacementReconciler) place(ctx context.Context, pod *corev1.Pod, p placement) error { // Route to the provider's virtual node. if pod.Spec.NodeSelector == nil { @@ -369,14 +370,6 @@ func (r *PodPlacementReconciler) place(ctx context.Context, pod *corev1.Pod, p p } pod.Spec.NodeSelector[nebulav1alpha1.ProviderLabel] = p.provider - // Carry the capacity tier and region the VK handler reads on CreatePod (inputs - // that are not otherwise on the Pod). Skip each when empty (provider default). - if p.capacityType != "" { - setAnnotation(pod, nebulav1alpha1.CapacityTypeAnnotation, string(p.capacityType)) - } - if p.region != "" { - setAnnotation(pod, nebulav1alpha1.RegionAnnotation, p.region) - } // Remove our gate, releasing the Pod to the scheduler. Preserve any other // gates a different controller may hold. pod.Spec.SchedulingGates = removeGate(pod.Spec.SchedulingGates, nebulav1alpha1.ProviderSelectionGate) @@ -384,14 +377,6 @@ func (r *PodPlacementReconciler) place(ctx context.Context, pod *corev1.Pod, p p return r.Update(ctx, pod) } -// setAnnotation sets one annotation on the Pod, allocating the map on first use. -func setAnnotation(pod *corev1.Pod, key, value string) { - if pod.Annotations == nil { - pod.Annotations = map[string]string{} - } - pod.Annotations[key] = value -} - // removeGate returns gates with the named gate removed, preserving order. func removeGate(gates []corev1.PodSchedulingGate, name string) []corev1.PodSchedulingGate { out := gates[:0] diff --git a/pkg/provider/modal/modal.go b/pkg/provider/modal/modal.go index 75f175e..cdaed01 100644 --- a/pkg/provider/modal/modal.go +++ b/pkg/provider/modal/modal.go @@ -261,8 +261,8 @@ func New(client Client, cat catalog.Lookup) *Provider { // regionSeparator joins several Modal regions into the ONE candidate placement // walks. It is deliberately a character no region name contains, so splitting is -// unambiguous, and deliberately not a comma: the value lands in RegionAnnotation and -// a comma reads like a list a consumer might re-split with different rules. +// unambiguous, and deliberately not a comma: the value is recorded on the NodeClaim, +// where a comma reads like a list a consumer might re-split with different rules. const regionSeparator = "|" // ExpandRegions implements provider.Provider, overriding catalog.Base's diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go index d59ad48..9917c3f 100644 --- a/pkg/provider/provider.go +++ b/pkg/provider/provider.go @@ -122,7 +122,7 @@ type Provider interface { // the provider's own error. Better than refusing a region that shipped last week. // // Expanding HERE, at the pool boundary, keeps everything downstream single-valued — - // ProvisionRequest.Region, RegionAnnotation and the blocklist key — so a capacity + // NodeClaimSpec.Region, ProvisionRequest.Region and the blocklist key — so a capacity // failure blocks the one candidate that failed, not the group it came from. // // How many candidates a declaration becomes depends on whether the provider can FAIL diff --git a/pkg/vnode/cluster.go b/pkg/vnode/cluster.go new file mode 100644 index 0000000..0ba00e3 --- /dev/null +++ b/pkg/vnode/cluster.go @@ -0,0 +1,136 @@ +/* +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" + "fmt" + "time" + + corev1 "k8s.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" +) + +// ClusterReader reads the two cluster-scoped objects every provision is derived from: +// +// NodePool - the pool's POLICY (the egress policy, the failover TTL). Admin-owned. +// NodeClaim - the placement DECISION for this Pod (capacity tier, region), recorded by +// the placement controller before it ungated. Controller-owned. +// +// Cluster-scoped is the point, not an incidental: it puts both objects out of reach of the +// namespaced RBAC a workload's owner holds. The Pod says what to run; these say under what +// policy and where. Anything the handler took off the Pod instead would be patchable by +// whoever the policy is meant to constrain — the Pod webhook runs on CREATE only, and +// placement stops watching a Pod once its gate is gone. +// +// Narrower than client.Reader — two methods, two types, no options — because that is the +// whole dependency: a fake in a test is a few lines, and no test needs a scheme. +type ClusterReader interface { + Pool(ctx context.Context, name string) (*nebulav1alpha1.NodePool, error) + Claim(ctx context.Context, name string) (*nebulav1alpha1.NodeClaim, error) +} + +// cachedClusterReader adapts the manager's client. +type cachedClusterReader struct{ reader client.Reader } + +// NewCachedClusterReader wraps a controller-runtime reader as a ClusterReader. Pass the +// manager's client: its cache is already synced before runnables start, and both informers +// are shared with the controllers that watch NodePools and NodeClaims, so the reads on the +// provisioning path cost no API call. +func NewCachedClusterReader(reader client.Reader) ClusterReader { + return &cachedClusterReader{reader: reader} +} + +func (c *cachedClusterReader) Pool(ctx context.Context, name string) (*nebulav1alpha1.NodePool, error) { + var pool nebulav1alpha1.NodePool + if err := c.reader.Get(ctx, client.ObjectKey{Name: name}, &pool); err != nil { + return nil, err + } + return &pool, nil +} + +func (c *cachedClusterReader) Claim(ctx context.Context, name string) (*nebulav1alpha1.NodeClaim, error) { + var claim nebulav1alpha1.NodeClaim + if err := c.reader.Get(ctx, client.ObjectKey{Name: name}, &claim); err != nil { + return nil, err + } + return &claim, nil +} + +// poolFor resolves the NodePool a Pod is placed against — the trusted source for the pool's +// own policy: the egress policy applied to the instance, and the TTL of any block a failed +// provision records. +// +// FAIL-CLOSED: no pool means no policy, and provisioning under an unknown policy is the +// failure this path exists to prevent. Callers treat the error as non-terminal, so a pool +// not yet in cache costs a retry rather than an unrestricted instance. +// +// Read ONCE per provision and passed down, so the egress policy applied and the TTL of any +// resulting block come from the same observation. The returned pool is shared informer +// state — read it, never mutate it. +func (h *Handler) poolFor(ctx context.Context, pod *corev1.Pod) (*nebulav1alpha1.NodePool, error) { + name := pod.Labels[nebulav1alpha1.PoolLabel] + if name == "" { + return nil, fmt.Errorf("no %s label on the Pod, so its pool policy cannot be established", + nebulav1alpha1.PoolLabel) + } + if h.cluster == nil { + // Wiring bug, not a user error: every production handler gets a reader (see + // NewRunner). Refusing keeps it a loud, immediate failure instead of a fleet that + // silently provisions unrestricted. + return nil, fmt.Errorf("no cluster reader configured, cannot establish the policy for pool %q", name) + } + pool, err := h.cluster.Pool(ctx, name) + if err != nil { + return nil, fmt.Errorf("read NodePool %q for its policy: %w", name, err) + } + return pool, nil +} + +// claimFor resolves the NodeClaim placement recorded for this Pod — the trusted source for +// the DECISION: which capacity tier and which region to provision in. +func (h *Handler) claimFor(ctx context.Context, pod *corev1.Pod, name string) (*nebulav1alpha1.NodeClaim, error) { + if h.cluster == nil { + return nil, fmt.Errorf("no cluster reader configured, cannot establish the placement for claim %q", name) + } + claim, err := h.cluster.Claim(ctx, name) + if err != nil { + return nil, fmt.Errorf("read NodeClaim %q for its placement decision: %w", name, err) + } + // The claim name is derived from namespace/name, which a recreated Pod reuses, so the + // UID is what proves this ledger is OURS. Placement already refuses to ungate against a + // stale claim, so this should never fire — and if it does, the alternative is + // provisioning in a region chosen for a different Pod. + if claim.Spec.PodRef.UID != string(pod.UID) { + return nil, fmt.Errorf("NodeClaim %q records Pod UID %q, not %q; refusing to provision against a stale ledger", + name, claim.Spec.PodRef.UID, pod.UID) + } + return claim, nil +} + +// blocklistTTLOf is the base exclusion a failed placement gets, from the pool's own +// FailoverPolicy. A non-positive TTL reads as unset, because zero would install a permanent +// block. Pure, and driven by the pool poolFor already returned, so the failure path never +// re-reads (and never has to decide what an unreadable pool means for a block). +func blocklistTTLOf(pool *nebulav1alpha1.NodePool) time.Duration { + if pool == nil || pool.Spec.Failover == nil || pool.Spec.Failover.BlocklistTTL.Duration <= 0 { + return defaultBlocklistTTL + } + return pool.Spec.Failover.BlocklistTTL.Duration +} diff --git a/pkg/vnode/doc.go b/pkg/vnode/doc.go index c2e6c65..07b50fe 100644 --- a/pkg/vnode/doc.go +++ b/pkg/vnode/doc.go @@ -20,10 +20,11 @@ limitations under the License. // // Ownership model ("VK owns provisioning"): the pod controller's CreatePod calls // provider.Provision and DeletePod calls provider.Terminate directly. The Pod is -// the single source of truth for the workload; the only provisioning input that -// is not on the Pod — the optimizer's capacity tier — rides on the -// CapacityTypeAnnotation, written by the placement controller when it ungates -// the Pod. Instance identity is derived deterministically from the Pod -// (ClaimName), so a provider whose List reports the claim tag can recover and -// reclaim an instance across a controller restart without a durable ledger. +// the single source of truth for the workload's SHAPE, and nothing more: every +// provisioning input that constrains rather than describes it is read from +// cluster state the workload's owner cannot patch — the NodePool for policy, the +// NodeClaim for the capacity tier and region placement chose (see ClusterReader). +// Instance identity is derived deterministically from the Pod (ClaimName), so a +// provider whose List reports the claim tag can recover and reclaim an instance +// across a controller restart without a durable ledger. package vnode diff --git a/pkg/vnode/env_test.go b/pkg/vnode/env_test.go index d6c54b6..c57f2f6 100644 --- a/pkg/vnode/env_test.go +++ b/pkg/vnode/env_test.go @@ -368,7 +368,7 @@ func TestCreatePod_PassesResolvedEnvToProvider(t *testing.T) { LocalObjectReference: corev1.LocalObjectReference{Name: "sec"}, Key: "K"}}}, }) client := fake.NewSimpleClientset(pod, secretObj("sec", map[string]string{"K": "t0ken"})) - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) if err := h.CreatePod(context.Background(), pod); err != nil { t.Fatalf("CreatePod: %v", err) @@ -393,7 +393,7 @@ func TestCreatePod_UnresolvableEnvIsNonTerminal(t *testing.T) { LocalObjectReference: corev1.LocalObjectReference{Name: "not-yet"}, Key: "K"}, }}}) client := fake.NewSimpleClientset(pod) - h := NewHandler(fp, client, bl, openPools()) + h := NewHandler(fp, client, bl, openCluster()) err := h.CreatePod(context.Background(), pod) if err == nil { diff --git a/pkg/vnode/exec_test.go b/pkg/vnode/exec_test.go index 20f2f1e..fac3bee 100644 --- a/pkg/vnode/exec_test.go +++ b/pkg/vnode/exec_test.go @@ -281,7 +281,7 @@ func TestRunExec_CancelReleasesTheProcess(t *testing.T) { func TestRunInContainer_RunsInTrackedInstance(t *testing.T) { proc := newExecProcess("root@sandbox:/#\n", "", 0) ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, proc) - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -311,7 +311,7 @@ func TestRunInContainer_RunsInTrackedInstance(t *testing.T) { func TestRunInContainer_IgnoresContainerName(t *testing.T) { for _, container := range []string{"", "main", "not-a-container"} { ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("ok\n", "", 0)) - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -328,7 +328,7 @@ func TestRunInContainer_IgnoresContainerName(t *testing.T) { func TestRunInContainer_NotFoundCases(t *testing.T) { // No exec support at all — a legitimate configuration (no agent, no key), not a bug. t.Run("provider does not support exec", func(t *testing.T) { - h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil, openPools()) + h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -340,7 +340,7 @@ func TestRunInContainer_NotFoundCases(t *testing.T) { // Another node's pod, or one this process never adopted. t.Run("pod not tracked", func(t *testing.T) { ep := newExecProvider(&fakeProvider{}, newExecProcess("", "", 0)) - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) err := h.RunInContainer(context.Background(), "default", "ghost", "main", []string{"sh"}, &fakeAttach{stdout: newSyncBuffer()}) assertNotFound(t, err) @@ -353,7 +353,7 @@ func TestRunInContainer_NotFoundCases(t *testing.T) { // nothing to run in. t.Run("tracked without an instance", func(t *testing.T) { ep := newExecProvider(&fakeProvider{provisionErr: errors.New("no capacity")}, newExecProcess("", "", 0)) - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err == nil { t.Fatal("CreatePod: expected the provision rejection to surface") } @@ -369,7 +369,7 @@ func TestRunInContainer_NotFoundCases(t *testing.T) { func TestRunInContainer_OnlyTheStartIsBounded(t *testing.T) { proc := newExecProcess("ok\n", "", 0) ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, proc) - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -395,7 +395,7 @@ func TestRunInContainer_OnlyTheStartIsBounded(t *testing.T) { func TestRunInContainer_StartErrorIsNotNotFound(t *testing.T) { ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("", "", 0)) ep.execErr = errors.New("timed out waiting for task id") - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -417,7 +417,7 @@ func TestRunInContainer_StartErrorIsNotNotFound(t *testing.T) { // provider — a provider that ran a default shell for it would be a surprise. func TestRunInContainer_EmptyCommandRejected(t *testing.T) { ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("", "", 0)) - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index c9d1178..a0f9298 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -91,11 +91,12 @@ type Handler struct { // no-op. blocklist Blocklister - // pools resolves the NodePool backing a Pod, which is where pool POLICY is read from - // rather than from the Pod carrying a copy of it. Unlike blocklist, nil is NOT a no-op: - // provisioning refuses without it, because the alternative is enforcing a containment - // policy nobody has read (see poolFor). - pools PoolReader + // cluster resolves the NodePool and NodeClaim behind a Pod: the pool's POLICY and + // placement's DECISION, both read from cluster state rather than from a copy carried on + // the Pod. Unlike blocklist, nil is NOT a no-op: provisioning refuses without it, + // because the alternative is enforcing a containment policy nobody has read (see + // poolFor and claimFor). + cluster ClusterReader mu sync.Mutex @@ -152,6 +153,24 @@ type trackedPod struct { // histogram under-samples the slowest boots. Fixing it means persisting the start // time, a write on the provisioning path we have not taken. provisionStart time.Time + + // placement is what this pod was provisioned against, so the poll loop can file its + // ready duration under the same dimensions as the provision counters. Set only where + // provisionStart is armed, which is the only path that observes. + placement +} + +// placement is the decision one provision was issued against: the two metric dimensions +// the Pod itself cannot supply, read from the NodeClaim at create (see claimFor). Carried +// per attempt rather than re-read at observation, because placement may write a newer +// decision onto the same claim on a re-provision, and because the poll loop holds h.mu +// while it observes — the NodeClaim stays the durable record, this is the historical one. +// +// The zero value means "unknown" and is what every path that never provisioned stores. +// Nothing is filed under it: those paths leave provisionStart zero, so they never observe. +type placement struct { + region string + tier nebulav1alpha1.CapacityType } // podMeta is the Pod metadata the virtual kubelet owns: the annotations it is the sole @@ -197,10 +216,10 @@ func (m podMeta) minus(done podMeta) podMeta { // NewHandler builds a Handler for one provider backend. The poll cadence comes from // Capabilities.PollInterval, falling back to defaultPollInterval. blocklist (failover -// recording) and client (the endpoint patch) may both be nil; pools may not, for anything -// that provisions — see poolFor. +// recording) and client (the endpoint patch) may both be nil; cluster may not, for anything +// that provisions — see poolFor and claimFor. func NewHandler( - prov provider.Provider, client kubernetes.Interface, blocklist Blocklister, pools PoolReader, + prov provider.Provider, client kubernetes.Interface, blocklist Blocklister, cluster ClusterReader, ) *Handler { poll := prov.Capabilities().PollInterval if poll <= 0 { @@ -210,7 +229,7 @@ func NewHandler( prov: prov, client: client, blocklist: blocklist, - pools: pools, + cluster: cluster, tracked: make(map[string]*trackedPod), nowFn: metav1.Now, pollEvery: poll, @@ -229,24 +248,22 @@ var ( func key(namespace, name string) string { return namespace + "/" + name } // CreatePod provisions an external instance for the Pod through the provider. -// The Pod carries the whole workload shape, and placement's choices (capacity tier, region) -// ride on annotations. The egress policy is the exception: it is read from the NodePool, -// because it constrains the Pod rather than describing it (see poolFor). +// +// The Pod carries the workload SHAPE (image, resources, env) and nothing else. Every +// provisioning input that constrains rather than describes it — the egress policy, the +// capacity tier, the region — comes from cluster state the workload's owner cannot write: +// the NodePool for policy, the NodeClaim for placement's decision (see poolFor, claimFor). func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { claim := util.ClaimName(pod.Namespace, pod.Name) - req := provider.ProvisionRequest{ - ClaimName: claim, - CapacityType: nebulav1alpha1.CapacityType(pod.Annotations[nebulav1alpha1.CapacityTypeAnnotation]), - Region: pod.Annotations[nebulav1alpha1.RegionAnnotation], - } + req := provider.ProvisionRequest{ClaimName: claim} log := logf.FromContext(ctx).WithName("vnode-handler").WithValues( "provider", h.prov.Name(), "pod", key(pod.Namespace, pod.Name), "claim", claim) - // The pool, which is the trusted source for every policy below: the containment policy - // here, and the TTL of any block this provision's failure records. Resolved before - // anything is requested and with the same non-terminal treatment as the env below, so a - // pool we cannot read costs a retry rather than an unrestricted instance. + // The pool: the containment policy applied here, and the TTL of any block this + // provision's failure records. Resolved before anything is requested and with the same + // non-terminal treatment as the env below, so a pool we cannot read costs a retry rather + // than an unrestricted instance. pool, err := h.poolFor(ctx, pod) if err != nil { log.Error(err, "cannot establish the pool's policy; nothing provisioned, retrying") @@ -256,6 +273,19 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { } req.Egress = pool.Spec.Egress + // The claim: WHERE and on WHAT TERMS placement decided to provision. Empty values here + // are legitimate ("the provider's default"), but a missing claim is not — a Pod on a + // virtual node without one never went through placement. + nc, err := h.claimFor(ctx, pod, claim) + if err != nil { + log.Error(err, "cannot establish the placement decision; nothing provisioned, retrying") + h.markStatus(pod, corev1.PodPending, reasonConfigError, err.Error()) + h.emit(pod) + return err + } + req.CapacityType = nc.Spec.CapacityType + req.Region = nc.Spec.Region + // Resolve BEFORE anything is requested: the Pod's env may point at Secrets and ConfigMaps // a provider cannot read (see resolveEnv), and nothing exists yet, so failing here is free. // @@ -302,7 +332,10 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // synchronous notify that can issue an API write, which would otherwise be charged // to the provider's latency. provisionStart := time.Now() - mlabels := h.metricLabels(pod) + // Carried into store so the poll loop's ready observation is filed under the same region + // and tier as the counters below, whatever the NodeClaim says by then. + place := placement{region: req.Region, tier: req.CapacityType} + labels := h.metricLabels(pod, place.region, place.tier) // Report Provisioning BEFORE the call: it can run for minutes (AWS sweeps a region's // zones on a capacity error), and until it returns this is the only explanation the @@ -315,7 +348,7 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // One call site for both outcomes, so the attempt and failure counters cannot drift. // An unreachable provider still counts as a failed attempt even though nothing gets // blocklisted for it. - metrics.ObserveProvision(mlabels, time.Since(callStart), err) + metrics.ObserveProvision(labels, time.Since(callStart), err) if err != nil { // An error the provider never attributed to this request — a transport failure, // our own timeout, a 503 — is not a rejection (see provider.IsRejection). The @@ -342,12 +375,13 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // → tier) instead of hot-looping here. The provider narrows its own error into a // BlockScope (a Spot shortage in one region blocks only that; auth/quota blocks // the whole provider); the TTL comes from the pool read at the top of this call. - h.recordBlock(ctx, pod, pool, err) + h.recordBlock(ctx, pod, pool, req.Region, err) // Surface the failure so placement can fail over, and return the error so the pod // controller retries with backoff. h.markStatus(pod, corev1.PodFailed, reasonProvisionFailed, err.Error()) - // Zero start: terminal, so it never reaches Running and has no ready-duration. - h.store(pod, claim, "", time.Time{}) + // Zero start: terminal, so it never reaches Running and has no ready-duration, which + // is also why the placement it would be filed under is not worth carrying. + h.store(pod, claim, "", time.Time{}, placement{}) h.emit(pod) return err } @@ -371,7 +405,7 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // tracked copy carries it, published by the emit below, re-offered every tick until a // write lands. Its reader is the NodeClaim controller (see InstanceIDAnnotation). setInstanceID(pod, res.InstanceID) - h.store(pod, claim, res.InstanceID, provisionStart) + h.store(pod, claim, res.InstanceID, provisionStart, place) // The TOKEN cannot ride the Pod (readable with `get pod`, unencrypted in etcd), so it // gets its own write — the only place it exists, since the provider mints it once and @@ -511,8 +545,9 @@ func (h *Handler) GetPod(ctx context.Context, namespace, name string) (*corev1.P pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}} applyState(pod, inst.State, inst.Endpoint, h.nowFn()) // Zero start: this process never provisioned it, so the real start time is gone and - // the ready-duration is not observable (see trackedPod.provisionStart). - h.store(pod, claim, inst.ID, time.Time{}) + // the ready-duration is not observable (see trackedPod.provisionStart) — hence no + // placement either, since nothing here will be filed under it. + h.store(pod, claim, inst.ID, time.Time{}, placement{}) log.Info("re-adopted live instance after cold tracking map (VK restart)", "claim", claim, "instanceID", inst.ID, "state", inst.State) return pod.DeepCopy(), nil @@ -677,12 +712,15 @@ func (h *Handler) reconcileOnce(ctx context.Context) { // a nonsense duration. // // ONE-SHOT — it consumes provisionStart, whose zero value covers both "never armed" and -// "already recorded" (see trackedPod.provisionStart). Callers must hold h.mu. +// "already recorded" (see trackedPod.provisionStart). That guard also keeps the poll loop +// cheap: labels are rendered behind it, so at most once per pod, never per 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)) + metrics.ObserveReady(h.metricLabels(tp.pod, tp.region, tp.tier), time.Since(tp.provisionStart)) tp.provisionStart = time.Time{} // spent; never observe this pod again } @@ -747,9 +785,12 @@ func statusSignature(pod *corev1.Pod) string { } // store records/updates the tracked pod under lock. provisionStart arms the -// ready-duration observation (see trackedPod.provisionStart); pass the zero Time from -// any path that cannot know it — a re-adoption, or an already-terminal pod. -func (h *Handler) store(pod *corev1.Pod, claim, instance string, provisionStart time.Time) { +// ready-duration observation (see trackedPod.provisionStart) and place is part of what that +// observation is filed under; pass the zero values from any path that cannot know them — +// a re-adoption, or an already-terminal pod. +func (h *Handler) store( + pod *corev1.Pod, claim, instance string, provisionStart time.Time, place placement, +) { h.mu.Lock() defer h.mu.Unlock() h.tracked[key(pod.Namespace, pod.Name)] = &trackedPod{ @@ -757,21 +798,25 @@ func (h *Handler) store(pod *corev1.Pod, claim, instance string, provisionStart claimName: claim, instance: instance, provisionStart: provisionStart, + placement: place, } } -// metricLabels renders the provisioning metric labels for a Pod, read off what placement -// stamped on it. +// metricLabels renders the provisioning metric labels for one provision. The shape comes +// off the Pod; where and on what terms it ran comes from the caller, which read it from the +// NodeClaim (see claimFor). // // Type and count stay SEPARATE here, unlike recordBlock, which files the joined pool key: // a blocklist needs one opaque key so an H100:8 shortage never excludes H100:1, while a // metric needs two dimensions to be aggregated either way. -func (h *Handler) metricLabels(pod *corev1.Pod) metrics.Labels { +func (h *Handler) metricLabels( + pod *corev1.Pod, region string, tier nebulav1alpha1.CapacityType, +) metrics.Labels { accel, count, _ := util.AcceleratorRequest(pod) return metrics.Labels{ Provider: h.prov.Name(), - Region: pod.Annotations[nebulav1alpha1.RegionAnnotation], - CapacityType: pod.Annotations[nebulav1alpha1.CapacityTypeAnnotation], + Region: region, + CapacityType: string(tier), Accelerator: accel, AcceleratorCount: count, } @@ -961,14 +1006,20 @@ func (h *Handler) markStatus(pod *corev1.Pod, phase corev1.PodPhase, reason, msg // from an instance-type shortage. The cost is one wasted re-probe by a sibling // accelerator, which is the right trade — over-broad would exclude serviceable // accelerators. DenyAll (auth/quota) ignores the accelerator: it fails for all. -func (h *Handler) recordBlock(ctx context.Context, pod *corev1.Pod, pool *nebulav1alpha1.NodePool, err error) { +func (h *Handler) recordBlock( + ctx context.Context, pod *corev1.Pod, pool *nebulav1alpha1.NodePool, region string, err error, +) { if h.blocklist == nil { return } - // Pool and region are properties of the REQUEST, not the error, so we resolve them off - // the Pod. The key is the POOL identity (type:count, e.g. "H100:8") — the same key - // placement queries a candidate by, since a block filed under any other key would - // never be read and failover would re-place onto the candidate that just failed. + // Accelerator and region are properties of the REQUEST, not the error: the accelerator + // is resolved off the Pod (it is the shape being asked for), the region comes in from the + // claim the request was built from — never from the Pod, since a block is process-wide + // and a patched region would let one tenant fence off a candidate for everyone. + // + // The key is the POOL identity (type:count, e.g. "H100:8") — the same key placement + // queries a candidate by, since a block filed under any other key would never be read + // and failover would re-place onto the candidate that just failed. // // The pool, NOT the provider's resolved SKU, because one launch may span several // interchangeable instance types (AWS fleets) and only fails when every one is dry, so @@ -977,7 +1028,6 @@ func (h *Handler) recordBlock(ctx context.Context, pod *corev1.Pod, pool *nebula // H100:1. "" means "not applicable" — a CPU-only Pod, or a region-simple provider. accel, count, _ := util.AcceleratorRequest(pod) accelerator := util.AcceleratorPool(accel, count) - region := pod.Annotations[nebulav1alpha1.RegionAnnotation] scope := h.prov.ClassifyProvisionError(err, accelerator, region) if scope == (provider.BlockScope{}) { diff --git a/pkg/vnode/handler_test.go b/pkg/vnode/handler_test.go index 634f581..52469e9 100644 --- a/pkg/vnode/handler_test.go +++ b/pkg/vnode/handler_test.go @@ -147,28 +147,39 @@ func (b *recordingBlocklist) Record(prov string, scope provider.BlockScope, ttl } // testPoolName is the pool every testPod is placed against. Provisioning resolves the -// pool by this label, so a Pod without it cannot be provisioned at all (see egressFor). +// pool by this label, so a Pod without it cannot be provisioned at all (see poolFor). const testPoolName = "pool-a" +// testPodUID is the UID every testPod carries, and the one the NodeClaim openCluster serves +// records. The two must agree: claimFor refuses a claim naming a different Pod incarnation, +// so a fake that ignored the UID would pass what the real reader rejects. +const testPodUID = "uid-1" + func testPod(ns, name string) *corev1.Pod { return &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ - Namespace: ns, Name: name, + Namespace: ns, Name: name, UID: testPodUID, Labels: map[string]string{nebulav1alpha1.PoolLabel: testPoolName}, }, Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "main", Image: "img"}}}, } } -// fakePools stands in for the manager's NodePool cache, serving pools from a map so a test -// can pin the policy the handler is supposed to read (and count that it read it). -type fakePools struct { +// fakeCluster stands in for the manager's cache of the two objects CreatePod provisions +// from, so a test can pin what the handler is supposed to find there — and count that it +// looked, instead of reading a copy off the Pod. +type fakeCluster struct { pools map[string]*nebulav1alpha1.NodePool - err error // when set, every Get fails with it - calls int + // claim is served under whatever name is asked for, since the name is derived from the + // Pod and every test Pod is placed the same way. Nil serves NotFound. + claim *nebulav1alpha1.NodeClaim + + err error // when set, every read fails with it + calls int // pool reads + claimCalls int } -func (f *fakePools) Get(_ context.Context, name string) (*nebulav1alpha1.NodePool, error) { +func (f *fakeCluster) Pool(_ context.Context, name string) (*nebulav1alpha1.NodePool, error) { f.calls++ if f.err != nil { return nil, f.err @@ -181,35 +192,66 @@ func (f *fakePools) Get(_ context.Context, name string) (*nebulav1alpha1.NodePoo return pool, nil } -// testPools serves testPoolName with the given policy; nil is an unrestricted pool, which -// is what a pool that never set spec.egress means. -func testPools(egress *nebulav1alpha1.EgressPolicy) *fakePools { - return &fakePools{pools: map[string]*nebulav1alpha1.NodePool{ - testPoolName: { - ObjectMeta: metav1.ObjectMeta{Name: testPoolName}, - Spec: nebulav1alpha1.NodePoolSpec{Egress: egress}, +func (f *fakeCluster) Claim(_ context.Context, name string) (*nebulav1alpha1.NodeClaim, error) { + f.claimCalls++ + if f.err != nil { + return nil, f.err + } + if f.claim == nil { + return nil, apierrors.NewNotFound( + schema.GroupResource{Group: nebulav1alpha1.GroupVersion.Group, Resource: "nodeclaims"}, name) + } + claim := f.claim.DeepCopy() + claim.Name = name + return claim, nil +} + +// clusterWithEgress serves testPoolName with the given policy; nil is an unrestricted pool, +// which is what a pool that never set spec.egress means. The claim it serves records no +// placement, the common case for a provider that has one tier and one region. +func clusterWithEgress(egress *nebulav1alpha1.EgressPolicy) *fakeCluster { + return &fakeCluster{ + pools: map[string]*nebulav1alpha1.NodePool{ + testPoolName: { + ObjectMeta: metav1.ObjectMeta{Name: testPoolName}, + Spec: nebulav1alpha1.NodePoolSpec{Egress: egress}, + }, }, - }} + claim: &nebulav1alpha1.NodeClaim{ + Spec: nebulav1alpha1.NodeClaimSpec{ + PodRef: nebulav1alpha1.PodReference{UID: testPodUID}, + }, + }, + } } -// openPools is the default for the many tests that provision without caring about egress. -func openPools() *fakePools { return testPools(nil) } +// openCluster is the default for the many tests that provision without caring about the +// pool's policy or where placement landed. +func openCluster() *fakeCluster { return clusterWithEgress(nil) } -// poolsWithTTL serves testPoolName with an explicit failover TTL and no egress policy, for -// the blocklist tests: the TTL is pool policy, read when a block is recorded. -func poolsWithTTL(ttl time.Duration) *fakePools { - pools := openPools() - pools.pools[testPoolName].Spec.Failover = &nebulav1alpha1.FailoverPolicy{ +// clusterWithTTL serves testPoolName with an explicit failover TTL, for the blocklist tests: +// the TTL is pool policy, read when a block is recorded. +func clusterWithTTL(ttl time.Duration) *fakeCluster { + c := openCluster() + c.pools[testPoolName].Spec.Failover = &nebulav1alpha1.FailoverPolicy{ BlocklistTTL: metav1.Duration{Duration: ttl}, } - return pools + return c +} + +// clusterWithPlacement serves a claim recording the decision placement made, which is where +// the tier and region a provision runs under come from. +func clusterWithPlacement(tier nebulav1alpha1.CapacityType, region string) *fakeCluster { + c := openCluster() + c.claim.Spec.CapacityType = tier + c.claim.Spec.Region = region + return c } func TestCreatePod_ProvisionsAndTracks(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, clusterWithPlacement(nebulav1alpha1.CapacitySpot, "")) pod := testPod("default", "p1") - pod.Annotations = map[string]string{nebulav1alpha1.CapacityTypeAnnotation: string(nebulav1alpha1.CapacitySpot)} if err := h.CreatePod(context.Background(), pod); err != nil { t.Fatalf("CreatePod: %v", err) @@ -221,7 +263,7 @@ func TestCreatePod_ProvisionsAndTracks(t *testing.T) { t.Fatalf("expected claim name default-p1, got %q", fp.lastReq.ClaimName) } if fp.lastReq.CapacityType != nebulav1alpha1.CapacitySpot { - t.Fatalf("expected capacity type read from annotation, got %q", fp.lastReq.CapacityType) + t.Fatalf("expected capacity type read from the claim, got %q", fp.lastReq.CapacityType) } got, err := h.GetPod(context.Background(), "default", "p1") @@ -238,7 +280,7 @@ func TestCreatePod_ReservedAdvancesToInitializing(t *testing.T) { // allocated) means the instance is committed and booting, so the Pod may leave // Provisioning immediately instead of waiting a whole poll tick for the same news. fp := &fakeProvider{provisionID: "inst-1", provisionReserved: true} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") if err := h.CreatePod(context.Background(), pod); err != nil { @@ -267,7 +309,7 @@ func TestCreatePod_UnreservedStaysProvisioning(t *testing.T) { // exactly true. Advancing to Initializing here would claim a commitment the // provider has not made. fp := &fakeProvider{provisionID: "sb-1", provisionReserved: false} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") if err := h.CreatePod(context.Background(), pod); err != nil { @@ -296,7 +338,7 @@ func TestCreatePod_EmitsProvisioningWhileProvisionInFlight(t *testing.T) { // Emitting an untracked Pod is what makes this safe: the tracked invariant forbids // storing one mid-provision, not reporting one. fp := &fakeProvider{provisionID: "inst-1", provisionReserved: true} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) var mu sync.Mutex var reasons []string @@ -337,7 +379,7 @@ func TestCreatePod_PollTickDuringProvisionEmitsNoTerminalStatus(t *testing.T) { // correct either way. The damage is the emit — VK writes it to the API server, where // Pod phases are terminal-sticky, and the NodeClaim then reclaims a live instance. fp := &fakeProvider{provisionID: "inst-1", provisionReserved: true} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) var mu sync.Mutex var emitted []string @@ -374,7 +416,7 @@ func TestCreatePod_PollTickDuringProvisionEmitsNoTerminalStatus(t *testing.T) { func TestCreatePod_ProvisionErrorSurfaces(t *testing.T) { fp := &fakeProvider{provisionErr: errors.New("no capacity")} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") if err := h.CreatePod(context.Background(), pod); err == nil { @@ -406,7 +448,7 @@ func TestCreatePod_UnattributableErrorLeavesPodProvisioning(t *testing.T) { classifyScope: provider.BlockScope{Accelerator: &accel}, } bl := &recordingBlocklist{} - h := NewHandler(fp, nil, bl, openPools()) + h := NewHandler(fp, nil, bl, openCluster()) var mu sync.Mutex var emitted []string @@ -456,7 +498,7 @@ func TestCreatePod_ProvisionFailureRecordsBlock(t *testing.T) { bl := &recordingBlocklist{} // A non-default TTL on the POOL, so this asserts the pool's policy is honored rather // than that it happens to equal defaultBlocklistTTL. - h := NewHandler(fp, nil, bl, poolsWithTTL(7*time.Minute)) + h := NewHandler(fp, nil, bl, clusterWithTTL(7*time.Minute)) h.jitterFn = func() time.Duration { return 0 } // pin jitter so the base TTL is asserted exactly pod := testPod("default", "p1") @@ -491,7 +533,7 @@ func TestCreatePod_EmptyScopeDoesNotBlock(t *testing.T) { // (which would exclude everything on the provider). fp := &fakeProvider{provisionErr: errors.New("weird"), classifyScope: provider.BlockScope{}} bl := &recordingBlocklist{} - h := NewHandler(fp, nil, bl, openPools()) + h := NewHandler(fp, nil, bl, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err == nil { t.Fatal("expected CreatePod to return the provision error") @@ -507,20 +549,20 @@ func TestCreatePod_EmptyScopeDoesNotBlock(t *testing.T) { // is a pure function of the pool rather than a second read that would have to decide. func TestCreatePod_BlocklistTTLFallsBackToDefault(t *testing.T) { for _, tc := range []struct { - name string - pools PoolReader + name string + cluster ClusterReader }{{ - name: "pool sets no failover policy", - pools: openPools(), + name: "pool sets no failover policy", + cluster: openCluster(), }, { // Zero would install a PERMANENT block, so it has to read as "unset". - name: "pool sets a zero TTL", - pools: poolsWithTTL(0), + name: "pool sets a zero TTL", + cluster: clusterWithTTL(0), }} { t.Run(tc.name, func(t *testing.T) { fp := &fakeProvider{provisionErr: errors.New("no capacity"), classifyScope: provider.BlockScope{DenyAll: true}} bl := &recordingBlocklist{} - h := NewHandler(fp, nil, bl, tc.pools) + h := NewHandler(fp, nil, bl, tc.cluster) h.jitterFn = func() time.Duration { return 0 } // pin jitter so the base default is exact if err := h.CreatePod(context.Background(), testPod("default", "p1")); err == nil { @@ -543,7 +585,7 @@ func TestCreatePod_BlocklistTTLFallsBackToDefault(t *testing.T) { func TestCreatePod_UnreadablePoolBlocksNothing(t *testing.T) { fp := &fakeProvider{provisionErr: errors.New("no capacity"), classifyScope: provider.BlockScope{DenyAll: true}} bl := &recordingBlocklist{} - h := NewHandler(fp, nil, bl, &fakePools{err: errors.New("cache not synced")}) + h := NewHandler(fp, nil, bl, &fakeCluster{err: errors.New("cache not synced")}) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err == nil { t.Fatal("expected CreatePod to fail closed on an unreadable pool") @@ -561,7 +603,7 @@ func TestCreatePod_UnreadablePoolBlocksNothing(t *testing.T) { func TestCreatePod_BlocklistTTLAddsJitter(t *testing.T) { fp := &fakeProvider{provisionErr: errors.New("no capacity"), classifyScope: provider.BlockScope{DenyAll: true}} bl := &recordingBlocklist{} - h := NewHandler(fp, nil, bl, poolsWithTTL(30*time.Second)) + h := NewHandler(fp, nil, bl, clusterWithTTL(30*time.Second)) h.jitterFn = func() time.Duration { return 20 * time.Second } // deterministic jitter if err := h.CreatePod(context.Background(), testPod("default", "p1")); err == nil { @@ -575,7 +617,7 @@ func TestCreatePod_BlocklistTTLAddsJitter(t *testing.T) { // The production jitter draw stays within [0, blocklistJitter) — never negative // (which would shorten the block below its base) and never at/over the bound. func TestProductionJitterInRange(t *testing.T) { - h := NewHandler(&fakeProvider{}, nil, nil, openPools()) + h := NewHandler(&fakeProvider{}, nil, nil, openCluster()) for i := 0; i < 1000; i++ { j := h.jitterFn() if j < 0 || j >= blocklistJitter { @@ -586,7 +628,7 @@ func TestProductionJitterInRange(t *testing.T) { func TestDeletePod_TerminatesAndUntracks(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") if err := h.CreatePod(context.Background(), pod); err != nil { @@ -608,7 +650,7 @@ func TestDeletePod_TerminatesAndUntracks(t *testing.T) { func TestDeletePod_Idempotent(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") _ = h.CreatePod(context.Background(), pod) @@ -622,7 +664,7 @@ func TestDeletePod_Idempotent(t *testing.T) { func TestReconcileOnce_ReportsRunning(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") _ = h.CreatePod(context.Background(), pod) @@ -669,7 +711,7 @@ func TestReconcileOnce_DNSEndpointNotWrittenToPodIP(t *testing.T) { // must be left empty. The reachable address is surfaced on the annotation // instead, which accepts any form. fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") _ = h.CreatePod(context.Background(), pod) @@ -710,7 +752,7 @@ func TestNotify_PersistsEndpointAnnotationOnce(t *testing.T) { }) fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) _ = h.CreatePod(context.Background(), pod) // Register the notify wrapper (this is where persistMetadata is injected). h.NotifyPods(context.Background(), func(*corev1.Pod) {}) @@ -760,7 +802,7 @@ func TestCreatePod_PersistsInstanceIDAlongsideEndpoint(t *testing.T) { }) fp := &fakeProvider{provisionID: "inst-1", provisionURL: url, provisionToken: "tok-abc"} - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) // Wrapper registered BEFORE the create, so the create-path emit is the write. h.NotifyPods(context.Background(), func(*corev1.Pod) {}) @@ -820,11 +862,10 @@ func secretValue(s *corev1.Secret, key string) string { func TestCreatePod_WritesConnectSecret(t *testing.T) { const url, token = "https://sb-1.modal.host", "tok-abc" pod := testPod("default", "p1") - pod.UID = "uid-1" client := fake.NewSimpleClientset(pod) fp := &fakeProvider{provisionID: "inst-1", provisionURL: url, provisionToken: token} - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) // Register the notifier BEFORE CreatePod, as VK does (it wires NotifyPods before // any pod work starts). The endpoint reaches the API server through it. h.NotifyPods(context.Background(), func(*corev1.Pod) {}) @@ -875,7 +916,6 @@ func TestCreatePod_WritesConnectSecret(t *testing.T) { // once and cannot be re-read, so no later tick has anything to write. func TestConnectSecret_WrittenOnceNotPerTick(t *testing.T) { pod := testPod("default", "p1") - pod.UID = "uid-1" client := fake.NewSimpleClientset(pod) var creates int @@ -889,7 +929,7 @@ func TestConnectSecret_WrittenOnceNotPerTick(t *testing.T) { provisionURL: "https://sb-1.modal.host", provisionToken: "tok-abc", } - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) if err := h.CreatePod(context.Background(), pod); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -919,11 +959,10 @@ func TestConnectSecret_WrittenOnceNotPerTick(t *testing.T) { // address is observed later rather than minted here. func TestCreatePod_NoSecretWithoutCredential(t *testing.T) { pod := testPod("default", "p1") - pod.UID = "uid-1" client := fake.NewSimpleClientset(pod) fp := &fakeProvider{provisionID: "inst-1"} // no URL, no token - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) if err := h.CreatePod(context.Background(), pod); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -939,11 +978,10 @@ func TestCreatePod_NoSecretWithoutCredential(t *testing.T) { func TestCreatePod_URLWithoutTokenPatchesEndpointOnly(t *testing.T) { const url = "https://sb-1.modal.host" pod := testPod("default", "p1") - pod.UID = "uid-1" client := fake.NewSimpleClientset(pod) fp := &fakeProvider{provisionID: "inst-1", provisionURL: url} - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) h.NotifyPods(context.Background(), func(*corev1.Pod) {}) if err := h.CreatePod(context.Background(), pod); err != nil { t.Fatalf("CreatePod: %v", err) @@ -965,10 +1003,11 @@ func TestCreatePod_URLWithoutTokenPatchesEndpointOnly(t *testing.T) { // would never be collected — leaking a live credential. Skip instead. func TestCreateConnectSecret_SkipsUIDLessPod(t *testing.T) { client := fake.NewSimpleClientset() - h := NewHandler(&fakeProvider{}, client, nil, openPools()) + h := NewHandler(&fakeProvider{}, client, nil, openCluster()) - h.createConnectSecret(context.Background(), testPod("default", "p1"), // no UID - "https://sb-9.modal.host", "tok-abc") + pod := testPod("default", "p1") + pod.UID = "" // the one thing an ownerReference cannot be built without + h.createConnectSecret(context.Background(), pod, "https://sb-9.modal.host", "tok-abc") if got := connectSecret(t, client, "default", "p1"); got != nil { t.Fatalf("expected no Secret for a UID-less Pod (it would never be GC'd), got %+v", got) @@ -979,11 +1018,10 @@ func TestCreateConnectSecret_SkipsUIDLessPod(t *testing.T) { // endpoint annotation would advertise an instance that does not exist. func TestCreatePod_NoCredentialPersistedOnProvisionFailure(t *testing.T) { pod := testPod("default", "p1") - pod.UID = "uid-1" client := fake.NewSimpleClientset(pod) fp := &fakeProvider{provisionErr: errors.New("no capacity")} - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) if err := h.CreatePod(context.Background(), pod); err == nil { t.Fatal("expected CreatePod to fail") } @@ -1006,11 +1044,10 @@ func TestCreatePod_NoCredentialPersistedOnProvisionFailure(t *testing.T) { func TestReconcileOnce_EmptyObservedEndpointDoesNotClearAnnotation(t *testing.T) { const url = "https://sb-1.modal.host" pod := testPod("default", "p1") - pod.UID = "uid-1" client := fake.NewSimpleClientset(pod) fp := &fakeProvider{provisionID: "inst-1", provisionURL: url, provisionToken: "tok-abc"} - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) if err := h.CreatePod(context.Background(), pod); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -1040,7 +1077,6 @@ func TestReconcileOnce_EmptyObservedEndpointDoesNotClearAnnotation(t *testing.T) func TestCreatePod_FailedEndpointPatchIsRetriedByPollLoop(t *testing.T) { const url = "https://sb-1.modal.host" pod := testPod("default", "p1") - pod.UID = "uid-1" client := fake.NewSimpleClientset(pod) var patches int @@ -1054,7 +1090,7 @@ func TestCreatePod_FailedEndpointPatchIsRetriedByPollLoop(t *testing.T) { }) fp := &fakeProvider{provisionID: "inst-1", provisionURL: url, provisionToken: "tok-abc"} - h := NewHandler(fp, client, nil, openPools()) + h := NewHandler(fp, client, nil, openCluster()) h.NotifyPods(context.Background(), func(*corev1.Pod) {}) if err := h.CreatePod(context.Background(), pod); err != nil { t.Fatalf("CreatePod must not fail on a failed endpoint patch: %v", err) @@ -1146,7 +1182,6 @@ func (s ctxRecordingSecrets) Create( func TestCreatePod_ProvisionDeadlineDoesNotLeakIntoCredentialWrite(t *testing.T) { const url, token = "https://sb-1.modal.host", "tok-abc" pod := testPod("default", "p1") - pod.UID = "uid-1" client := fake.NewSimpleClientset(pod) var secretCtx context.Context @@ -1158,7 +1193,7 @@ func TestCreatePod_ProvisionDeadlineDoesNotLeakIntoCredentialWrite(t *testing.T) // A tiny budget the call is guaranteed to exhaust. capabilities: provider.Capabilities{ProvisionTimeout: time.Millisecond}, } - h := NewHandler(fp, ctxRecordingClient{client, &secretCtx}, nil, openPools()) + h := NewHandler(fp, ctxRecordingClient{client, &secretCtx}, nil, openCluster()) h.NotifyPods(context.Background(), func(*corev1.Pod) {}) if err := h.CreatePod(context.Background(), pod); err != nil { t.Fatalf("CreatePod: %v", err) @@ -1200,7 +1235,7 @@ func TestReconcileOnce_NotifiesOnProvisioningToInitializing(t *testing.T) { // check would swallow this and strand the Pod on the stale "Provisioning" // reason; the reason must move and a notification must fire. fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") _ = h.CreatePod(context.Background(), pod) @@ -1243,7 +1278,7 @@ func TestReconcileOnce_NotifiesOnProvisioningToInitializing(t *testing.T) { func TestReconcileOnce_AbsentInstanceIsTerminated(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) pod := testPod("default", "p1") _ = h.CreatePod(context.Background(), pod) @@ -1270,7 +1305,7 @@ func TestReconcileOnce_ListErrorLeavesStatusUntouched(t *testing.T) { // recoverable ClaimName cannot be reported at all (an empty claim and an omitted // sandbox both read as absent here). That choice is only safe because of this. fp := &fakeProvider{provisionID: "inst-1", provisionReserved: true} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) _ = h.CreatePod(context.Background(), testPod("default", "p1")) var mu sync.Mutex @@ -1302,11 +1337,11 @@ func TestReconcileOnce_ListErrorLeavesStatusUntouched(t *testing.T) { func TestNewHandler_PollIntervalFromCapabilities(t *testing.T) { // A provider that declares a cadence overrides the default. custom := &fakeProvider{capabilities: provider.Capabilities{PollInterval: 5 * time.Second}} - if got := NewHandler(custom, nil, nil, openPools()).pollEvery; got != 5*time.Second { + if got := NewHandler(custom, nil, nil, openCluster()).pollEvery; got != 5*time.Second { t.Fatalf("expected the provider's PollInterval, got %v", got) } // A provider that leaves it zero falls back to the vnode default. - if got := NewHandler(&fakeProvider{}, nil, nil, openPools()).pollEvery; got != defaultPollInterval { + if got := NewHandler(&fakeProvider{}, nil, nil, openCluster()).pollEvery; got != defaultPollInterval { t.Fatalf("expected the default cadence, got %v", got) } } @@ -1321,7 +1356,7 @@ func TestGetPod_ReAdoptsLiveInstanceAfterRestart(t *testing.T) { fp := &fakeProvider{list: []provider.Instance{{ ID: "inst-9", ClaimName: "default-p1", State: provider.InstanceRunning, Endpoint: "1.2.3.4", }}} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) got, err := h.GetPod(context.Background(), "default", "p1") if err != nil { @@ -1348,7 +1383,7 @@ func TestGetPod_ReAdoptsLiveInstanceAfterRestart(t *testing.T) { // 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, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) got, err := h.GetPod(context.Background(), "default", "p1") if err == nil { @@ -1389,7 +1424,7 @@ func TestGetPod_UnknownClaimStaysNotFound(t *testing.T) { fp := &fakeProvider{list: []provider.Instance{{ ID: "inst-1", ClaimName: "default-other", State: provider.InstanceRunning, }}} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) if _, err := h.GetPod(context.Background(), "default", "p1"); !errdefs.IsNotFound(err) { t.Fatalf("expected NotFound for an unknown, unlisted claim, got %v", err) @@ -1398,7 +1433,7 @@ func TestGetPod_UnknownClaimStaysNotFound(t *testing.T) { func TestGetPods_ReturnsTracked(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, openCluster()) _ = h.CreatePod(context.Background(), testPod("default", "p1")) _ = h.CreatePod(context.Background(), testPod("default", "p2")) @@ -1440,7 +1475,7 @@ func TestCreatePod_ReadsEgressPolicyFromPool(t *testing.T) { }} { t.Run(tc.name, func(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} - pools := testPools(tc.pool) + pools := clusterWithEgress(tc.pool) h := NewHandler(fp, nil, nil, pools) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { @@ -1459,42 +1494,121 @@ func TestCreatePod_ReadsEgressPolicyFromPool(t *testing.T) { } } -// TestCreatePod_FailsClosedWithoutAPool covers the three ways the trusted state can be -// missing. None may fall back to a default: the pool IS the policy, so an unreadable pool -// means the policy is unknown, and provisioning under an unknown containment policy is the -// failure this whole path exists to prevent. Nothing is provisioned, and the Pod carries the -// reason rather than failing silently. -func TestCreatePod_FailsClosedWithoutAPool(t *testing.T) { +// TestCreatePod_ReadsPlacementFromClaim pins where the tier and region come FROM. Unlike the +// egress policy they cannot be recomputed from the pool — the pool declares an ordered +// fallback list, and which candidate won depends on the blocklist as it stood when placement +// ran — so the decision has to be CARRIED. It rides the NodeClaim, not the Pod: the claim is +// cluster-scoped and controller-written, while the Pod is patchable between ungate and here, +// where a patched tier bills the cluster owner OnDemand rates for a pool pinned to Spot and a +// patched region provisions outside a residency boundary. +func TestCreatePod_ReadsPlacementFromClaim(t *testing.T) { + fp := &fakeProvider{provisionID: "inst-1"} + cluster := clusterWithPlacement(nebulav1alpha1.CapacitySpot, "us-east-1") + h := NewHandler(fp, nil, nil, cluster) + + pod := testPod("default", "p1") + // Exactly what such a patch would look like, in the keys this used to read. Spelled out + // rather than referenced: the constants are gone, and nothing may resurrect them. + pod.Annotations = map[string]string{ + "nebula.inftyai.com/capacity-type": string(nebulav1alpha1.CapacityOnDemand), + "nebula.inftyai.com/region": "eu-central-1", + } + + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + if cluster.claimCalls != 1 { + t.Errorf("claim reads = %d, want 1; the placement must be read from the claim", cluster.claimCalls) + } + if fp.lastReq.CapacityType != nebulav1alpha1.CapacitySpot { + t.Errorf("req.CapacityType = %q, want Spot from the claim", fp.lastReq.CapacityType) + } + if fp.lastReq.Region != "us-east-1" { + t.Errorf("req.Region = %q, want us-east-1 from the claim", fp.lastReq.Region) + } +} + +// The region a failure is BLOCKLISTED under comes from the claim for a sharper reason: the +// blocklist is one process-wide map shared by every tenant, so a Pod-borne region would let +// whoever can patch a Pod fence a region off for everybody by failing a provision once. +func TestCreatePod_BlocklistRegionComesFromTheClaim(t *testing.T) { + fp := &fakeProvider{ + provisionErr: errors.New("no capacity"), + classifyScope: provider.BlockScope{DenyAll: true}, + } + bl := &recordingBlocklist{} + h := NewHandler(fp, nil, bl, clusterWithPlacement(nebulav1alpha1.CapacitySpot, "us-east-1")) + + pod := testPod("default", "p1") + pod.Annotations = map[string]string{"nebula.inftyai.com/region": "eu-central-1"} + + if err := h.CreatePod(context.Background(), pod); err == nil { + t.Fatal("expected CreatePod to return the provision error") + } + if bl.calls != 1 { + t.Fatalf("Record calls = %d, want 1", bl.calls) + } + if fp.classifyRegion != "us-east-1" { + t.Errorf("classified region = %q, want us-east-1 from the claim", fp.classifyRegion) + } +} + +// TestCreatePod_FailsClosedWithoutClusterState covers every way the two objects a provision +// is derived from can fail to resolve. None may fall back to a default: the pool IS the +// policy and the claim IS the placement decision, so provisioning without either means +// provisioning on terms nobody chose — the failure this whole path exists to prevent. +// Nothing is provisioned, and the Pod carries the reason rather than failing silently. +func TestCreatePod_FailsClosedWithoutClusterState(t *testing.T) { for _, tc := range []struct { - name string - pools PoolReader - pod func() *corev1.Pod + name string + cluster func() ClusterReader + pod func() *corev1.Pod }{{ // A Pod that never went through placement: a scheduling gate can be removed by // anyone who can patch the Pod, so reaching a virtual node proves nothing. - name: "no pool label", - pools: openPools(), + name: "no pool label", + cluster: func() ClusterReader { return openCluster() }, pod: func() *corev1.Pod { pod := testPod("default", "p1") delete(pod.Labels, nebulav1alpha1.PoolLabel) return pod }, }, { - name: "pool does not exist", - pools: &fakePools{}, - pod: func() *corev1.Pod { return testPod("default", "p1") }, + name: "pool does not exist", + cluster: func() ClusterReader { return &fakeCluster{} }, + pod: func() *corev1.Pod { return testPod("default", "p1") }, + }, { + name: "no reader wired", + cluster: func() ClusterReader { return nil }, + pod: func() *corev1.Pod { return testPod("default", "p1") }, }, { - name: "no reader wired", - pools: nil, - pod: func() *corev1.Pod { return testPod("default", "p1") }, + // The claim is the ledger placement writes BEFORE it ungates, so its absence means + // this Pod's provisioning terms were never recorded — or never decided. + name: "claim does not exist", + cluster: func() ClusterReader { + c := openCluster() + c.claim = nil + return c + }, + pod: func() *corev1.Pod { return testPod("default", "p1") }, + }, { + // A same-named claim from a PRIOR Pod incarnation. Its region and tier were chosen + // for a different workload, so honouring it would provision on someone else's terms. + name: "claim names a different Pod", + cluster: func() ClusterReader { + c := openCluster() + c.claim.Spec.PodRef.UID = "pod-uid-stale" + return c + }, + pod: func() *corev1.Pod { return testPod("default", "p1") }, }} { t.Run(tc.name, func(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, tc.pools) + h := NewHandler(fp, nil, nil, tc.cluster()) pod := tc.pod() if err := h.CreatePod(context.Background(), pod); err == nil { - t.Fatal("CreatePod succeeded; an unresolvable egress policy must refuse to provision") + t.Fatal("CreatePod succeeded; unresolvable cluster state must refuse to provision") } if fp.provisionCnt != 0 { t.Errorf("provision calls = %d, want 0; nothing may be requested", fp.provisionCnt) diff --git a/pkg/vnode/kubelet_test.go b/pkg/vnode/kubelet_test.go index aebf7a3..2a2726a 100644 --- a/pkg/vnode/kubelet_test.go +++ b/pkg/vnode/kubelet_test.go @@ -100,7 +100,7 @@ func TestKubeletServer_AdvertisesInternalIPOnly(t *testing.T) { // the provider's bytes. func TestKubeletServer_ServesLogsOverTLS(t *testing.T) { lp := newLoggingProvider(&fakeProvider{provisionID: "inst-1"}, "hello over tls\n") - h := NewHandler(lp, nil, nil, openPools()) + h := NewHandler(lp, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -137,7 +137,7 @@ func TestKubeletServer_ServesLogsOverTLS(t *testing.T) { // negotiated, and the command's output and exit code both survive the wire. func TestKubeletServer_ServesExecOverTLS(t *testing.T) { ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("hi from exec\n", "", 0)) - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -165,7 +165,7 @@ func TestKubeletServer_ServesExecOverTLS(t *testing.T) { // failed to the client, with its own status, not like a broken kubelet. func TestKubeletServer_ExecReportsExitCode(t *testing.T) { ep := newExecProvider(&fakeProvider{provisionID: "inst-1"}, newExecProcess("", "boom\n", 3)) - h := NewHandler(ep, nil, nil, openPools()) + h := NewHandler(ep, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -212,7 +212,7 @@ func execCommand(t *testing.T, base, namespace, pod string, cmd []string) (strin func TestKubeletServer_ResolvesAcrossProviders(t *testing.T) { modalish := newLoggingProvider(&fakeProvider{provisionID: "sb-1"}, "from modal\n") awsish := newLoggingProvider(&fakeProvider{provisionID: "i-1"}, "from aws\n") - hModal, hAWS := NewHandler(modalish, nil, nil, openPools()), NewHandler(awsish, nil, nil, openPools()) + hModal, hAWS := NewHandler(modalish, nil, nil, openCluster()), NewHandler(awsish, nil, nil, openCluster()) if err := hModal.CreatePod(context.Background(), testPod("default", "on-modal")); err != nil { t.Fatalf("CreatePod(modal): %v", err) } @@ -238,7 +238,7 @@ func TestKubeletServer_ResolvesAcrossProviders(t *testing.T) { func TestKubeletServer_ProviderErrorIsNotSwallowed(t *testing.T) { broken := newLoggingProvider(&fakeProvider{provisionID: "inst-1"}, "") broken.logsErr = fmt.Errorf("provider API unreachable") - h := NewHandler(broken, nil, nil, openPools()) + h := NewHandler(broken, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } diff --git a/pkg/vnode/logs_test.go b/pkg/vnode/logs_test.go index 8aa51c0..a5e7632 100644 --- a/pkg/vnode/logs_test.go +++ b/pkg/vnode/logs_test.go @@ -343,7 +343,7 @@ func (p *loggingProvider) Logs(_ context.Context, instanceID string) (io.ReadClo func TestGetContainerLogs_StreamsTrackedPod(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} lp := newLoggingProvider(fp, "hello from the sandbox\n") - h := NewHandler(lp, nil, nil, openPools()) + h := NewHandler(lp, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -370,7 +370,7 @@ func TestGetContainerLogs_StreamsTrackedPod(t *testing.T) { // stream behind it. func TestGetContainerLogs_IgnoresContainerName(t *testing.T) { lp := newLoggingProvider(&fakeProvider{provisionID: "inst-1"}, "out\n") - h := NewHandler(lp, nil, nil, openPools()) + h := NewHandler(lp, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -396,7 +396,7 @@ func TestGetContainerLogs_IgnoresContainerName(t *testing.T) { func TestGetContainerLogs_NotFoundCases(t *testing.T) { // No log support at all — a legitimate configuration, not an internal error. t.Run("provider does not stream logs", func(t *testing.T) { - h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil, openPools()) + h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -406,7 +406,7 @@ func TestGetContainerLogs_NotFoundCases(t *testing.T) { // Another node's pod, or one this process never adopted. t.Run("pod not tracked", func(t *testing.T) { - h := NewHandler(newLoggingProvider(&fakeProvider{}, "x\n"), nil, nil, openPools()) + h := NewHandler(newLoggingProvider(&fakeProvider{}, "x\n"), nil, nil, openCluster()) _, err := h.GetContainerLogs(context.Background(), "default", "ghost", "main", vkapi.ContainerLogOpts{}) assertNotFound(t, err) }) @@ -416,7 +416,7 @@ func TestGetContainerLogs_NotFoundCases(t *testing.T) { t.Run("tracked without an instance", func(t *testing.T) { fp := &fakeProvider{provisionErr: errors.New("no capacity")} lp := newLoggingProvider(fp, "x\n") - h := NewHandler(lp, nil, nil, openPools()) + h := NewHandler(lp, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err == nil { t.Fatal("CreatePod: expected the provision rejection to surface") } @@ -433,7 +433,7 @@ func TestGetContainerLogs_NotFoundCases(t *testing.T) { func TestGetContainerLogs_ProviderErrorIsNotNotFound(t *testing.T) { lp := newLoggingProvider(&fakeProvider{provisionID: "inst-1"}, "") lp.logsErr = errors.New("provider API unreachable") - h := NewHandler(lp, nil, nil, openPools()) + h := NewHandler(lp, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -454,7 +454,7 @@ func TestGetContainerLogs_ProviderErrorIsNotNotFound(t *testing.T) { // otherwise --tail/--limit-bytes would silently do nothing. func TestGetContainerLogs_AppliesOpts(t *testing.T) { lp := newLoggingProvider(&fakeProvider{provisionID: "inst-1"}, "a\nb\nc\n") - h := NewHandler(lp, nil, nil, openPools()) + h := NewHandler(lp, nil, nil, openCluster()) if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil { t.Fatalf("CreatePod: %v", err) } diff --git a/pkg/vnode/metrics_test.go b/pkg/vnode/metrics_test.go index b70b95e..0365882 100644 --- a/pkg/vnode/metrics_test.go +++ b/pkg/vnode/metrics_test.go @@ -56,18 +56,20 @@ func histCount(t *testing.T, h *prometheus.HistogramVec, l prometheus.Labels) ui 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. +// metricPod carries the half of the label set that comes off the Pod: the accelerator label +// the pool identity is derived from. Pair it with metricCluster, which supplies the other +// half — the tier and region, which are read from the NodeClaim, not the Pod. 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[nebulav1alpha1.AcceleratorTypeLabel] = "H100" return pod } +// metricCluster records the placement the labels below expect. +func metricCluster() *fakeCluster { + return clusterWithPlacement(nebulav1alpha1.CapacitySpot, "us-east-1") +} + // 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). @@ -90,7 +92,7 @@ func TestCreatePod_RecordsSuccessfulProvisionAttempt(t *testing.T) { beforeAttempts := testutil.ToFloat64(metrics.ProvisionAttempts.With(success)) beforeDuration := histCount(t, metrics.ProvisionDuration, success) - h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil, openPools()) + h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil, metricCluster()) if err := h.CreatePod(context.Background(), metricPod("default", "m1")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -112,7 +114,7 @@ func TestCreatePod_RecordsRejectionReason(t *testing.T) { beforeFailures := testutil.ToFloat64(metrics.ProvisionFailures.With(capacity)) fp := &fakeProvider{provisionErr: provider.ErrNoCapacity} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, metricCluster()) if err := h.CreatePod(context.Background(), metricPod("default", "m2")); err == nil { t.Fatal("expected the provision error") } @@ -138,7 +140,7 @@ func TestCreatePod_UnreachableProviderCountedSeparately(t *testing.T) { 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, openPools()) + h := NewHandler(fp, nil, nil, metricCluster()) if err := h.CreatePod(context.Background(), metricPod("default", "m3")); err == nil { t.Fatal("expected the provision error") } @@ -163,7 +165,7 @@ func TestReconcileOnce_ObservesReadyDurationExactlyOnce(t *testing.T) { before := histCount(t, metrics.InstanceReadyDuration, ready) fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, metricCluster()) if err := h.CreatePod(context.Background(), metricPod("default", "m4")); err != nil { t.Fatalf("CreatePod: %v", err) } @@ -195,10 +197,11 @@ func TestReconcileOnce_ObservesReadyDurationExactlyOnce(t *testing.T) { // 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. + // A re-adopted pod is tracked with no placement and a synthesized Pod carrying no + // accelerator request, so a wrongly taken observation would land on the series below: + // the provider is known, everything the lost provision knew reads "none". That, not the + // fully-labelled one, is the assertion carrying 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", @@ -212,7 +215,7 @@ func TestGetPod_ReAdoptedPodIsNotReadyObserved(t *testing.T) { fp := &fakeProvider{list: []provider.Instance{{ ID: "inst-9", ClaimName: "default-m5", State: provider.InstanceRunning, }}} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, metricCluster()) if _, err := h.GetPod(context.Background(), "default", "m5"); err != nil { t.Fatalf("GetPod: %v", err) } @@ -234,7 +237,7 @@ func TestObserveReady_IndependentOfPinnedStatusClock(t *testing.T) { before := histCount(t, metrics.InstanceReadyDuration, ready) fp := &fakeProvider{provisionID: "inst-1"} - h := NewHandler(fp, nil, nil, openPools()) + h := NewHandler(fp, nil, nil, metricCluster()) // 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 } diff --git a/pkg/vnode/node.go b/pkg/vnode/node.go index cf87c81..027c1ca 100644 --- a/pkg/vnode/node.go +++ b/pkg/vnode/node.go @@ -108,9 +108,10 @@ func NodeName(providerName string) string { // RBAC for the virtual kubelet: the pod controller reports Pod status and reads the // config/secret/service objects a Pod references; the node controller maintains the -// Node, its lease, and events. NodePools are read because pool policy is resolved from the -// pool at provision time rather than from the Pod (see Handler.poolFor). -// +kubebuilder:rbac:groups=nebula.inftyai.com,resources=nodepools,verbs=get;list;watch +// Node, its lease, and events. NodePools and NodeClaims are read because every provisioning +// input that is not the workload's own shape is resolved from cluster state at provision +// time rather than from the Pod (see Handler.poolFor and Handler.claimFor). +// +kubebuilder:rbac:groups=nebula.inftyai.com,resources=nodepools;nodeclaims,verbs=get;list;watch // +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;update;patch;delete // +kubebuilder:rbac:groups="",resources=pods/status,verbs=get;update;patch // +kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch;create;update;patch;delete @@ -131,7 +132,7 @@ type Runner struct { prov provider.Provider client kubernetes.Interface blocklist Blocklister - pools PoolReader + cluster ClusterReader nodeName string // kubelet is the shared endpoint serving `kubectl logs` for every node. Nil is @@ -141,18 +142,18 @@ type Runner struct { } // NewRunner builds the virtual-node runner for one provider. blocklist (Provision -// failures) and kubelet (the log endpoint) are both shared, and both may be nil. pools is -// how the handler reads pool policy from the pool instead of from the Pod, so a nil one -// leaves this node unable to provision at all (see Handler.poolFor). +// failures) and kubelet (the log endpoint) are both shared, and both may be nil. cluster is +// how the handler reads policy and placement from the NodePool and NodeClaim instead of from +// the Pod, so a nil one leaves this node unable to provision at all (see Handler.poolFor). func NewRunner( prov provider.Provider, client kubernetes.Interface, blocklist Blocklister, - kubelet *KubeletServer, pools PoolReader, + kubelet *KubeletServer, cluster ClusterReader, ) *Runner { return &Runner{ prov: prov, client: client, blocklist: blocklist, - pools: pools, + cluster: cluster, nodeName: NodeName(prov.Name()), kubelet: kubelet, } @@ -165,7 +166,7 @@ var _ manager.Runnable = (*Runner)(nil) func (r *Runner) Start(ctx context.Context) error { log := logf.FromContext(ctx).WithValues("virtualNode", r.nodeName, "provider", r.prov.Name()) - handler := NewHandler(r.prov, r.client, r.blocklist, r.pools) + handler := NewHandler(r.prov, r.client, r.blocklist, r.cluster) nodeSpec := nodeSpec(r.nodeName, r.prov.Name()) // Register on the endpoint AND advertise it, so the API server can proxy `kubectl diff --git a/pkg/vnode/pool.go b/pkg/vnode/pool.go deleted file mode 100644 index 42f2ccc..0000000 --- a/pkg/vnode/pool.go +++ /dev/null @@ -1,99 +0,0 @@ -/* -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" - "fmt" - "time" - - corev1 "k8s.io/api/core/v1" - "sigs.k8s.io/controller-runtime/pkg/client" - - nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" -) - -// PoolReader reads the NodePool a Pod is placed against. It exists so the handler can -// resolve pool POLICY from the pool itself, instead of trusting a copy of it carried on -// the Pod (see Handler.poolFor). -// -// Narrower than client.Reader — one method, one type, no options — because that is the -// whole dependency: a fake in a test is three lines, and no test needs a scheme. -type PoolReader interface { - Get(ctx context.Context, name string) (*nebulav1alpha1.NodePool, error) -} - -// cachedPoolReader adapts the manager's client. NodePool is cluster-scoped, so the name -// is the whole key. -type cachedPoolReader struct{ reader client.Reader } - -// NewCachedPoolReader wraps a controller-runtime reader as a PoolReader. Pass the -// manager's client: its cache is already synced before runnables start, and the pool -// informer is shared with the controllers that watch NodePools, so the read on the -// provisioning path costs no API call. -func NewCachedPoolReader(reader client.Reader) PoolReader { - return &cachedPoolReader{reader: reader} -} - -func (c *cachedPoolReader) Get(ctx context.Context, name string) (*nebulav1alpha1.NodePool, error) { - var pool nebulav1alpha1.NodePool - if err := c.reader.Get(ctx, client.ObjectKey{Name: name}, &pool); err != nil { - return nil, err - } - return &pool, nil -} - -// poolFor resolves the NodePool a Pod is placed against — the trusted source for every -// policy the handler needs to provision (the egress policy, the failover TTL), none of which -// may come off the Pod, where the workload's own owner can patch it. -// -// FAIL-CLOSED: no pool means no policy, and provisioning under an unknown policy is the -// failure this path exists to prevent. Callers treat the error as non-terminal, so a pool -// not yet in cache costs a retry rather than an unrestricted instance. -// -// Read ONCE per provision and passed down, so the egress policy applied and the TTL of any -// resulting block come from the same observation of the pool. The returned pool is shared -// informer state — read it, never mutate it. -func (h *Handler) poolFor(ctx context.Context, pod *corev1.Pod) (*nebulav1alpha1.NodePool, error) { - name := pod.Labels[nebulav1alpha1.PoolLabel] - if name == "" { - return nil, fmt.Errorf("no %s label on the Pod, so its pool policy cannot be established", - nebulav1alpha1.PoolLabel) - } - if h.pools == nil { - // Wiring bug, not a user error: every production handler gets a reader (see - // NewRunner). Refusing keeps it a loud, immediate failure instead of a fleet that - // silently provisions unrestricted. - return nil, fmt.Errorf("no NodePool reader configured, cannot establish the policy for pool %q", name) - } - pool, err := h.pools.Get(ctx, name) - if err != nil { - return nil, fmt.Errorf("read NodePool %q for its policy: %w", name, err) - } - return pool, nil -} - -// blocklistTTLOf is the base exclusion a failed placement gets, from the pool's own -// FailoverPolicy. A non-positive TTL reads as unset, because zero would install a permanent -// block. Pure, and driven by the pool poolFor already returned, so the failure path never -// re-reads (and never has to decide what an unreadable pool means for a block). -func blocklistTTLOf(pool *nebulav1alpha1.NodePool) time.Duration { - if pool == nil || pool.Spec.Failover == nil || pool.Spec.Failover.BlocklistTTL.Duration <= 0 { - return defaultBlocklistTTL - } - return pool.Spec.Failover.BlocklistTTL.Duration -} From 7a5cce98877a12fea3c3f08d8462b07ed2ffc1f1 Mon Sep 17 00:00:00 2001 From: kerthcet Date: Sat, 22 Aug 2026 16:05:14 +0100 Subject: [PATCH 2/3] simplify the recordBlock Signed-off-by: kerthcet --- pkg/vnode/handler.go | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index a0f9298..b1c55a9 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -374,8 +374,8 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // Record the failure so placement fails over to the next candidate (zone → region // → tier) instead of hot-looping here. The provider narrows its own error into a // BlockScope (a Spot shortage in one region blocks only that; auth/quota blocks - // the whole provider); the TTL comes from the pool read at the top of this call. - h.recordBlock(ctx, pod, pool, req.Region, err) + // the whole provider). + h.recordBlock(ctx, pod, req.Region, blocklistTTLOf(pool), err) // Surface the failure so placement can fail over, and return the error so the pod // controller retries with backoff. h.markStatus(pod, corev1.PodFailed, reasonProvisionFailed, err.Error()) @@ -993,9 +993,9 @@ func (h *Handler) markStatus(pod *corev1.Pod, phase corev1.PodPhase, reason, msg setPhase(pod, phase, reason, msg, h.nowFn()) } -// recordBlock classifies a Provision failure into a BlockScope and records it for the -// pool's BlocklistTTL, so placement fails over instead of retrying the same candidate. -// A no-op with no blocklist wired, or when the error yields an empty scope. +// recordBlock classifies a Provision failure into a BlockScope and records it for baseTTL +// plus jitter, so placement fails over instead of retrying the same candidate. A no-op with +// no blocklist wired, or when the error yields an empty scope. // // The provider owns the scope: the handler resolves the requested accelerator off the Pod // (the error does not carry it) and passes it in, but never assembles the scope itself, so @@ -1007,7 +1007,7 @@ func (h *Handler) markStatus(pod *corev1.Pod, phase corev1.PodPhase, reason, msg // accelerator, which is the right trade — over-broad would exclude serviceable // accelerators. DenyAll (auth/quota) ignores the accelerator: it fails for all. func (h *Handler) recordBlock( - ctx context.Context, pod *corev1.Pod, pool *nebulav1alpha1.NodePool, region string, err error, + ctx context.Context, pod *corev1.Pod, region string, baseTTL time.Duration, err error, ) { if h.blocklist == nil { return @@ -1036,10 +1036,10 @@ func (h *Handler) recordBlock( return } - // TTL = base (pool policy or default) + jitter, so Pods that failed for the SAME scope - // do not all re-probe the just-freed candidate at once. Coalescing keeps the latest - // expiry, so jittered records spread the shared deadline instead of pinning it. - ttl := blocklistTTLOf(pool) + h.jitterFn() + // baseTTL + jitter, so Pods that failed for the SAME scope do not all re-probe the + // just-freed candidate at once. Coalescing keeps the latest expiry, so jittered records + // spread the shared deadline instead of pinning it. + ttl := baseTTL + h.jitterFn() // The ctx logger, because it carries the virtualNode/provider values attached // upstream; a fresh context.Background() would fall back to the global delegate and // could be dropped before the real sink is installed. From 0ff2331b5f4fa6e95208e946c17f0d45c74dee10 Mon Sep 17 00:00:00 2001 From: kerthcet Date: Sat, 22 Aug 2026 20:49:10 +0100 Subject: [PATCH 3/3] Add region to Terminate Signed-off-by: kerthcet --- internal/controller/nodeclaim_controller.go | 4 +- .../controller/nodeclaim_controller_test.go | 23 +++++- pkg/provider/aws/aws.go | 69 +++-------------- pkg/provider/aws/aws_test.go | 44 +++++++---- pkg/provider/fake/fake.go | 2 +- pkg/provider/fake/fake_test.go | 7 +- pkg/provider/modal/modal.go | 6 +- pkg/provider/provider.go | 4 +- pkg/vnode/handler.go | 56 +++++++------- pkg/vnode/handler_test.go | 76 ++++++++++++++++--- pkg/vnode/metrics_test.go | 5 -- 11 files changed, 169 insertions(+), 127 deletions(-) diff --git a/internal/controller/nodeclaim_controller.go b/internal/controller/nodeclaim_controller.go index 884f7d7..325ca92 100644 --- a/internal/controller/nodeclaim_controller.go +++ b/internal/controller/nodeclaim_controller.go @@ -212,7 +212,9 @@ func (r *NodeClaimReconciler) reconcileDelete(ctx context.Context, nc *nebulav1a // already terminated this instance, so this is a redundant no-op; the call is // only load-bearing when DeletePod never ran. Idempotency also makes retries // after a transient error safe. - if err := prov.Terminate(ctx, id); err != nil { + // spec.Region is written before provisioning and never rewritten, so unlike the region + // VK holds in memory it is still readable in the case this backstop exists for. + if err := prov.Terminate(ctx, id, nc.Spec.Region); err != nil { log.Error(err, "terminate failed; will retry", "instanceID", id) return ctrl.Result{}, err } diff --git a/internal/controller/nodeclaim_controller_test.go b/internal/controller/nodeclaim_controller_test.go index 4a87e2a..696a16e 100644 --- a/internal/controller/nodeclaim_controller_test.go +++ b/internal/controller/nodeclaim_controller_test.go @@ -45,6 +45,7 @@ type fakeProvider struct { list []provider.Instance // what List returns listErr error // if set, List fails terminated []string // instance ids passed to Terminate, in order + regions []string // regions passed to Terminate, positionally paired with terminated terminateErr error // if set, Terminate fails gpus []string // accelerators MapAccelerator offers; nil = offer any spot bool // Capabilities().SupportsSpot (placement skips Spot without it) @@ -60,8 +61,9 @@ func (f *fakeProvider) Capabilities() provider.Capabilities { func (f *fakeProvider) Provision(context.Context, *corev1.Pod, provider.ProvisionRequest) (provider.ProvisionResult, error) { return provider.ProvisionResult{}, nil } -func (f *fakeProvider) Terminate(_ context.Context, id string) error { +func (f *fakeProvider) Terminate(_ context.Context, id, region string) error { f.terminated = append(f.terminated, id) + f.regions = append(f.regions, region) return f.terminateErr } func (f *fakeProvider) Get(context.Context, string) (*provider.Instance, error) { return nil, nil } @@ -478,6 +480,25 @@ func TestReconcileDelete_UsesRecordedInstanceID(t *testing.T) { } } +// The backstop runs when VK never did, so the region VK held in memory is gone too. It +// must pass spec.Region, written before provisioning and never rewritten — otherwise a +// region-partitioned provider has to search for the instance and can conclude "already +// gone" about one it never looked for (see provider.Terminate). +func TestReconcileDelete_PassesTheClaimRegion(t *testing.T) { + claim := newClaim("c1", "p1", "default", "uid-1", "fake") + claim.Spec.Region = "eu-west-1" + claim.Status.InstanceID = "inst-1" + deleteClaim(t, claim) + prov := &fakeProvider{name: "fake"} + r, _ := newClaimReconciler(t, []client.Object{claim}, prov) + + reconcileClaim(t, r, "c1") + + if len(prov.regions) != 1 || prov.regions[0] != "eu-west-1" { + t.Fatalf("regions passed to Terminate = %v, want [eu-west-1]", prov.regions) + } +} + func TestReconcileDelete_NoInstanceIsIdempotentNoOp(t *testing.T) { // The happy path: VK's DeletePod already terminated, so List finds nothing. // Terminate is called with "" (idempotent no-op) and the finalizer releases. diff --git a/pkg/provider/aws/aws.go b/pkg/provider/aws/aws.go index c6da5ba..20e8e4c 100644 --- a/pkg/provider/aws/aws.go +++ b/pkg/provider/aws/aws.go @@ -359,18 +359,9 @@ func (p *Provider) sweepRegions() []string { return out } -// clientFor returns the Client for region (defaulting an empty region), building -// and caching it on first use. Construction is serialized under mu: a region's -// Client is built at most once, and a concurrent caller for the same or another -// region waits — acceptable because a build is a rare one-time, per-region event -// (an AMI + subnet resolution), not a hot path. A build failure is NOT cached, so -// a transient resolution error (throttle, a not-yet-enabled region) is retried on -// the next call rather than poisoning the region permanently. +// clientFor returns the Client for region, building and caching it on first use. func (p *Provider) clientFor(ctx context.Context, region string) (Client, error) { if region == "" { - // No region to build a client for. Unreachable on the normal path (every - // request carries a region), so this only guards a legacy unqualified instance - // id reaching Terminate/Get — surface it rather than silently guessing. return nil, fmt.Errorf("aws: no region for client: %w", ErrConfig) } p.mu.Lock() @@ -511,30 +502,24 @@ func (p *Provider) Provision( return provider.ProvisionResult{InstanceID: id, Reserved: true}, nil } -// Terminate implements provider.Provider. Idempotent by the Client contract. The -// instanceID is a raw EC2 id, which does not carry its region, so teardown sweeps -// the swept regions and terminates the instance wherever it lives. This is safe -// and cheap: TerminateInstance is idempotent and a wrong-region lookup returns -// InvalidInstanceID.NotFound, which the Client maps to nil — so terminating in a -// region the instance is not in is a harmless no-op. It stops at the first region -// that actually owns the instance. +// Terminate implements provider.Provider. Idempotent by the Client contract. // -// A legacy region-qualified id ("/i-...") from before this change is still -// honored: splitID peels the region off and it routes straight to that region. -func (p *Provider) Terminate(ctx context.Context, instanceID string) error { +// An EC2 id is only reachable through its own region's endpoint, so the region picks the +// client — and terminating there needs no Describe first, since TerminateInstance is +// idempotent and maps InvalidInstanceID.NotFound to nil. +func (p *Provider) Terminate(ctx context.Context, instanceID, region string) error { if instanceID == "" { return nil // nothing provisioned yet; treat as already gone } - // Back-compat: a legacy qualified id routes directly to its region. - if region, rawID := splitID(instanceID); region != "" { + if region != "" { client, err := p.clientFor(ctx, region) if err != nil { return err } - return client.TerminateInstance(ctx, rawID) + return client.TerminateInstance(ctx, instanceID) } - // Raw id: sweep the regions and terminate wherever it lives. Confirm ownership + // Sweep the regions and terminate wherever it lives. Confirm ownership // with a Describe first so we only issue TerminateInstance against the region // that actually has it — and so a region whose client cannot be built does not // mask a successful terminate elsewhere. @@ -565,25 +550,7 @@ func (p *Provider) Terminate(ctx context.Context, instanceID string) error { // region's view of the instance. A per-region client-build/describe error is // tolerated and the sweep continues; only if every region errored (and none held // the instance) is that error surfaced. -// -// A legacy region-qualified id ("/i-...") routes directly to its region. func (p *Provider) Get(ctx context.Context, instanceID string) (*provider.Instance, error) { - if region, rawID := splitID(instanceID); region != "" { - client, err := p.clientFor(ctx, region) - if err != nil { - return nil, err - } - ec2, err := client.DescribeInstance(ctx, rawID) - if err != nil { - return nil, err - } - if ec2 == nil { - return nil, nil // absent => terminated, per interface contract - } - inst := p.toInstance(*ec2) - return &inst, nil - } - var lastErr error for _, region := range p.sweepRegions() { client, err := p.clientFor(ctx, region) @@ -726,24 +693,6 @@ func findByClaim(ctx context.Context, client Client, claimName string) (*EC2Inst return nil, nil } -// idSep separates the region prefix from the raw EC2 id in a legacy region-qualified -// instance id ("/"). "/" cannot appear in either a region name or -// an EC2 instance id, so it is an unambiguous delimiter. Current ids are raw EC2 -// ids; this exists only so splitID can still route ids recorded before that change. -const idSep = "/" - -// splitID recognizes a LEGACY region-qualified id ("/i-..."): it returns -// the region and the raw EC2 id. A current, raw EC2 id (no separator) yields an -// empty region, signaling the caller to locate the instance by sweeping regions -// instead. It is the one remaining reader of the old format, kept so ids persisted -// on a NodeClaim before the id stopped being qualified still terminate correctly. -func splitID(instanceID string) (region, rawID string) { - if i := strings.Index(instanceID, idSep); i >= 0 { - return instanceID[:i], instanceID[i+len(idSep):] - } - return "", instanceID -} - // instanceSpecFromPod reads the workload off the Pod (source of truth) and the // accelerator type (from the AcceleratorTypeLabel), maps it to an EC2 instance // type via the catalog, and stamps the claim tag, capacity tier, and region. diff --git a/pkg/provider/aws/aws_test.go b/pkg/provider/aws/aws_test.go index cb191a1..89081e7 100644 --- a/pkg/provider/aws/aws_test.go +++ b/pkg/provider/aws/aws_test.go @@ -478,34 +478,50 @@ func TestTerminate_Idempotent(t *testing.T) { p := newTestProvider(f) // Empty id: nothing was provisioned; treat as already gone (no client call). - if err := p.Terminate(context.Background(), ""); err != nil { + if err := p.Terminate(context.Background(), "", testRegion); err != nil { t.Fatalf("Terminate(\"\"): %v", err) } if len(f.terminated) != 0 { t.Fatalf("Terminate(\"\") called client, want no-op") } - // A raw EC2 id: Terminate sweeps regions, confirms the instance lives in one, and - // terminates it there. - if err := p.Terminate(context.Background(), "i-1"); err != nil { + // With a region: straight to that region's client. + if err := p.Terminate(context.Background(), "i-1", testRegion); err != nil { t.Fatalf("Terminate: %v", err) } if len(f.terminated) != 1 || f.terminated[0] != "i-1" { t.Fatalf("terminated = %v, want [i-1]", f.terminated) } + // Without one: sweep the regions, confirm the instance lives in one, terminate there. + if err := p.Terminate(context.Background(), "i-1", ""); err != nil { + t.Fatalf("Terminate(no region): %v", err) + } + if len(f.terminated) != 2 { + t.Fatalf("terminated = %v, want a second [i-1] from the sweep", f.terminated) + } } -// TestTerminate_LegacyQualifiedID covers the back-compat path: an id persisted in the -// old "/i-..." form (before the id stopped being region-qualified) still -// routes straight to that region and terminates the raw id, no sweep needed. -func TestTerminate_LegacyQualifiedID(t *testing.T) { - f := &fakeClient{} - p := newTestProvider(f) +// The region is what makes teardown reliable. An instance in a region the sweep does not +// visit — no pool declares it and nothing is in the client cache, the state of a freshly +// restarted process — is invisible to the sweep, which then reports success having +// terminated nothing and leaves the instance billing. The region reaches it regardless. +func TestTerminate_RegionOutsideTheSweep(t *testing.T) { + f := &fakeClient{ + instances: []EC2Instance{{ID: "i-2", State: stateRunning, Region: "eu-west-1"}}, + } + p := New( + func(context.Context, string) (Client, error) { return f, nil }, + fakeCatalog{}, + func() []string { return nil }, // no pool declares a region + ) - if err := p.Terminate(context.Background(), testRegion+"/i-legacy"); err != nil { - t.Fatalf("Terminate(legacy): %v", err) + if regions := p.sweepRegions(); len(regions) != 0 { + t.Fatalf("sweepRegions = %v, want none (the premise of this test)", regions) + } + if err := p.Terminate(context.Background(), "i-2", "eu-west-1"); err != nil { + t.Fatalf("Terminate: %v", err) } - if len(f.terminated) != 1 || f.terminated[0] != "i-legacy" { - t.Fatalf("terminated = %v, want [i-legacy]", f.terminated) + if len(f.terminated) != 1 || f.terminated[0] != "i-2" { + t.Fatalf("terminated = %v, want [i-2]", f.terminated) } } diff --git a/pkg/provider/fake/fake.go b/pkg/provider/fake/fake.go index 9811a2b..5aab5cd 100644 --- a/pkg/provider/fake/fake.go +++ b/pkg/provider/fake/fake.go @@ -126,7 +126,7 @@ func (p *Provider) Provision( // Terminate forgets the instance. Idempotent: terminating an already-gone (or // never-created) instance returns nil. -func (p *Provider) Terminate(_ context.Context, instanceID string) error { +func (p *Provider) Terminate(_ context.Context, instanceID, _ string) error { p.mu.Lock() defer p.mu.Unlock() delete(p.instances, instanceID) diff --git a/pkg/provider/fake/fake_test.go b/pkg/provider/fake/fake_test.go index 9d15153..7839c92 100644 --- a/pkg/provider/fake/fake_test.go +++ b/pkg/provider/fake/fake_test.go @@ -115,7 +115,8 @@ func TestTerminateIsIdempotent(t *testing.T) { t.Fatalf("Provision: %v", err) } id := res.InstanceID - if err := p.Terminate(ctx, id); err != nil { + // The fake keeps one flat map, so the region argument is irrelevant to it. + if err := p.Terminate(ctx, id, ""); err != nil { t.Fatalf("Terminate: %v", err) } // Gone from Get/List. @@ -123,10 +124,10 @@ func TestTerminateIsIdempotent(t *testing.T) { t.Fatalf("Get after Terminate = %v, want nil (terminated)", inst) } // A repeat Terminate (and terminating an unknown id) is a no-op, not an error. - if err := p.Terminate(ctx, id); err != nil { + if err := p.Terminate(ctx, id, ""); err != nil { t.Fatalf("repeat Terminate: %v", err) } - if err := p.Terminate(ctx, "never-existed"); err != nil { + if err := p.Terminate(ctx, "never-existed", ""); err != nil { t.Fatalf("Terminate unknown id: %v", err) } } diff --git a/pkg/provider/modal/modal.go b/pkg/provider/modal/modal.go index cdaed01..8c20709 100644 --- a/pkg/provider/modal/modal.go +++ b/pkg/provider/modal/modal.go @@ -370,8 +370,10 @@ func (p *Provider) Provision( }, nil } -// Terminate implements provider.Provider. Idempotent by the Client contract. -func (p *Provider) Terminate(ctx context.Context, instanceID string) error { +// Terminate implements provider.Provider. Idempotent by the Client contract. The region +// is ignored: Modal's API is global, and a sandbox id addresses it from anywhere (region +// is only ever a placement input, see Provision). +func (p *Provider) Terminate(ctx context.Context, instanceID, _ string) error { if instanceID == "" { return nil // nothing provisioned yet; treat as already gone } diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go index 9917c3f..35c121a 100644 --- a/pkg/provider/provider.go +++ b/pkg/provider/provider.go @@ -72,7 +72,9 @@ type Provider interface { // Terminate destroys the instance by id. Must be idempotent — terminating an // already-gone instance returns nil, so the finalizer that guarantees no paid instance // leaks can retry safely. - Terminate(ctx context.Context, instanceID string) error + // + // region is needed for providers that cannot infer it from the instance id (AWS) only. + Terminate(ctx context.Context, instanceID, region string) error // Get returns the current state of one instance, or (nil, nil) if it no // longer exists (treat absence as terminated). diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index b1c55a9..ceff1ad 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -154,20 +154,21 @@ type trackedPod struct { // time, a write on the provisioning path we have not taken. provisionStart time.Time - // placement is what this pod was provisioned against, so the poll loop can file its - // ready duration under the same dimensions as the provision counters. Set only where - // provisionStart is armed, which is the only path that observes. + // placement is what this pod was provisioned against. Two readers: the poll loop files + // the ready duration under the same dimensions as the provision counters, and DeletePod + // takes the region the instance is reachable in. Set only where provisionStart is armed. placement } -// placement is the decision one provision was issued against: the two metric dimensions -// the Pod itself cannot supply, read from the NodeClaim at create (see claimFor). Carried -// per attempt rather than re-read at observation, because placement may write a newer -// decision onto the same claim on a re-provision, and because the poll loop holds h.mu -// while it observes — the NodeClaim stays the durable record, this is the historical one. +// placement is the decision one provision was issued against: the two facts the Pod itself +// cannot supply, read from the NodeClaim at create (see claimFor). Kept per attempt rather +// than re-read where it is used, because the poll loop holds h.mu while it observes, and +// because the claim can be deleted while its instance is still tracked. The NodeClaim +// stays the durable record; this is the historical one. // -// The zero value means "unknown" and is what every path that never provisioned stores. -// Nothing is filed under it: those paths leave provisionStart zero, so they never observe. +// The zero value means "unknown" and is what every path that never provisioned stores. Its +// readers degrade rather than guess: no ready sample is filed (provisionStart is zero on +// those same paths), and teardown falls back to reading the claim. type placement struct { region string tier nebulav1alpha1.CapacityType @@ -468,17 +469,31 @@ func (h *Handler) UpdatePod(_ context.Context, pod *corev1.Pod) error { func (h *Handler) DeletePod(ctx context.Context, pod *corev1.Pod) error { h.mu.Lock() tp, ok := h.tracked[key(pod.Namespace, pod.Name)] - instance := "" + instance, region := "", "" if ok { instance = tp.instance + region = tp.region } h.mu.Unlock() log := logf.FromContext(ctx).WithName("vnode-handler").WithValues( "provider", h.prov.Name(), "pod", key(pod.Namespace, pod.Name), "instanceID", instance) - log.Info("terminating external instance") - if err := h.prov.Terminate(ctx, instance); err != nil { + // Only a pod THIS process provisioned has a tracked region; one re-adopted after a + // restart has none, so fall back to the NodeClaim, the durable record. Without a region + // the provider can only search for the instance (see provider.Terminate). + if region == "" { + claimName := util.ClaimName(pod.Namespace, pod.Name) + if claim, err := h.claimFor(ctx, pod, claimName); err != nil { + log.Info("no recorded region for teardown; the provider must search for the instance", + "claim", claimName, "reason", err.Error()) + } else { + region = claim.Spec.Region + } + } + + log.Info("terminating external instance", "region", region) + if err := h.prov.Terminate(ctx, instance, region); err != nil { // The leak-risk path: VK retries DeletePod, and if that never succeeds the // NodeClaim backstop is the last line of defense. Log loudly. log.Error(err, "terminate failed; external instance may still be running (NodeClaim backstop will retry)") @@ -1012,20 +1027,7 @@ func (h *Handler) recordBlock( if h.blocklist == nil { return } - // Accelerator and region are properties of the REQUEST, not the error: the accelerator - // is resolved off the Pod (it is the shape being asked for), the region comes in from the - // claim the request was built from — never from the Pod, since a block is process-wide - // and a patched region would let one tenant fence off a candidate for everyone. - // - // The key is the POOL identity (type:count, e.g. "H100:8") — the same key placement - // queries a candidate by, since a block filed under any other key would never be read - // and failover would re-place onto the candidate that just failed. - // - // The pool, NOT the provider's resolved SKU, because one launch may span several - // interchangeable instance types (AWS fleets) and only fails when every one is dry, so - // the pool truthfully names the request whichever alternate was tried. Distinct - // (type, count) pools stay on distinct keys, so an H100:8 shortage never excludes - // H100:1. "" means "not applicable" — a CPU-only Pod, or a region-simple provider. + accel, count, _ := util.AcceleratorRequest(pod) accelerator := util.AcceleratorPool(accel, count) scope := h.prov.ClassifyProvisionError(err, accelerator, region) diff --git a/pkg/vnode/handler_test.go b/pkg/vnode/handler_test.go index 52469e9..6b6d431 100644 --- a/pkg/vnode/handler_test.go +++ b/pkg/vnode/handler_test.go @@ -52,17 +52,18 @@ type fakeProvider struct { provisionReserved bool // provisionURL/provisionToken are the connect credential Provision returns — the // one-shot value the handler must persist, since it can never be re-read. - provisionURL string - provisionToken string - provisionErr error - provisionCnt int - lastReq provider.ProvisionRequest - terminateCnt int - terminateID string - terminateErr error - list []provider.Instance - listErr error - capabilities provider.Capabilities + provisionURL string + provisionToken string + provisionErr error + provisionCnt int + lastReq provider.ProvisionRequest + terminateCnt int + terminateID string + terminateRegion string + terminateErr error + list []provider.Instance + listErr error + capabilities provider.Capabilities // classifyScope is what ClassifyProvisionError returns for a failure; the zero // value (empty scope) means "not blocklistable". classifyAccel/classifyRegion // record what the handler passed in, so a test can assert it resolved them off the @@ -109,11 +110,12 @@ func (f *fakeProvider) Provision( }, nil } -func (f *fakeProvider) Terminate(_ context.Context, id string) error { +func (f *fakeProvider) Terminate(_ context.Context, id, region string) error { f.mu.Lock() defer f.mu.Unlock() f.terminateCnt++ f.terminateID = id + f.terminateRegion = region return f.terminateErr } @@ -662,6 +664,56 @@ func TestDeletePod_Idempotent(t *testing.T) { } } +// Teardown must name the region the instance was provisioned in: on a region-partitioned +// provider the id alone does not say which endpoint owns the instance. For a pod this +// process provisioned, that comes from the tracked placement — no re-read of the claim, +// which by then may already be deleted (the claim controller races the Pod's deletion). +func TestDeletePod_TerminatesInTheProvisionedRegion(t *testing.T) { + fp := &fakeProvider{provisionID: "inst-1"} + cluster := clusterWithPlacement(nebulav1alpha1.CapacitySpot, "eu-west-1") + h := NewHandler(fp, nil, nil, cluster) + pod := testPod("default", "p1") + + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + reads := cluster.claimCalls + if err := h.DeletePod(context.Background(), pod); err != nil { + t.Fatalf("DeletePod: %v", err) + } + if fp.terminateRegion != "eu-west-1" { + t.Fatalf("terminate region = %q, want eu-west-1", fp.terminateRegion) + } + if cluster.claimCalls != reads { + t.Fatalf("claim reads = %d, want %d (the tracked region is enough)", cluster.claimCalls, reads) + } +} + +// A pod re-adopted after a restart lost the tracked region with the process, so teardown +// reads it back off the NodeClaim — the durable record. Without it the provider would have +// to search for the instance, which can silently miss (see provider.Terminate). +func TestDeletePod_ReAdoptedPodReadsRegionFromClaim(t *testing.T) { + fp := &fakeProvider{list: []provider.Instance{{ + ID: "inst-9", ClaimName: "default-p1", State: provider.InstanceRunning, + }}} + h := NewHandler(fp, nil, nil, clusterWithPlacement(nebulav1alpha1.CapacitySpot, "ap-south-1")) + pod := testPod("default", "p1") + + // Cold map: GetPod re-adopts the live instance, with no placement to inherit. + if _, err := h.GetPod(context.Background(), "default", "p1"); err != nil { + t.Fatalf("GetPod: %v", err) + } + if err := h.DeletePod(context.Background(), pod); err != nil { + t.Fatalf("DeletePod: %v", err) + } + if fp.terminateID != "inst-9" { + t.Fatalf("terminate id = %q, want inst-9", fp.terminateID) + } + if fp.terminateRegion != "ap-south-1" { + t.Fatalf("terminate region = %q, want ap-south-1 (read from the claim)", fp.terminateRegion) + } +} + func TestReconcileOnce_ReportsRunning(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} h := NewHandler(fp, nil, nil, openCluster()) diff --git a/pkg/vnode/metrics_test.go b/pkg/vnode/metrics_test.go index 0365882..1429c7b 100644 --- a/pkg/vnode/metrics_test.go +++ b/pkg/vnode/metrics_test.go @@ -197,11 +197,6 @@ func TestReconcileOnce_ObservesReadyDurationExactlyOnce(t *testing.T) { // 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 tracked with no placement and a synthesized Pod carrying no - // accelerator request, so a wrongly taken observation would land on the series below: - // the provider is known, everything the lost provision knew reads "none". That, not the - // fully-labelled one, is the assertion carrying 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",