cleanup: remove egress annotation for safe concern

api/v1alpha1/groupversion_info.go

@@ -112,22 +112,6 @@ const (
 	// region-simple providers (Modal, RunPod) ignore it.
 	RegionAnnotation = "nebula.inftyai.com/region"
 
-	// BlocklistTTLAnnotation carries the pool's FailoverPolicy.BlocklistTTL down to the
-	// virtual kubelet. The TTL is NodePool policy, but the VK handler provisions per-Pod
-	// and never sees the pool, so it needs the value here to bound the block it records
-	// after a Provision failure. Stamped at ungate; absent/unparseable means the
-	// handler's built-in default.
-	BlocklistTTLAnnotation = "nebula.inftyai.com/blocklist-ttl"
-
-	// EgressAnnotation carries the pool's EgressPolicy.Mode and EgressTargetsAnnotation its
-	// Targets, comma-separated (no CIDR or hostname contains one). Same flow and
-	// reason as BlocklistTTLAnnotation: pool policy the VK handler must honour but never
-	// sees the pool to read.
-	//
-	// Absent means Open. Mode is authoritative — without it the target list is ignored.
-	EgressAnnotation        = "nebula.inftyai.com/egress"
-	EgressTargetsAnnotation = "nebula.inftyai.com/egress-targets"
-
 	// 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

api/v1alpha1/nodepool_types.go

@@ -82,13 +82,15 @@ type NodePoolSpec struct {
 // Allowlist disjoint, so "no egress" has one spelling instead of three.
 // +kubebuilder:validation:XValidation:rule="self.mode == 'Allowlist' || !has(self.targets)",message="targets is only valid with mode Allowlist"
 // +kubebuilder:validation:XValidation:rule="self.mode != 'Allowlist' || (has(self.targets) && self.targets.size() > 0)",message="mode Allowlist requires at least one target; use mode Blocked to permit nothing"
+// +kubebuilder:validation:XValidation:rule="!has(self.targets) || self.targets.all(t, !t.contains(','))",message="a target must not contain a comma; list each target as its own entry"
 type EgressPolicy struct {
 	// Mode is required once spec.egress is set, so a half-written policy is rejected
 	// rather than defaulted into a weaker one.
 	Mode EgressMode `json:"mode"`
 
 	// Targets is what mode Allowlist permits: CIDRs, bare IPs and domain names with an
 	// optional wildcard, mixed in one list, e.g. ["10.0.0.0/8", "*.huggingface.co"].
+	//
 	// +optional
 	// +kubebuilder:validation:MaxItems=64
 	// +kubebuilder:validation:items:MaxLength=253
@@ -216,6 +218,7 @@ type FailoverPolicy struct {
 	// (up to 30s) on top so Pods that failed for the same reason do not all retry the
 	// just-freed candidate in lockstep, so the effective exclusion is this value plus
 	// that jitter.
+	//
 	// +kubebuilder:default="30s"
 	BlocklistTTL metav1.Duration `json:"blocklistTTL,omitempty"`
 }

cmd/main.go

@@ -458,7 +458,10 @@ func setupVirtualNodes(mgr ctrl.Manager, blocklist vnode.Blocklister, kubeletSrv
 		if !ok {
 			continue
 		}
-		if err := mgr.Add(vnode.NewRunner(prov, clientset, blocklist, kubeletSrv)); err != nil {
+		// 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 {
 			return err
 		}
 		setupLog.Info("registered virtual node", "provider", name, "node", vnode.NodeName(name))

config/crd/bases/nebula.inftyai.com_nodepools.yaml

@@ -128,6 +128,9 @@ spec:
                     to permit nothing
                   rule: self.mode != 'Allowlist' || (has(self.targets) && self.targets.size()
                     > 0)
+                - message: a target must not contain a comma; list each target as
+                    its own entry
+                  rule: '!has(self.targets) || self.targets.all(t, !t.contains('',''))'
               failover:
                 description: |-
                   Failover controls how a provider that fails at provision time (e.g.

internal/controller/nodepool_validation_test.go

@@ -117,6 +117,60 @@ var _ = Describe("NodePool spec validation", func() {
 		Expect(err.Error()).To(ContainSubstring("Unsupported value: \"Reserved\""))
 	})
 
+	// The egress rules. The comma one is the load-bearing case: Targets rides to the VK
+	// handler as ONE comma-separated annotation, so an entry containing a comma would be
+	// decoded as several permitted targets rather than the single invalid one the author
+	// wrote — a policy wider than the pool declares, which is the wrong way for a
+	// containment control to fail. Admission is the only thing standing between the two.
+	newEgressPool := func(name string, egress *nebulav1alpha1.EgressPolicy) *nebulav1alpha1.NodePool {
+		return &nebulav1alpha1.NodePool{
+			ObjectMeta: metav1.ObjectMeta{Name: name},
+			Spec: nebulav1alpha1.NodePoolSpec{
+				Providers: []nebulav1alpha1.ProviderSpec{{Name: "modal"}},
+				Strategy:  nebulav1alpha1.StrategyOrdered,
+				Egress:    egress,
+			},
+		}
+	}
+
+	It("rejects a target containing a comma", func() {
+		pool := newEgressPool("egress-packed-target", &nebulav1alpha1.EgressPolicy{
+			Mode:    nebulav1alpha1.EgressAllowlist,
+			Targets: []string{"10.0.0.0/8,api.openai.com"},
+		})
+		err := k8sClient.Create(ctx, pool)
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("must not contain a comma"))
+	})
+
+	It("admits the same two targets listed separately", func() {
+		pool := newEgressPool("egress-split-targets", &nebulav1alpha1.EgressPolicy{
+			Mode:    nebulav1alpha1.EgressAllowlist,
+			Targets: []string{"10.0.0.0/8", "*.huggingface.co"},
+		})
+		Expect(k8sClient.Create(ctx, pool)).To(Succeed())
+		Expect(k8sClient.Delete(ctx, pool)).To(Succeed())
+	})
+
+	It("rejects targets without mode Allowlist", func() {
+		pool := newEgressPool("egress-blocked-with-targets", &nebulav1alpha1.EgressPolicy{
+			Mode:    nebulav1alpha1.EgressBlocked,
+			Targets: []string{"10.0.0.0/8"},
+		})
+		err := k8sClient.Create(ctx, pool)
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("targets is only valid with mode Allowlist"))
+	})
+
+	It("rejects mode Allowlist with no targets", func() {
+		pool := newEgressPool("egress-allowlist-empty", &nebulav1alpha1.EgressPolicy{
+			Mode: nebulav1alpha1.EgressAllowlist,
+		})
+		err := k8sClient.Create(ctx, pool)
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring("requires at least one target"))
+	})
+
 	It("admits the Spot and OnDemand capacity tiers", func() {
 		pool := &nebulav1alpha1.NodePool{
 			ObjectMeta: metav1.ObjectMeta{Name: "ok-capacity-types"},

internal/controller/pod_placement_controller.go

@@ -176,7 +176,7 @@ func (r *PodPlacementReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	}
 
 	// Stamp the routing decision and release the Pod to the scheduler.
-	if err := r.place(ctx, &pod, pool, placement); err != nil {
+	if err := r.place(ctx, &pod, placement); err != nil {
 		return ctrl.Result{}, err
 	}
 	// Counted only after the Update lands, which is also what keeps it from

internal/controller/pod_placement_controller_test.go

@@ -19,6 +19,7 @@ package controller
 import (
 	"context"
 	"slices"
+	"strings"
 	"testing"
 	"time"
 
@@ -597,10 +598,7 @@ func TestPlacement_SpotOnlyPoolStaysGatedOnOnDemandOnlyProvider(t *testing.T) {
 	}
 }
 
-func TestPlacement_StampsEgressAnnotationsFromPool(t *testing.T) {
-	// The VK handler never sees the pool, so the policy has to ride the Pod. Both
-	// annotations must land: the mode alone is authoritative, and without the targets
-	// an Allowlist pool would reach the adapter as "permit nothing" — silently Blocked.
+func TestPlacement_CopiesNoEgressPolicyOntoThePod(t *testing.T) {
 	pod := gatedPod("p1", "default", "uid-1", "pool-a", "H100")
 	pool := poolWith("pool-a", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, provider.ProviderModal)
 	pool.Spec.Egress = &nebulav1alpha1.EgressPolicy{
@@ -616,33 +614,18 @@ func TestPlacement_StampsEgressAnnotationsFromPool(t *testing.T) {
 	if hasGateNamed(got) {
 		t.Fatal("expected the Pod placed on a provider that enforces egress")
 	}
-	if v := got.Annotations[nebulav1alpha1.EgressAnnotation]; v != string(nebulav1alpha1.EgressAllowlist) {
-		t.Errorf("egress annotation = %q, want %q", v, nebulav1alpha1.EgressAllowlist)
-	}
-	if v := got.Annotations[nebulav1alpha1.EgressTargetsAnnotation]; v != "10.0.0.0/8,*.huggingface.co" {
-		t.Errorf("egress-targets annotation = %q, want the comma-joined list", v)
-	}
-}
-
-func TestPlacement_OpenPoolStampsNoEgressAnnotation(t *testing.T) {
-	// Absence IS Open (see EgressAnnotation), so an unrestricted pool must leave the Pod
-	// clean rather than stamping "Open" — otherwise every Pod in the cluster grows an
-	// annotation that means nothing.
-	pod := gatedPod("p1", "default", "uid-1", "pool-a", "H100")
-	pool := poolWith("pool-a", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, provider.ProviderModal)
-	// No egress on the pool at all, which is the common case.
-	prov := &fakeProvider{name: provider.ProviderModal, gpus: []string{"H100"}} // egress: false
-	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 an Open pool to place on a provider without egress support")
+	// Swept by substring rather than checked against the two retired keys, because the claim
+	// is that NO key carries the policy — a copy under a fresh spelling is the same bug and
+	// should fail here too.
+	for k, v := range got.Annotations {
+		if strings.Contains(k, "egress") {
+			t.Errorf("annotation %s=%q copies the pool's egress policy onto the Pod; the "+
+				"handler must read it from the pool", k, v)
+		}
 	}
-	if _, ok := got.Annotations[nebulav1alpha1.EgressAnnotation]; ok {
-		t.Errorf("expected no egress annotation for an Open pool, got %q",
-			got.Annotations[nebulav1alpha1.EgressAnnotation])
+	// The pool label is what the handler resolves the policy through, so it must be set.
+	if got.Labels[nebulav1alpha1.PoolLabel] != "pool-a" {
+		t.Errorf("pool label = %q, want pool-a", got.Labels[nebulav1alpha1.PoolLabel])
 	}
 }
 
@@ -764,23 +747,6 @@ func TestPlacement_NoServableCandidateDoesNotRequeue(t *testing.T) {
 	}
 }
 
-func TestPlacement_StampsBlocklistTTLAnnotation(t *testing.T) {
-	// The pool's FailoverPolicy.BlocklistTTL must reach the Pod so the VK handler
-	// knows how long to blocklist a placement that fails.
-	pod := gatedPod("p1", "default", "uid-1", "pool-a", "H100")
-	pool := poolWith("pool-a", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, provider.ProviderModal)
-	pool.Spec.Failover = &nebulav1alpha1.FailoverPolicy{BlocklistTTL: metav1.Duration{Duration: 7 * time.Minute}}
-	prov := &fakeProvider{name: provider.ProviderModal, gpus: []string{"H100"}}
-	r, c := newPlacementReconciler(t, []client.Object{pod, pool}, prov)
-
-	reconcilePod(t, r, "default", "p1")
-
-	got := getPod(t, c, "default", "p1")
-	if got.Annotations[nebulav1alpha1.BlocklistTTLAnnotation] != "7m0s" {
-		t.Fatalf("expected blocklist-ttl annotation 7m0s, got %q", got.Annotations[nebulav1alpha1.BlocklistTTLAnnotation])
-	}
-}
-
 // terminalOwnedPod builds an opted-in Pod in a terminal phase. When ownedByRS is
 // true it carries a controlling ReplicaSet ownerReference (so a controller would
 // recreate it); otherwise it is a bare Pod.

internal/controller/pod_placement_helpers.go

@@ -19,7 +19,6 @@ package controller
 import (
 	"context"
 	"hash/fnv"
-	"strings"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -358,7 +357,12 @@ func (r *PodPlacementReconciler) ensureClaim(ctx context.Context, pod *corev1.Po
 // 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.
-func (r *PodPlacementReconciler) place(ctx context.Context, pod *corev1.Pod, pool *nebulav1alpha1.NodePool, p placement) error {
+//
+// 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.
+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 {
 		pod.Spec.NodeSelector = map[string]string{}
@@ -373,22 +377,6 @@ func (r *PodPlacementReconciler) place(ctx context.Context, pod *corev1.Pod, poo
 	if p.region != "" {
 		setAnnotation(pod, nebulav1alpha1.RegionAnnotation, p.region)
 	}
-	// Carry the pool's blocklist TTL so the VK handler (which never sees the pool)
-	// knows how long to exclude a placement that fails. Only stamp an explicit
-	// policy value; an unset policy leaves the handler on its own default.
-	if pool.Spec.Failover != nil && pool.Spec.Failover.BlocklistTTL.Duration > 0 {
-		setAnnotation(pod, nebulav1alpha1.BlocklistTTLAnnotation, pool.Spec.Failover.BlocklistTTL.Duration.String())
-	}
-	// Same for the egress policy. Stamped only when it restricts something: the absence of
-	// the annotation IS Open, and selectPlacement has already ensured p.provider can enforce
-	// whatever is written here.
-	if pool.Spec.Egress.RestrictsEgress() {
-		setAnnotation(pod, nebulav1alpha1.EgressAnnotation, string(pool.Spec.Egress.ModeOrOpen()))
-		if len(pool.Spec.Egress.Targets) > 0 {
-			setAnnotation(pod, nebulav1alpha1.EgressTargetsAnnotation, strings.Join(pool.Spec.Egress.Targets, ","))
-		}
-	}
-
 	// 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)

pkg/provider/provider.go

@@ -235,7 +235,8 @@ type ProvisionRequest struct {
 	// that this provider can enforce it (Capabilities.SupportsEgressPolicy), so an adapter
 	// receiving a restrictive policy must apply it or fail the Provision — never silently
 	// drop it, which would leave the workload on the open internet under a policy that says
-	// otherwise. Read off the Pod's annotations, not the pool; see EgressAnnotation.
+	// otherwise. Resolved from the NodePool at provision time, never from the Pod, which
+	// the workload's own owner can patch (see vnode.Handler.egressFor).
 	Egress *nebulav1alpha1.EgressPolicy
 	// Env is the container's environment, fully RESOLVED: literals plus everything
 	// envFrom/valueFrom referenced, merged in kubelet precedence (envFrom in listed order,

pkg/vnode/env_test.go

@@ -247,7 +247,7 @@ func TestResolveEnv_MemoizesMisses(t *testing.T) {
 func TestResolveEnv_FieldRef(t *testing.T) {
 	pod := envPod(nil, nil)
 	pod.UID = "uid-1"
-	pod.Labels = map[string]string{"app": "vllm"}
+	pod.Labels["app"] = "vllm"
 	pod.Annotations = map[string]string{"team": "infra"}
 	pod.Spec.NodeName = "nebula-fake"
 	pod.Spec.ServiceAccountName = "sa"
@@ -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)
+	h := NewHandler(fp, client, nil, openPools())
 
 	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)
+	h := NewHandler(fp, client, bl, openPools())
 
 	err := h.CreatePod(context.Background(), pod)
 	if err == nil {

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)
+	h := NewHandler(ep, nil, nil, openPools())
 	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)
+		h := NewHandler(ep, nil, nil, openPools())
 		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)
+		h := NewHandler(&fakeProvider{provisionID: "inst-1"}, nil, nil, openPools())
 		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)
+		h := NewHandler(ep, nil, nil, openPools())
 		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)
+		h := NewHandler(ep, nil, nil, openPools())
 		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)
+	h := NewHandler(ep, nil, nil, openPools())
 	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)
+	h := NewHandler(ep, nil, nil, openPools())
 	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)
+	h := NewHandler(ep, nil, nil, openPools())
 	if err := h.CreatePod(context.Background(), testPod("default", "p1")); err != nil {
 		t.Fatalf("CreatePod: %v", err)
 	}