From 73e2dcb11fb453ea63368df79eafc534e2c6d0c7 Mon Sep 17 00:00:00 2001 From: kerthcet Date: Thu, 20 Aug 2026 23:20:24 +0100 Subject: [PATCH] feat: support egress policy Signed-off-by: kerthcet --- api/v1alpha1/groupversion_info.go | 9 ++ api/v1alpha1/nodepool_types.go | 59 ++++++++++++ api/v1alpha1/zz_generated.deepcopy.go | 25 +++++ .../bases/nebula.inftyai.com_nodepools.yaml | 34 +++++++ config/samples/nodepool.yaml | 16 ++++ .../controller/nodeclaim_controller_test.go | 3 +- .../pod_placement_controller_test.go | 77 ++++++++++++++++ internal/controller/pod_placement_helpers.go | 36 +++++++- pkg/metrics/placement.go | 1 + pkg/provider/aws/aws.go | 16 ++-- pkg/provider/fake/fake.go | 11 ++- pkg/provider/modal/client.go | 42 ++++++++- pkg/provider/modal/modal.go | 26 ++++-- pkg/provider/modal/modal_test.go | 92 +++++++++++++++++++ pkg/provider/provider.go | 16 +++- pkg/util/network.go | 45 +++++++++ pkg/util/network_test.go | 70 ++++++++++++++ pkg/vnode/handler.go | 24 +++++ pkg/vnode/handler_test.go | 54 +++++++++++ 19 files changed, 628 insertions(+), 28 deletions(-) create mode 100644 pkg/util/network.go create mode 100644 pkg/util/network_test.go diff --git a/api/v1alpha1/groupversion_info.go b/api/v1alpha1/groupversion_info.go index 0cf5b54..af8f440 100644 --- a/api/v1alpha1/groupversion_info.go +++ b/api/v1alpha1/groupversion_info.go @@ -119,6 +119,15 @@ const ( // 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 diff --git a/api/v1alpha1/nodepool_types.go b/api/v1alpha1/nodepool_types.go index aa6ff47..83861ed 100644 --- a/api/v1alpha1/nodepool_types.go +++ b/api/v1alpha1/nodepool_types.go @@ -69,6 +69,65 @@ type NodePoolSpec struct { // RunPod reports no capacity) is temporarily excluded and re-tried. // +optional Failover *FailoverPolicy `json:"failover,omitempty"` + + // Egress restricts OUTBOUND connections from this pool's workloads; omitted means + // Open. Inbound is never affected — a Blocked sandbox still serves its consumer's + // tunnel and connect token, it just cannot call out. + // + // +optional + Egress *EgressPolicy `json:"egress,omitempty"` +} + +// EgressPolicy is a pool's outbound network policy. The rules below keep Blocked and +// 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" +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 + Targets []string `json:"targets,omitempty"` +} + +// EgressMode is how a pool treats outbound traffic. No mode restricts inbound. +// +kubebuilder:validation:Enum=Open;Blocked;Allowlist +type EgressMode string + +const ( + // EgressOpen places no restriction, and is what an omitted spec.egress means. + EgressOpen EgressMode = "Open" + // EgressBlocked permits no outbound connection at all. + EgressBlocked EgressMode = "Blocked" + // EgressAllowlist permits EgressPolicy.Targets and nothing else. + EgressAllowlist EgressMode = "Allowlist" +) + +// ModeOrOpen reads a nil policy as Open, since an omitted spec.egress and an explicit +// Open are the same thing and no caller should nil-check for it. +func (p *EgressPolicy) ModeOrOpen() EgressMode { + if p == nil || p.Mode == "" { + return EgressOpen + } + return p.Mode +} + +// GetTargets reads Targets off a possibly-nil policy, for the same reason as ModeOrOpen. +func (p *EgressPolicy) GetTargets() []string { + if p == nil { + return nil + } + return p.Targets +} + +// RestrictsEgress reports whether the policy needs a provider to enforce anything. +func (p *EgressPolicy) RestrictsEgress() bool { + return p.ModeOrOpen() != EgressOpen } // ProviderSpec is one provider's entry in a pool: which provider, and the diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 2f20efa..632a0d5 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -26,6 +26,26 @@ import ( runtime "k8s.io/apimachinery/pkg/runtime" ) +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EgressPolicy) DeepCopyInto(out *EgressPolicy) { + *out = *in + if in.Targets != nil { + in, out := &in.Targets, &out.Targets + *out = make([]string, len(*in)) + copy(*out, *in) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EgressPolicy. +func (in *EgressPolicy) DeepCopy() *EgressPolicy { + if in == nil { + return nil + } + out := new(EgressPolicy) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *FailoverPolicy) DeepCopyInto(out *FailoverPolicy) { *out = *in @@ -211,6 +231,11 @@ func (in *NodePoolSpec) DeepCopyInto(out *NodePoolSpec) { *out = new(FailoverPolicy) **out = **in } + if in.Egress != nil { + in, out := &in.Egress, &out.Egress + *out = new(EgressPolicy) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new NodePoolSpec. diff --git a/config/crd/bases/nebula.inftyai.com_nodepools.yaml b/config/crd/bases/nebula.inftyai.com_nodepools.yaml index 197bcc9..b0c58f6 100644 --- a/config/crd/bases/nebula.inftyai.com_nodepools.yaml +++ b/config/crd/bases/nebula.inftyai.com_nodepools.yaml @@ -94,6 +94,40 @@ spec: type: string minItems: 1 type: array + egress: + description: |- + Egress restricts OUTBOUND connections from this pool's workloads; omitted means + Open. Inbound is never affected — a Blocked sandbox still serves its consumer's + tunnel and connect token, it just cannot call out. + properties: + mode: + description: |- + Mode is required once spec.egress is set, so a half-written policy is rejected + rather than defaulted into a weaker one. + enum: + - Open + - Blocked + - Allowlist + type: string + targets: + description: |- + 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"]. + items: + maxLength: 253 + type: string + maxItems: 64 + type: array + required: + - mode + type: object + x-kubernetes-validations: + - message: targets is only valid with mode Allowlist + rule: self.mode == 'Allowlist' || !has(self.targets) + - message: mode Allowlist requires at least one target; use mode Blocked + to permit nothing + rule: self.mode != 'Allowlist' || (has(self.targets) && self.targets.size() + > 0) failover: description: |- Failover controls how a provider that fails at provision time (e.g. diff --git a/config/samples/nodepool.yaml b/config/samples/nodepool.yaml index a6dcc25..f68529f 100644 --- a/config/samples/nodepool.yaml +++ b/config/samples/nodepool.yaml @@ -28,3 +28,19 @@ spec: strategy: Ordered failover: blocklistTTL: 30s + # Outbound network policy. Omitted (as here) means Open. Inbound is never restricted, so a + # sandbox stays reachable through its connect URL and token under every mode — it just + # cannot call out. + # + # Blocked permits nothing: + # egress: + # mode: Blocked + # + # Allowlist permits only what it lists — CIDRs, bare IPs and domain names (wildcards + # allowed) in ONE list; the adapter sorts them by kind. Use this when the workload pulls + # model weights at startup, which hangs under Blocked: + # egress: + # mode: Allowlist + # targets: + # - "*.huggingface.co" + # - 10.0.0.0/8 diff --git a/internal/controller/nodeclaim_controller_test.go b/internal/controller/nodeclaim_controller_test.go index f3ed23c..4a87e2a 100644 --- a/internal/controller/nodeclaim_controller_test.go +++ b/internal/controller/nodeclaim_controller_test.go @@ -48,13 +48,14 @@ type fakeProvider struct { terminateErr error // if set, Terminate fails gpus []string // accelerators MapAccelerator offers; nil = offer any spot bool // Capabilities().SupportsSpot (placement skips Spot without it) + egress bool // Capabilities().SupportsEgressPolicy (placement skips restricted pools without it) // expandRegions overrides ExpandRegions; nil = pass the declaration through. expandRegions func([]string) []string } func (f *fakeProvider) Name() string { return f.name } func (f *fakeProvider) Capabilities() provider.Capabilities { - return provider.Capabilities{SupportsSpot: f.spot} + return provider.Capabilities{SupportsSpot: f.spot, SupportsEgressPolicy: f.egress} } func (f *fakeProvider) Provision(context.Context, *corev1.Pod, provider.ProvisionRequest) (provider.ProvisionResult, error) { return provider.ProvisionResult{}, nil diff --git a/internal/controller/pod_placement_controller_test.go b/internal/controller/pod_placement_controller_test.go index e2d372a..23f5a62 100644 --- a/internal/controller/pod_placement_controller_test.go +++ b/internal/controller/pod_placement_controller_test.go @@ -597,6 +597,83 @@ 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. + pod := gatedPod("p1", "default", "uid-1", "pool-a", "H100") + pool := poolWith("pool-a", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, provider.ProviderModal) + pool.Spec.Egress = &nebulav1alpha1.EgressPolicy{ + Mode: nebulav1alpha1.EgressAllowlist, + Targets: []string{"10.0.0.0/8", "*.huggingface.co"}, + } + prov := &fakeProvider{name: provider.ProviderModal, gpus: []string{"H100"}, 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 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") + } + if _, ok := got.Annotations[nebulav1alpha1.EgressAnnotation]; ok { + t.Errorf("expected no egress annotation for an Open pool, got %q", + got.Annotations[nebulav1alpha1.EgressAnnotation]) + } +} + +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 + // anyway would provision a workload with open internet access under a pool that says + // Blocked, with nothing to reveal the substitution. No requeue hint — only a pool edit + // or a provider gaining support fixes it, and both emit their own event. + pod := gatedPod("p1", "default", "uid-1", "pool-a", "H100") + pool := poolWith("pool-a", []nebulav1alpha1.CapacityType{nebulav1alpha1.CapacityOnDemand}, provider.ProviderModal) + pool.Spec.Egress = &nebulav1alpha1.EgressPolicy{Mode: nebulav1alpha1.EgressBlocked} + prov := &fakeProvider{name: provider.ProviderModal, gpus: []string{"H100"}} // egress: false + r, c := newPlacementReconciler(t, []client.Object{pod, pool}, prov) + + res, err := r.Reconcile(context.Background(), reconcile.Request{ + NamespacedName: types.NamespacedName{Namespace: "default", Name: "p1"}, + }) + if err != nil { + t.Fatalf("reconcile: %v", err) + } + if res.RequeueAfter != 0 { + t.Fatalf("expected no requeue hint for a capability gap, got %v", res.RequeueAfter) + } + + got := getPod(t, c, "default", "p1") + if !hasGateNamed(got) { + t.Fatal("expected the Pod to stay gated when no provider can enforce the egress policy") + } +} + func TestPlacement_AllCandidatesBlockedRequeuesForBlockExpiry(t *testing.T) { // Every (tier, provider, region) candidate is blocked (DenyAll on the provider), // but the candidates are servable — the block is a transient failover exclusion. diff --git a/internal/controller/pod_placement_helpers.go b/internal/controller/pod_placement_helpers.go index ea4698c..288987d 100644 --- a/internal/controller/pod_placement_helpers.go +++ b/internal/controller/pod_placement_helpers.go @@ -19,6 +19,7 @@ package controller import ( "context" "hash/fnv" + "strings" "time" corev1 "k8s.io/api/core/v1" @@ -116,12 +117,18 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev "provider", ref.Name, "capacityType", tier) continue // unregistered; NodePool status surfaces this separately } - if !servesCapacity(prov, tier) { + if !servesCapacityTier(prov, tier) { metrics.RecordCandidateSkip(ref.Name, tier, "", metrics.SkipCapacityUnsupported) log.V(1).Info("skipping candidate: provider does not offer the capacity tier", "provider", ref.Name, "capacityType", tier) continue } + if !servesEgress(prov, pool.Spec.Egress) { + metrics.RecordCandidateSkip(ref.Name, tier, "", metrics.SkipEgressUnsupported) + log.V(1).Info("skipping candidate: provider cannot enforce the pool's egress policy", + "provider", ref.Name, "egressMode", pool.Spec.Egress.ModeOrOpen()) + continue + } // A CPU-only Pod (no accelerator) matches any provider; an accelerator // Pod only matches a provider whose catalog serves that (type, count). // MapAccelerator is consulted only for that servability check — the block @@ -205,7 +212,7 @@ func capacityTiers(pool *nebulav1alpha1.NodePool) []nebulav1alpha1.CapacityType return pool.Spec.CapacityTypes } -// servesCapacity reports whether prov can deliver the candidate's capacity tier. Only Spot +// servesCapacityTier reports whether prov can deliver the candidate's capacity tier. Only Spot // is ever refused: an OnDemand-only provider (Modal) has no interruptible tier, so placing a // Spot candidate there would stamp CapacityType=Spot on the Pod, hand it to an adapter that // drops the field, and bill OnDemand rates for capacity the user asked to be cheap — with no @@ -214,13 +221,27 @@ func capacityTiers(pool *nebulav1alpha1.NodePool) []nebulav1alpha1.CapacityType // leaves the Pod visibly unplaceable rather than quietly overcharged. // // The empty tier is "the provider's default", which every provider serves, so it passes. -func servesCapacity(prov provider.Provider, tier nebulav1alpha1.CapacityType) bool { +func servesCapacityTier(prov provider.Provider, tier nebulav1alpha1.CapacityType) bool { if tier != nebulav1alpha1.CapacitySpot { return true } return prov.Capabilities().SupportsSpot } +// servesEgress reports whether prov can enforce the pool's egress policy. Open needs no +// enforcement, so every provider serves it; anything else needs SupportsEgressPolicy. +// +// Same reasoning as servesCapacity, and load-bearing for a different reason: a provider +// that drops the field would put the workload on the open internet while the pool claims +// containment. Skipping makes that visible — an AWS-only pool asking for Blocked leaves the +// Pod unplaceable instead of silently unprotected. +func servesEgress(prov provider.Provider, policy *nebulav1alpha1.EgressPolicy) bool { + if !policy.RestrictsEgress() { + return true + } + return prov.Capabilities().SupportsEgressPolicy +} + // regionsFor is the inner axis for one provider ref: the concrete regions to try, in // expansion order. The pool's declaration is a CONSTRAINT, not a list of regions — // it may be omitted (unconstrained), name a geography group ("us"), or name regions @@ -358,6 +379,15 @@ func (r *PodPlacementReconciler) place(ctx context.Context, pod *corev1.Pod, poo 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. diff --git a/pkg/metrics/placement.go b/pkg/metrics/placement.go index 6266186..9cc5c4d 100644 --- a/pkg/metrics/placement.go +++ b/pkg/metrics/placement.go @@ -51,6 +51,7 @@ const ( SkipProviderUnregistered = "provider_unregistered" SkipCapacityUnsupported = "capacity_type_unsupported" SkipAcceleratorUnsupported = "accelerator_unsupported" + SkipEgressUnsupported = "egress_policy_unsupported" SkipBlocked = "blocked" ) diff --git a/pkg/provider/aws/aws.go b/pkg/provider/aws/aws.go index 3288a7f..c6da5ba 100644 --- a/pkg/provider/aws/aws.go +++ b/pkg/provider/aws/aws.go @@ -390,12 +390,16 @@ func (p *Provider) clientFor(ctx context.Context, region string) (Client, error) // trait is set the way it is. func (p *Provider) Capabilities() provider.Capabilities { return provider.Capabilities{ - SupportsStop: true, // EC2 instances stop/start - SupportsSpot: true, // real interruptible tier - NativeTags: true, // EC2 tags carry identity - PreemptionNotice: preemptionNotice, // Spot 2-minute warning - PollInterval: spotPollInterval, // Spot reclaims are abrupt; poll faster than default - ProvisionTimeout: provisionTimeout, // caps the per-zone capacity failover loop + SupportsStop: true, // EC2 instances stop/start + SupportsSpot: true, // real interruptible tier + // Instances launch into the default VPC, whose security group allows all egress and + // which routes to an internet gateway. Enforcing a pool's policy means managing SG + // egress rules (and no NAT for the Blocked case), so it is unsupported until then. + SupportsEgressPolicy: false, + NativeTags: true, // EC2 tags carry identity + PreemptionNotice: preemptionNotice, // Spot 2-minute warning + PollInterval: spotPollInterval, // Spot reclaims are abrupt; poll faster than default + ProvisionTimeout: provisionTimeout, // caps the per-zone capacity failover loop } } diff --git a/pkg/provider/fake/fake.go b/pkg/provider/fake/fake.go index 1be4282..9811a2b 100644 --- a/pkg/provider/fake/fake.go +++ b/pkg/provider/fake/fake.go @@ -73,11 +73,12 @@ func New() *Provider { // paths. func (p *Provider) Capabilities() provider.Capabilities { return provider.Capabilities{ - SupportsStop: false, - SupportsSpot: false, - NativeTags: true, - PreemptionNotice: 0, - PollInterval: 0, // use the vnode default cadence + SupportsStop: false, + SupportsSpot: false, + SupportsEgressPolicy: false, // nothing to enforce against; egress pools skip it + NativeTags: true, + PreemptionNotice: 0, + PollInterval: 0, // use the vnode default cadence } } diff --git a/pkg/provider/modal/client.go b/pkg/provider/modal/client.go index dbcbae6..fb374a0 100644 --- a/pkg/provider/modal/client.go +++ b/pkg/provider/modal/client.go @@ -32,8 +32,10 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/util/intstr" + nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" "github.com/InftyAI/Nebula/pkg/provider" "github.com/InftyAI/Nebula/pkg/provider/catalog" + "github.com/InftyAI/Nebula/pkg/util" ) // sdkClient is the real Client, backed by Modal's official Go SDK (beta). Every @@ -123,6 +125,11 @@ func (c *sdkClient) CreateSandbox(ctx context.Context, spec SandboxSpec) (string return "", Credential{}, fmt.Errorf("modal: readiness probe: %w", err) } + outCIDRs, outDomains, err := outboundAllowlists(spec) + if err != nil { + return "", Credential{}, err + } + sb, err := c.mc.Sandboxes.Create(ctx, app, image, &modal.SandboxCreateParams{ Command: spec.Command, // Env is the whole environment, including values resolved from this cluster's @@ -136,10 +143,17 @@ func (c *sdkClient) CreateSandbox(ctx context.Context, spec SandboxSpec) (string EncryptedPorts: spec.Ports, // Nil leaves Modal's SchedulerPlacement unset entirely (the SDK only builds one // when Regions is non-empty), which is the unconstrained, un-multiplied case. - Regions: spec.Regions, - Timeout: spec.Timeout, - Tags: spec.Tags, - ReadinessProbe: probe, + Regions: spec.Regions, + // Egress policy. Non-nil selects Modal's ALLOWLIST mode and the entries are then the + // complete permitted set, so an empty pair blocks all outbound traffic; nil leaves + // the sandbox OPEN. Both are set together whenever either is, because + // UpdateNetworkPolicy requires both and a later policy change should need no + // reshaping here. + OutboundCIDRAllowlist: outCIDRs, + OutboundDomainAllowlist: outDomains, + Timeout: spec.Timeout, + Tags: spec.Tags, + ReadinessProbe: probe, }) if err != nil { return "", Credential{}, err @@ -705,6 +719,26 @@ func (c *sdkClient) forgetReady(id string) { delete(c.waiting, id) } +// outboundAllowlists renders the spec's egress policy as Modal's two outbound allowlists. +// Both nil means OPEN. Both non-nil selects ALLOWLIST, where the entries are the COMPLETE +// permitted set — so the empty pair is what blocks everything, and nil vs empty is the whole +// signal (see SandboxCreateParams.OutboundCIDRAllowlist). +// +// An unrecognized mode fails the Provision rather than defaulting to open. +func outboundAllowlists(spec SandboxSpec) (cidrs, domains *modal.Allowlist, err error) { + switch spec.EgressMode { + case "", nebulav1alpha1.EgressOpen: + return nil, nil, nil + case nebulav1alpha1.EgressBlocked: + return &modal.Allowlist{}, &modal.Allowlist{}, nil + case nebulav1alpha1.EgressAllowlist: + c, d := util.SplitEgressTargets(spec.EgressTargets) + return &modal.Allowlist{Entries: c}, &modal.Allowlist{Entries: d}, nil + default: + return nil, nil, fmt.Errorf("modal: unsupported egress mode %q", spec.EgressMode) + } +} + // gpuReservation renders Modal's GPU reservation string. Modal expresses count // as a "type:count" suffix (e.g. "A100:2"); a count of 0/1 needs no suffix, and // an empty type means a CPU-only sandbox. diff --git a/pkg/provider/modal/modal.go b/pkg/provider/modal/modal.go index 1ef4055..75f175e 100644 --- a/pkg/provider/modal/modal.go +++ b/pkg/provider/modal/modal.go @@ -155,6 +155,14 @@ type SandboxSpec struct { Timeout time.Duration // Tags carry Nebula identity; ClaimTagKey holds the NodeClaim name. Tags map[string]string + // EgressMode is the pool's outbound policy, from provider.ProvisionRequest.Egress. + // Empty and Open both mean unrestricted. Blocked and Allowlist both reach Modal as an + // outbound ALLOWLIST, Blocked being the allowlist of nothing — never as BlockNetwork, + // which severs inbound too and would kill the connect URL (see CreateSandbox). + EgressMode nebulav1alpha1.EgressMode + // EgressTargets is what Allowlist permits: CIDRs, bare IPs and domain names mixed in one + // list, which util.SplitEgressTargets sorts into Modal's two separate allowlist fields. + EgressTargets []string // ReadinessProbe, when non-nil, is the Pod's first-container readinessProbe // carried through so the Client can configure Modal's own readiness probe at // create time. Modal enforces the probe internally (it gates its own traffic @@ -169,9 +177,10 @@ type SandboxSpec struct { // a pointer, so %v would print an address, and only its presence matters. func (s SandboxSpec) String() string { return fmt.Sprintf("SandboxSpec{Image:%s Command:%v Env:%s GPU:%s GPUCount:%d CPU:%g "+ - "MemoryMiB:%d Ports:%v Regions:%v Timeout:%s Tags:%v ReadinessProbe:%t}", + "MemoryMiB:%d Ports:%v Regions:%v Egress:%s EgressTargets:%v Timeout:%s Tags:%v ReadinessProbe:%t}", s.Image, s.Command, provider.RedactedEnv(s.Env), s.GPU, s.GPUCount, s.CPU, - s.MemoryMiB, s.Ports, s.Regions, s.Timeout, s.Tags, s.ReadinessProbe != nil) + s.MemoryMiB, s.Ports, s.Regions, s.EgressMode, s.EgressTargets, + s.Timeout, s.Tags, s.ReadinessProbe != nil) } // GoString implements fmt.GoStringer so %#v is redacted too. @@ -294,11 +303,12 @@ func (p *Provider) ExpandRegions(declared []string) []string { // trait is set the way it is. func (p *Provider) Capabilities() provider.Capabilities { return provider.Capabilities{ - SupportsStop: false, // create/terminate only - SupportsSpot: false, // no user-facing preemptible tier - NativeTags: true, // sandbox tags carry identity - PreemptionNotice: 0, // no push; poll-based detection - PollInterval: 0, // OnDemand-only (never preempts) → the default cadence is fine + SupportsStop: false, // create/terminate only + SupportsSpot: false, // no user-facing preemptible tier + SupportsEgressPolicy: true, // outbound allowlists on the sandbox itself + NativeTags: true, // sandbox tags carry identity + PreemptionNotice: 0, // no push; poll-based detection + PollInterval: 0, // OnDemand-only (never preempts) → the default cadence is fine } } @@ -500,6 +510,8 @@ func (p *Provider) sandboxSpecFromPod(pod *corev1.Pod, req provider.ProvisionReq // reach Modal as "no placement constraint" — its widest pool and its // un-multiplied price. See SandboxSpec.Regions. Regions: regionsOf(req.Region), + EgressMode: req.Egress.ModeOrOpen(), + EgressTargets: req.Egress.GetTargets(), Timeout: sandboxTimeout(pod), Tags: tags, ReadinessProbe: c.ReadinessProbe, diff --git a/pkg/provider/modal/modal_test.go b/pkg/provider/modal/modal_test.go index 0520a77..497b048 100644 --- a/pkg/provider/modal/modal_test.go +++ b/pkg/provider/modal/modal_test.go @@ -1350,3 +1350,95 @@ func TestSandboxSpec_StringRedactsEnv(t *testing.T) { t.Fatalf("expected key names to survive redaction: %s", got) } } + +// TestOutboundAllowlists pins the nil-versus-empty distinction the whole feature rests on: +// a nil allowlist leaves Modal in OPEN mode, while a non-nil empty one selects ALLOWLIST +// with nothing permitted. Getting these confused silently either blocks an open pool or +// opens a Blocked one, and neither shows up as an error. +func TestOutboundAllowlists(t *testing.T) { + t.Run("open sends nil, leaving Modal in OPEN mode", func(t *testing.T) { + for _, mode := range []nebulav1alpha1.EgressMode{"", nebulav1alpha1.EgressOpen} { + cidrs, domains, err := outboundAllowlists(SandboxSpec{EgressMode: mode}) + if err != nil { + t.Fatalf("mode %q: %v", mode, err) + } + if cidrs != nil || domains != nil { + t.Errorf("mode %q: got %v/%v, want nil pair (non-nil would select ALLOWLIST)", + mode, cidrs, domains) + } + } + }) + + t.Run("blocked sends an empty pair, permitting nothing", func(t *testing.T) { + cidrs, domains, err := outboundAllowlists(SandboxSpec{EgressMode: nebulav1alpha1.EgressBlocked}) + if err != nil { + t.Fatalf("outboundAllowlists: %v", err) + } + if cidrs == nil || domains == nil { + t.Fatal("got a nil allowlist for Blocked, which is Modal's OPEN mode") + } + if len(cidrs.Entries) != 0 || len(domains.Entries) != 0 { + t.Errorf("got entries %v/%v, want both empty", cidrs.Entries, domains.Entries) + } + }) + + t.Run("allowlist carries the split entries", func(t *testing.T) { + cidrs, domains, err := outboundAllowlists(SandboxSpec{ + EgressMode: nebulav1alpha1.EgressAllowlist, + EgressTargets: []string{"10.0.0.0/8", "*.huggingface.co"}, + }) + if err != nil { + t.Fatalf("outboundAllowlists: %v", err) + } + if !slices.Equal(cidrs.Entries, []string{"10.0.0.0/8"}) { + t.Errorf("cidrs = %v", cidrs.Entries) + } + if !slices.Equal(domains.Entries, []string{"*.huggingface.co"}) { + t.Errorf("domains = %v", domains.Entries) + } + }) + + // Failing beats defaulting: a mode this adapter does not know means it and placement + // have drifted, and guessing "open" would hand back the internet access the pool asked + // to remove. + t.Run("an unknown mode fails rather than opening up", func(t *testing.T) { + if _, _, err := outboundAllowlists(SandboxSpec{EgressMode: "Sideways"}); err == nil { + t.Fatal("expected an error for an unknown egress mode") + } + }) +} + +// TestProvision_CarriesEgressPolicy is the end-to-end half: the policy has to survive the +// ProvisionRequest → SandboxSpec hop, since that is where the annotation the placement +// controller stamped turns into something Modal can enforce. +func TestProvision_CarriesEgressPolicy(t *testing.T) { + f := &fakeClient{createID: "sb-1"} + p := newTestProvider(f) + req := provider.ProvisionRequest{ + ClaimName: "claim-a", + Egress: &nebulav1alpha1.EgressPolicy{ + Mode: nebulav1alpha1.EgressAllowlist, + Targets: []string{"*.huggingface.co"}, + }, + } + if _, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), req); err != nil { + t.Fatalf("Provision: %v", err) + } + if got := f.lastSpec.EgressMode; got != nebulav1alpha1.EgressAllowlist { + t.Errorf("spec.EgressMode = %q, want %q", got, nebulav1alpha1.EgressAllowlist) + } + if !slices.Equal(f.lastSpec.EgressTargets, []string{"*.huggingface.co"}) { + t.Errorf("spec.EgressTargets = %v", f.lastSpec.EgressTargets) + } + + // A nil policy is Open, which is what every pool that never set the field sends. + f = &fakeClient{createID: "sb-2"} + p = newTestProvider(f) + if _, err := p.Provision(context.Background(), gpuPod("claim-b", "H100", 1), + provider.ProvisionRequest{ClaimName: "claim-b"}); err != nil { + t.Fatalf("Provision: %v", err) + } + if got := f.lastSpec.EgressMode; got != nebulav1alpha1.EgressOpen { + t.Errorf("spec.EgressMode = %q, want %q for a nil policy", got, nebulav1alpha1.EgressOpen) + } +} diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go index 1378dcd..2e21fbd 100644 --- a/pkg/provider/provider.go +++ b/pkg/provider/provider.go @@ -231,6 +231,12 @@ type ProvisionRequest struct { // regions leaves it empty, which on Modal is the widest and cheapest option (pinning // costs 1.5-1.75x). AWS cannot honour it, but its ExpandRegions never produces it. Region string + // Egress is the pool's outbound policy, or nil for Open. Placement has already checked + // 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. + 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, // then env overriding it). A provider forwards it and never re-reads the Pod's env, @@ -250,8 +256,8 @@ type ProvisionRequest struct { // log.Info("...", "req", req). Key names print, since they are in the Pod spec already and // are what makes a "wrong env" report actionable; only values are withheld. func (r ProvisionRequest) String() string { - return fmt.Sprintf("ProvisionRequest{ClaimName:%s CapacityType:%s Region:%s Env:%s}", - r.ClaimName, r.CapacityType, r.Region, RedactedEnv(r.Env)) + return fmt.Sprintf("ProvisionRequest{ClaimName:%s CapacityType:%s Region:%s Egress:%s Env:%s}", + r.ClaimName, r.CapacityType, r.Region, r.Egress.ModeOrOpen(), RedactedEnv(r.Env)) } // GoString implements fmt.GoStringer so %#v is redacted too. @@ -306,6 +312,12 @@ type Capabilities struct { SupportsStop bool // SupportsSpot is true if the provider offers interruptible capacity. SupportsSpot bool + // SupportsEgressPolicy is true if the provider can enforce NodePoolSpec.Egress on the + // instances it creates. False means placement SKIPS this provider for any pool that + // restricts egress, rather than provisioning something with open internet access under + // a policy that says otherwise (AWS: false — its instances land in the default VPC, so + // enforcement needs security-group egress rules and no NAT, not one API field). + SupportsEgressPolicy bool // NativeTags is true if the provider has real instance tags/labels; when // false, identity is encoded in the instance name (RunPod: false). NativeTags bool diff --git a/pkg/util/network.go b/pkg/util/network.go new file mode 100644 index 0000000..39d278a --- /dev/null +++ b/pkg/util/network.go @@ -0,0 +1,45 @@ +/* +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 util + +import ( + "fmt" + "net" +) + +// SplitEgressTargets sorts an EgressPolicy.Targets list into prefixes and domain names. +// Which kind an entry is can be decided by parsing, so a pool declares one list (users +// think "let it reach S3 and huggingface", not "which field is this") and adapters that +// take the two separately split here rather than each rolling its own. +func SplitEgressTargets(entries []string) (cidrs, domains []string) { + for _, e := range entries { + if _, _, err := net.ParseCIDR(e); err == nil { + cidrs = append(cidrs, e) + continue + } + if ip := net.ParseIP(e); ip != nil { + bits := 32 + if ip.To4() == nil { + bits = 128 + } + cidrs = append(cidrs, fmt.Sprintf("%s/%d", ip, bits)) + continue + } + domains = append(domains, e) + } + return cidrs, domains +} diff --git a/pkg/util/network_test.go b/pkg/util/network_test.go new file mode 100644 index 0000000..958369e --- /dev/null +++ b/pkg/util/network_test.go @@ -0,0 +1,70 @@ +/* +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 util + +import ( + "slices" + "testing" +) + +// TestSplitEgressTargets covers the one decision a pool's single target list defers to +// the adapter: which axis an entry belongs to. +func TestSplitEgressTargets(t *testing.T) { + for _, tc := range []struct { + name string + entries []string + wantCIDRs []string + wantDomains []string + }{{ + name: "prefixes stay verbatim", + entries: []string{"10.0.0.0/8", "2001:db8::/32"}, + wantCIDRs: []string{"10.0.0.0/8", "2001:db8::/32"}, + }, { + // The CIDR axis takes prefixes, so a bare IP has to be widened rather than + // dropped or sent as-is. + name: "bare IPs become single-address prefixes", + entries: []string{"1.2.3.4", "2001:db8::1"}, + wantCIDRs: []string{"1.2.3.4/32", "2001:db8::1/128"}, + }, { + name: "names and wildcards are domains", + entries: []string{"api.openai.com", "*.huggingface.co"}, + wantDomains: []string{"api.openai.com", "*.huggingface.co"}, + }, { + // The mixed list is the reason the split exists at all: users declare one list + // and never say which kind an entry is. + name: "a mixed list divides", + entries: []string{"10.0.0.0/8", "*.huggingface.co", "1.2.3.4"}, + wantCIDRs: []string{"10.0.0.0/8", "1.2.3.4/32"}, + wantDomains: []string{"*.huggingface.co"}, + }, { + // The provider owns the domain vocabulary, so an unparseable entry is forwarded + // for it to reject with its own error rather than silently dropped here. + name: "an unrecognized entry goes to domains", + entries: []string{"not a host"}, + wantDomains: []string{"not a host"}, + }} { + t.Run(tc.name, func(t *testing.T) { + cidrs, domains := SplitEgressTargets(tc.entries) + if !slices.Equal(cidrs, tc.wantCIDRs) { + t.Errorf("cidrs = %v, want %v", cidrs, tc.wantCIDRs) + } + if !slices.Equal(domains, tc.wantDomains) { + t.Errorf("domains = %v, want %v", domains, tc.wantDomains) + } + }) + } +} diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index 6bf2288..d036540 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -22,6 +22,7 @@ import ( "fmt" "io" "math/rand/v2" + "strings" "sync" "time" @@ -228,6 +229,7 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { ClaimName: claim, CapacityType: nebulav1alpha1.CapacityType(pod.Annotations[nebulav1alpha1.CapacityTypeAnnotation]), Region: pod.Annotations[nebulav1alpha1.RegionAnnotation], + Egress: egressFromPod(pod), } log := logf.FromContext(ctx).WithName("vnode-handler").WithValues( @@ -991,6 +993,28 @@ func blocklistTTL(pod *corev1.Pod) time.Duration { return d } +// egressFromPod reads the pool's egress policy off the annotations placement stamped. nil +// (no annotation) means Open, which is what every pool that never set the field gets. +// +// The mode is taken verbatim rather than validated against the enum: admission already +// rejected anything else, and an unknown value must not silently become Open — the adapter +// fails the Provision instead, which is the safe direction for a containment policy. +func egressFromPod(pod *corev1.Pod) *nebulav1alpha1.EgressPolicy { + mode := pod.Annotations[nebulav1alpha1.EgressAnnotation] + if mode == "" { + return nil + } + policy := &nebulav1alpha1.EgressPolicy{Mode: nebulav1alpha1.EgressMode(mode)} + if raw := pod.Annotations[nebulav1alpha1.EgressTargetsAnnotation]; raw != "" { + for _, e := range strings.Split(raw, ",") { + if e = strings.TrimSpace(e); e != "" { + policy.Targets = append(policy.Targets, e) + } + } + } + return policy +} + func ptrNow(t metav1.Time) *metav1.Time { return &t } // Tracks reports whether this virtual node is the one running the Pod. The kubelet API diff --git a/pkg/vnode/handler_test.go b/pkg/vnode/handler_test.go index 5f756be..ab09ce3 100644 --- a/pkg/vnode/handler_test.go +++ b/pkg/vnode/handler_test.go @@ -1320,3 +1320,57 @@ func TestGetPods_ReturnsTracked(t *testing.T) { t.Fatalf("expected 2 tracked pods, got %d", len(pods)) } } + +// TestCreatePod_ReadsEgressPolicyFromAnnotations covers the second half of the pinning +// decision behind EgressAnnotation: the handler never reads the NodePool, so if it does not +// reconstruct the policy from these annotations the adapter gets nil and provisions a +// workload with open egress under a pool that asked for containment. +func TestCreatePod_ReadsEgressPolicyFromAnnotations(t *testing.T) { + for _, tc := range []struct { + name string + annotations map[string]string + wantMode nebulav1alpha1.EgressMode + wantTargets []string + }{{ + // Absence is Open, which is what every Pod placed by a pool without the field + // carries — so nil here must not be mistaken for a missing value. + name: "no annotation is Open", + annotations: nil, + wantMode: nebulav1alpha1.EgressOpen, + }, { + name: "blocked needs no targets", + annotations: map[string]string{nebulav1alpha1.EgressAnnotation: string(nebulav1alpha1.EgressBlocked)}, + wantMode: nebulav1alpha1.EgressBlocked, + }, { + name: "target entries are split back out", + annotations: map[string]string{ + nebulav1alpha1.EgressAnnotation: string(nebulav1alpha1.EgressAllowlist), + nebulav1alpha1.EgressTargetsAnnotation: "10.0.0.0/8,*.huggingface.co", + }, + wantMode: nebulav1alpha1.EgressAllowlist, + wantTargets: []string{"10.0.0.0/8", "*.huggingface.co"}, + }, { + // The mode is authoritative: a list with no mode cannot narrow anything on its + // own, so it is ignored rather than treated as an implicit Allowlist. + name: "targets without a mode are ignored", + annotations: map[string]string{nebulav1alpha1.EgressTargetsAnnotation: "10.0.0.0/8"}, + wantMode: nebulav1alpha1.EgressOpen, + }} { + t.Run(tc.name, func(t *testing.T) { + fp := &fakeProvider{provisionID: "inst-1"} + h := NewHandler(fp, nil, nil) + pod := testPod("default", "p1") + pod.Annotations = tc.annotations + + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + if got := fp.lastReq.Egress.ModeOrOpen(); got != tc.wantMode { + t.Errorf("req.Egress mode = %q, want %q", got, tc.wantMode) + } + if got := fp.lastReq.Egress.GetTargets(); !slices.Equal(got, tc.wantTargets) { + t.Errorf("req.Egress allow = %v, want %v", got, tc.wantTargets) + } + }) + } +}