diff --git a/api/v1alpha1/nodeclaim_types.go b/api/v1alpha1/nodeclaim_types.go index c938f54..e1bdec9 100644 --- a/api/v1alpha1/nodeclaim_types.go +++ b/api/v1alpha1/nodeclaim_types.go @@ -41,16 +41,19 @@ type NodeClaimSpec struct { // +optional Region string `json:"region,omitempty"` - // AcceleratorID is the provider's own identifier for what actually serves this - // claim (e.g. AWS "p5.48xlarge", RunPod "NVIDIA H100 80GB HBM3"), resolved at - // placement time from the Pod's requested accelerator type + count via the - // provider's MapAccelerator. Unlike Provider/CapacityType/Region it is NOT a - // provisioning input (the provider re-derives it from the Pod) — it is recorded - // for reporting, like PoolRef, so `kubectl get nc` shows the concrete SKU behind - // each instance without cross-referencing the Pod. Empty for a CPU-only claim, - // which requests no accelerator. + // Accelerator is the requested accelerator pool this claim serves, as + // "type:count" (e.g. "H100:8"), resolved at placement time from the Pod's + // accelerator type + count. It names the POOL, not the concrete SKU: a launch + // may span several interchangeable provider instance types (AWS's fleet tries + // alternates), so the exact instance type is only known post-launch from the + // observed instance — this field stays truthful regardless of which alternate + // lands. Unlike Provider/CapacityType/Region it is NOT a provisioning input (the + // provider re-derives it from the Pod) — it is recorded for reporting, like + // PoolRef, so `kubectl get nc` shows what each instance serves without + // cross-referencing the Pod. Empty for a CPU-only claim, which requests no + // accelerator. // +optional - AcceleratorID string `json:"acceleratorID,omitempty"` + Accelerator string `json:"accelerator,omitempty"` // PoolRef is the NodePool whose policy produced this claim, for reporting. // +optional @@ -150,7 +153,7 @@ type NodeClaimStatus struct { // +kubebuilder:subresource:status // +kubebuilder:printcolumn:name="Provider",type=string,JSONPath=`.spec.provider` // +kubebuilder:printcolumn:name="Region",type=string,JSONPath=`.spec.region` -// +kubebuilder:printcolumn:name="ACCELERATOR_ID",type=string,JSONPath=`.spec.acceleratorID` +// +kubebuilder:printcolumn:name="ACCELERATOR",type=string,JSONPath=`.spec.accelerator` // +kubebuilder:printcolumn:name="CAPACITY_TYPE",type=string,JSONPath=`.spec.capacityType` // +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase` // +kubebuilder:printcolumn:name="Instance",type=string,JSONPath=`.status.instanceID` diff --git a/api/v1alpha1/nodepool_types.go b/api/v1alpha1/nodepool_types.go index ba2696c..b73c7dc 100644 --- a/api/v1alpha1/nodepool_types.go +++ b/api/v1alpha1/nodepool_types.go @@ -140,7 +140,7 @@ const ( type FailoverPolicy struct { // BlocklistTTL is how long a failed placement is excluded before the // provider becomes a candidate for it again. - // +kubebuilder:default="10m" + // +kubebuilder:default="3m" BlocklistTTL metav1.Duration `json:"blocklistTTL,omitempty"` } diff --git a/config/crd/bases/nebula.inftyai.com_nodeclaims.yaml b/config/crd/bases/nebula.inftyai.com_nodeclaims.yaml index 5cba3db..2aad621 100644 --- a/config/crd/bases/nebula.inftyai.com_nodeclaims.yaml +++ b/config/crd/bases/nebula.inftyai.com_nodeclaims.yaml @@ -23,8 +23,8 @@ spec: - jsonPath: .spec.region name: Region type: string - - jsonPath: .spec.acceleratorID - name: ACCELERATOR_ID + - jsonPath: .spec.accelerator + name: ACCELERATOR type: string - jsonPath: .spec.capacityType name: CAPACITY_TYPE @@ -70,16 +70,19 @@ spec: ledger, not a spec: its reason to exist is to survive the Node so teardown can reclaim the external instance and never leak a paid GPU. properties: - acceleratorID: + accelerator: description: |- - AcceleratorID is the provider's own identifier for what actually serves this - claim (e.g. AWS "p5.48xlarge", RunPod "NVIDIA H100 80GB HBM3"), resolved at - placement time from the Pod's requested accelerator type + count via the - provider's MapAccelerator. Unlike Provider/CapacityType/Region it is NOT a - provisioning input (the provider re-derives it from the Pod) — it is recorded - for reporting, like PoolRef, so `kubectl get nc` shows the concrete SKU behind - each instance without cross-referencing the Pod. Empty for a CPU-only claim, - which requests no accelerator. + Accelerator is the requested accelerator pool this claim serves, as + "type:count" (e.g. "H100:8"), resolved at placement time from the Pod's + accelerator type + count. It names the POOL, not the concrete SKU: a launch + may span several interchangeable provider instance types (AWS's fleet tries + alternates), so the exact instance type is only known post-launch from the + observed instance — this field stays truthful regardless of which alternate + lands. Unlike Provider/CapacityType/Region it is NOT a provisioning input (the + provider re-derives it from the Pod) — it is recorded for reporting, like + PoolRef, so `kubectl get nc` shows what each instance serves without + cross-referencing the Pod. Empty for a CPU-only claim, which requests no + accelerator. type: string capacityType: description: |- diff --git a/config/crd/bases/nebula.inftyai.com_nodepools.yaml b/config/crd/bases/nebula.inftyai.com_nodepools.yaml index 19aa2a7..8f95a7f 100644 --- a/config/crd/bases/nebula.inftyai.com_nodepools.yaml +++ b/config/crd/bases/nebula.inftyai.com_nodepools.yaml @@ -101,7 +101,7 @@ spec: RunPod reports no capacity) is temporarily excluded and re-tried. properties: blocklistTTL: - default: 10m + default: 3m description: |- BlocklistTTL is how long a failed placement is excluded before the provider becomes a candidate for it again. diff --git a/config/samples/deployment.yaml b/config/samples/deployment.yaml index f45239b..84268ae 100644 --- a/config/samples/deployment.yaml +++ b/config/samples/deployment.yaml @@ -23,7 +23,7 @@ apiVersion: apps/v1 kind: Deployment metadata: - name: gpu-workload-sample + name: gpu-workload-sample-a100 namespace: default labels: app.kubernetes.io/managed-by: nebula diff --git a/config/samples/nebula_v1alpha1_nodepool.yaml b/config/samples/nebula_v1alpha1_nodepool.yaml index e33ced8..7b2c852 100644 --- a/config/samples/nebula_v1alpha1_nodepool.yaml +++ b/config/samples/nebula_v1alpha1_nodepool.yaml @@ -7,7 +7,6 @@ metadata: spec: # strategy (the inner, provider-ranking axis). providers: - - name: modal - name: aws regions: - us-east-1 @@ -18,7 +17,7 @@ spec: - eu-west-1 - ca-central-1 - sa-east-1 - - name: modal + # - name: modal # - name: runpod # Outer axis: try OnDemand on every provider first, fall back to Spot. capacityTypes: @@ -27,4 +26,4 @@ spec: # Inner axis: within the active capacity tier. strategy: Ordered failover: - blocklistTTL: 10m + blocklistTTL: 3m diff --git a/internal/controller/nodeclaim_controller_test.go b/internal/controller/nodeclaim_controller_test.go index ab93452..e3b2b9c 100644 --- a/internal/controller/nodeclaim_controller_test.go +++ b/internal/controller/nodeclaim_controller_test.go @@ -62,16 +62,16 @@ func (f *fakeProvider) List(context.Context) ([]provider.Instance, error) { return f.list, f.listErr } func (f *fakeProvider) Offerings(context.Context) ([]provider.Offering, error) { return nil, nil } -func (f *fakeProvider) MapAccelerator(c string, _ int32) (string, bool) { +func (f *fakeProvider) MapAccelerator(c string, _ int32) ([]string, bool) { if f.gpus == nil { - return c, true // offer any accelerator + return []string{c}, true // offer any accelerator } for _, g := range f.gpus { if g == c { - return c, true + return []string{c}, true } } - return "", false + return nil, false } func (f *fakeProvider) ClassifyProvisionError(error, string, string) provider.BlockScope { return provider.BlockScope{} diff --git a/internal/controller/pod_placement_controller.go b/internal/controller/pod_placement_controller.go index c6aafea..c201a4a 100644 --- a/internal/controller/pod_placement_controller.go +++ b/internal/controller/pod_placement_controller.go @@ -186,10 +186,11 @@ type placement struct { // Empty means the provider's configured default region; region-simple // providers leave it empty. region string - // acceleratorID is the provider's own identifier for what serves this request - // (e.g. AWS "p5.48xlarge"), resolved via MapAccelerator(type, count). Empty for - // a CPU-only Pod, which requests no accelerator. - acceleratorID string + // accelerator is the requested pool identity (type:count, e.g. "H100:8"). It is + // what the blocklist keys on and the NodeClaim reports — the pool, not the + // provider SKU, so it stays truthful when a launch spans interchangeable + // instance types. Empty for a CPU-only Pod, which requests no accelerator. + accelerator string } // needsPlacement reports whether the Pod is an opted-in workload still held by diff --git a/internal/controller/pod_placement_controller_test.go b/internal/controller/pod_placement_controller_test.go index 70faad7..ccf6b24 100644 --- a/internal/controller/pod_placement_controller_test.go +++ b/internal/controller/pod_placement_controller_test.go @@ -49,7 +49,7 @@ func (b *fakeBlocklist) BlockedUntil(c failover.Candidate) (time.Duration, bool) if e.Provider != "" && e.Provider != c.Provider { continue } - if e.AcceleratorID != "" && e.AcceleratorID != c.AcceleratorID { + if e.Accelerator != "" && e.Accelerator != c.Accelerator { continue } if e.CapacityType != "" && e.CapacityType != c.CapacityType { @@ -181,10 +181,10 @@ 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) } - // The provider's resolved accelerator id is recorded for reporting (the fake's - // MapAccelerator echoes the canonical type, so H100 -> "H100"). - if nc.Spec.AcceleratorID != "H100" { - t.Fatalf("expected claim to record acceleratorID H100, got %q", nc.Spec.AcceleratorID) + // 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" { + t.Fatalf("expected claim to record accelerator H100:1, got %q", nc.Spec.Accelerator) } } @@ -251,13 +251,13 @@ func TestPlacement_CPUOnlyPodMatchesAnyProvider(t *testing.T) { if got.Spec.NodeSelector[nebulav1alpha1.ProviderLabel] != provider.ProviderModal { t.Fatalf("expected a CPU-only Pod to place on modal, got %v", got.Spec.NodeSelector) } - // A CPU-only claim requests no accelerator, so the id is left empty. + // A CPU-only claim requests no accelerator, so the pool identity is left empty. var nc nebulav1alpha1.NodeClaim if err := c.Get(context.Background(), types.NamespacedName{Name: "default-p1"}, &nc); err != nil { t.Fatalf("expected NodeClaim default-p1: %v", err) } - if nc.Spec.AcceleratorID != "" { - t.Fatalf("expected empty acceleratorID for a CPU-only claim, got %q", nc.Spec.AcceleratorID) + if nc.Spec.Accelerator != "" { + t.Fatalf("expected empty accelerator for a CPU-only claim, got %q", nc.Spec.Accelerator) } } @@ -406,7 +406,7 @@ func TestPlacement_FailsOverToNextRegionWhenBlocked(t *testing.T) { prov := &fakeProvider{name: provider.ProviderModal, gpus: []string{"H100"}} r, c := newPlacementReconciler(t, []client.Object{pod, pool}, prov) r.Blocklist = &fakeBlocklist{blocked: []failover.Candidate{ - {Provider: provider.ProviderModal, AcceleratorID: "H100", CapacityType: nebulav1alpha1.CapacityOnDemand, Region: "us-east-1"}, + {Provider: provider.ProviderModal, Accelerator: "H100:1", CapacityType: nebulav1alpha1.CapacityOnDemand, Region: "us-east-1"}, }} reconcilePod(t, r, "default", "p1") diff --git a/internal/controller/pod_placement_helpers.go b/internal/controller/pod_placement_helpers.go index 9efd171..0077795 100644 --- a/internal/controller/pod_placement_helpers.go +++ b/internal/controller/pod_placement_helpers.go @@ -111,25 +111,25 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev continue // unregistered; NodePool status surfaces this separately } // A CPU-only Pod (no accelerator) matches any provider; an accelerator - // Pod only matches a provider whose catalog serves that (type, count). The - // resolved id is also what the block is keyed on, so it is captured here - // from the same lookup that decides servability. - acceleratorID := "" + // Pod only matches a provider whose catalog serves that (type, count). + // MapAccelerator is consulted only for that servability check — the block + // key and the reported identity are the POOL (type:count), not the + // provider's SKU, so a launch spanning alternates and a post-launch SKU + // swap never desync the key. Empty for a CPU-only Pod. + accelerator := util.AcceleratorPool(accel, count) if accel != "" { - id, offered := prov.MapAccelerator(accel, count) - if !offered { + if _, offered := prov.MapAccelerator(accel, count); !offered { log.V(1).Info("skipping candidate: provider does not offer the accelerator", "provider", ref.Name, "accelerator", accel, "count", count) continue } - acceleratorID = id } for _, region := range regionsFor(ref) { // inner: region - if until, blocked := r.blockedUntil(ref.Name, acceleratorID, tier, region); blocked { + if until, blocked := r.blockedUntil(ref.Name, accelerator, tier, region); blocked { // Servable but failed recently; try the next region, then the next // tier, and remember when this one frees so we can requeue for it. log.Info("skipping candidate: blocked by failover blocklist", - "provider", ref.Name, "acceleratorID", acceleratorID, + "provider", ref.Name, "accelerator", accelerator, "capacityType", tier, "region", region, "freesIn", until.String()) if until > 0 && (soonest == 0 || until < soonest) { soonest = until @@ -139,10 +139,10 @@ func (r *PodPlacementReconciler) selectPlacement(ctx context.Context, pod *corev log.Info("selected placement candidate", "provider", ref.Name, "capacityType", tier, "region", region) return placement{ - provider: ref.Name, - capacityType: tier, - region: region, - acceleratorID: acceleratorID, + provider: ref.Name, + capacityType: tier, + region: region, + accelerator: accelerator, }, true, 0 } } @@ -174,21 +174,21 @@ func regionsFor(ref nebulav1alpha1.ProviderSpec) []string { return ref.Regions } -// blockedUntil reports whether the (provider, acceleratorID, tier, region) +// blockedUntil reports whether the (provider, accelerator, tier, region) // candidate is currently excluded by the failover blocklist and, if so, how long -// until it frees (for the requeue hint). acceleratorID is the provider's resolved -// id for the request (see selectPlacement), so a capacity block matches only -// candidates on the same instance type / pool. It is nil-safe: with no blocklist -// wired (tests, or a blocklist-less build) nothing is ever blocked. -func (r *PodPlacementReconciler) blockedUntil(provName, acceleratorID string, tier nebulav1alpha1.CapacityType, region string) (time.Duration, bool) { +// until it frees (for the requeue hint). accelerator is the request's pool +// identity (type:count; see selectPlacement), so a capacity block matches only +// candidates on the same pool. It is nil-safe: with no blocklist wired (tests, or +// a blocklist-less build) nothing is ever blocked. +func (r *PodPlacementReconciler) blockedUntil(provName, accelerator string, tier nebulav1alpha1.CapacityType, region string) (time.Duration, bool) { if r.Blocklist == nil { return 0, false } return r.Blocklist.BlockedUntil(failover.Candidate{ - Provider: provName, - AcceleratorID: acceleratorID, - CapacityType: tier, - Region: region, + Provider: provName, + Accelerator: accelerator, + CapacityType: tier, + Region: region, }) } @@ -235,11 +235,11 @@ func (r *PodPlacementReconciler) ensureClaim(ctx context.Context, pod *corev1.Po Name: pod.Name, UID: string(pod.UID), }, - Provider: p.provider, - CapacityType: p.capacityType, - Region: p.region, - AcceleratorID: p.acceleratorID, - PoolRef: pool.Name, + Provider: p.provider, + CapacityType: p.capacityType, + Region: p.region, + Accelerator: p.accelerator, + PoolRef: pool.Name, }, } err = r.Create(ctx, claim) diff --git a/pkg/failover/blocklist.go b/pkg/failover/blocklist.go index de1928f..cbc4c29 100644 --- a/pkg/failover/blocklist.go +++ b/pkg/failover/blocklist.go @@ -51,18 +51,18 @@ type entry struct { expiresAt time.Time } -// Candidate is a placement being considered — the (provider, acceleratorID, tier, +// Candidate is a placement being considered — the (provider, accelerator, tier, // region) tuple the blocklist is queried against. It is the query counterpart to a // recorded BlockScope: Blocked reports whether any live entry's scope covers it. -// AcceleratorID is the provider's RESOLVED id for what would serve the request -// (the (type, count) mapped through MapAccelerator), not the bare accelerator -// type, so a block recorded against one instance type/capacity pool matches only -// candidates that share it. +// Accelerator is the request's POOL identity (type:count, e.g. "H100:8"), not the +// provider's SKU id, so a block recorded against one (type, count) pool matches +// only candidates that share it — and stays stable even when a launch spans +// several interchangeable provider instance types. type Candidate struct { - Provider string - AcceleratorID string - CapacityType nebulav1alpha1.CapacityType - Region string + Provider string + Accelerator string + CapacityType nebulav1alpha1.CapacityType + Region string } // Blocklist is a concurrency-safe, TTL-bounded set of provider blocks. The zero @@ -207,7 +207,7 @@ func scopeCovers(scope provider.BlockScope, c Candidate) bool { if scope.DenyAll { return true } - if !ptrMatches(scope.AcceleratorID, c.AcceleratorID) { + if !ptrMatches(scope.Accelerator, c.Accelerator) { return false } if scope.CapacityType != "" && scope.CapacityType != c.CapacityType { @@ -230,7 +230,7 @@ func scopeCovers(scope provider.BlockScope, c Candidate) bool { func scopeEqual(a, b provider.BlockScope) bool { return a.DenyAll == b.DenyAll && a.CapacityType == b.CapacityType && - ptrEqual(a.AcceleratorID, b.AcceleratorID) && + ptrEqual(a.Accelerator, b.Accelerator) && ptrEqual(a.Region, b.Region) } diff --git a/pkg/failover/blocklist_test.go b/pkg/failover/blocklist_test.go index 84d719a..53bc3fa 100644 --- a/pkg/failover/blocklist_test.go +++ b/pkg/failover/blocklist_test.go @@ -35,16 +35,16 @@ func ptr(s string) *string { return &s } func spotH100EastScope() provider.BlockScope { return provider.BlockScope{ - AcceleratorID: ptr("H100"), - CapacityType: nebulav1alpha1.CapacitySpot, - Region: ptr("us-east-1"), + Accelerator: ptr("H100"), + CapacityType: nebulav1alpha1.CapacitySpot, + Region: ptr("us-east-1"), } } // cand builds a query Candidate compactly, keeping the table rows within the // line-length limit. func cand(prov, accel string, tier nebulav1alpha1.CapacityType, region string) Candidate { - return Candidate{Provider: prov, AcceleratorID: accel, CapacityType: tier, Region: region} + return Candidate{Provider: prov, Accelerator: accel, CapacityType: tier, Region: region} } const ( @@ -100,13 +100,13 @@ func TestBlocked_ScopeMatchIsPrecise(t *testing.T) { func TestBlocked_WildcardFieldsMatchAnyValue(t *testing.T) { fc := testclock.NewFakeClock(baseTime) bl := newWithClock(fc) - // Wildcard AcceleratorID + wildcard Region (&"") => blocks Spot on aws + // Wildcard Accelerator + wildcard Region (&"") => blocks Spot on aws // everywhere, any GPU. The pointers are deliberately non-nil-empty: nil would // mean "no such axis" and match only an empty candidate field. bl.Record("aws", provider.BlockScope{ - AcceleratorID: ptr(""), - CapacityType: spot, - Region: ptr(""), + Accelerator: ptr(""), + CapacityType: spot, + Region: ptr(""), }, 10*time.Minute) if !bl.Blocked(cand("aws", "H100", spot, "us-east-1")) { @@ -122,12 +122,12 @@ func TestBlocked_WildcardFieldsMatchAnyValue(t *testing.T) { // A nil axis is "not applicable": it matches only a candidate whose field is empty, // never a populated one. This is how a region-simple provider (Modal: nil Region) -// or a CPU-only Pod (nil AcceleratorID) blocks without widening across an axis it +// or a CPU-only Pod (nil Accelerator) blocks without widening across an axis it // never had. Contrast &"" (wildcard), which matches any value — see the test above. func TestBlocked_NilFieldMatchesOnlyEmptyCandidate(t *testing.T) { fc := testclock.NewFakeClock(baseTime) bl := newWithClock(fc) - // A region-simple provider's block: nil Region, nil AcceleratorID, Spot only. + // A region-simple provider's block: nil Region, nil Accelerator, Spot only. bl.Record("modal", provider.BlockScope{CapacityType: spot}, 10*time.Minute) // A region-simple candidate carries an empty region and (CPU Pod) empty @@ -138,7 +138,7 @@ func TestBlocked_NilFieldMatchesOnlyEmptyCandidate(t *testing.T) { // A populated candidate field is NOT matched by a nil axis: it is out of scope, // not wildcarded in. if bl.Blocked(cand("modal", "H100", spot, "")) { - t.Error("nil AcceleratorID must not match a populated accelerator") + t.Error("nil Accelerator must not match a populated accelerator") } if bl.Blocked(cand("modal", "", spot, "us-east-1")) { t.Error("nil Region must not match a populated region") @@ -155,7 +155,7 @@ func TestBlocked_DenyAllBlocksWholeProvider(t *testing.T) { t.Error("DenyAll must block any aws candidate") } // ...but a different provider is untouched (a block never spans providers). - if bl.Blocked(Candidate{Provider: "modal", AcceleratorID: "H100"}) { + if bl.Blocked(Candidate{Provider: "modal", Accelerator: "H100"}) { t.Error("DenyAll on aws must not block modal") } } @@ -226,10 +226,10 @@ func TestRecord_NoOpOnInvalidInput(t *testing.T) { bl.Record("aws", provider.BlockScope{DenyAll: true}, 0) bl.Record("aws", provider.BlockScope{DenyAll: true}, -time.Minute) - if bl.Blocked(Candidate{Provider: "aws", AcceleratorID: "H100"}) { + if bl.Blocked(Candidate{Provider: "aws", Accelerator: "H100"}) { t.Error("invalid Record inputs must not block anything") } - if bl.Blocked(Candidate{Provider: "", AcceleratorID: "H100"}) { + if bl.Blocked(Candidate{Provider: "", Accelerator: "H100"}) { t.Error("empty-provider block must not exist") } } @@ -282,10 +282,10 @@ func TestRecord_DistinctScopesStaySeparate(t *testing.T) { // Scopes differing on any single axis are different blocks and must NOT coalesce. bl.Record("aws", spotH100EastScope(), 10*time.Minute) // H100/Spot/us-east-1 bl.Record("aws", provider.BlockScope{ // different region - AcceleratorID: ptr("H100"), CapacityType: spot, Region: ptr("us-west-2"), + Accelerator: ptr("H100"), CapacityType: spot, Region: ptr("us-west-2"), }, 10*time.Minute) bl.Record("aws", provider.BlockScope{ // different tier - AcceleratorID: ptr("H100"), CapacityType: onDemand, Region: ptr("us-east-1"), + Accelerator: ptr("H100"), CapacityType: onDemand, Region: ptr("us-east-1"), }, 10*time.Minute) bl.Record("modal", spotH100EastScope(), 10*time.Minute) // different provider @@ -301,7 +301,7 @@ func TestRecord_NilAndWildcardScopesDoNotCoalesce(t *testing.T) { // nil (axis not applicable) and &"" (wildcard) are semantically different scopes, // so scopeEqual must keep them as separate entries. nilAxes := provider.BlockScope{CapacityType: spot} - wildcardAxes := provider.BlockScope{AcceleratorID: ptr(""), Region: ptr(""), CapacityType: spot} + wildcardAxes := provider.BlockScope{Accelerator: ptr(""), Region: ptr(""), CapacityType: spot} bl.Record("aws", nilAxes, 10*time.Minute) bl.Record("aws", wildcardAxes, 10*time.Minute) diff --git a/pkg/provider/aws/aws.go b/pkg/provider/aws/aws.go index d4c5db4..1ffc669 100644 --- a/pkg/provider/aws/aws.go +++ b/pkg/provider/aws/aws.go @@ -116,9 +116,13 @@ type Client interface { // RunInstances call. The adapter builds it from the Pod (source of truth) plus // the resolved instance type and capacity tier. type InstanceSpec struct { - // InstanceType is the EC2 instance type resolved from the accelerator (e.g. - // "p5.48xlarge"), via the catalog's accelerator_id column. - InstanceType string + // InstanceTypes are the EC2 instance types that can serve the accelerator (e.g. + // "p5.48xlarge"), resolved from the catalog's accelerator_id column, PRIMARY + // first then interchangeable alternates. The launch fleet spans them all in one + // request so EC2 lands on whichever (type, AZ) pair has capacity; the primary + // (InstanceTypes[0]) is the one the blocklist keys on. Always has at least one + // element for a GPU launch. + InstanceTypes []string // Image is the container image, from the Pod's first container. The Client is // responsible for launching it (e.g. via a GPU AMI + user-data, or ECS/EKS). Image string @@ -595,11 +599,13 @@ func (p *Provider) List(ctx context.Context) ([]provider.Instance, error) { // - EC2 capacity is per-region and this adapter is multi-region, so an // accelerator/capacity block is confined to the region that ACTUALLY FAILED // (the region the failing request targeted) — a "no capacity in us-east-1" -// failure does not disqualify the same request in us-west-2. Whole-provider -// blocks (auth/quota) are left region-wide, since bad credentials fail in every -// region. The region is passed in because the error alone does not carry it and -// the adapter has no default region to assume. -func (p *Provider) ClassifyProvisionError(err error, acceleratorID, region string) provider.BlockScope { +// failure does not disqualify the same request in us-west-2. This includes +// quota: EC2 vCPU limits are regional (and per-instance-family, per-tier), so a +// VcpuLimitExceeded in one region must not fence off the others. Only a DenyAll +// block (auth) is left region-wide, since bad credentials fail in every region. +// The region is passed in because the error alone does not carry it and the +// adapter has no default region to assume. +func (p *Provider) ClassifyProvisionError(err error, accelerator, region string) provider.BlockScope { if err == nil { return provider.BlockScope{} } @@ -607,9 +613,9 @@ func (p *Provider) ClassifyProvisionError(err error, acceleratorID, region strin if errors.Is(err, ErrSpotCapacity) { tier = nebulav1alpha1.CapacitySpot } - scope := provider.ClassifyError(err, tier, acceleratorID) - // Confine an accelerator/capacity block to the region that failed. A DenyAll - // (auth/quota) fails in every region, so it stays region-wide (Region left nil). + scope := provider.ClassifyError(err, tier, accelerator) + // Confine an accelerator/capacity/quota block to the region that failed. A + // DenyAll (auth) fails in every region, so it stays region-wide (Region left nil). // An empty region (should not happen — every request carries one) maps to &"" = // the wildcard, which blocks the accelerator in ALL regions: the safe over-broad // choice, since a stale-but-effective block beats a block pinned to a guessed @@ -678,8 +684,9 @@ func (p *Provider) instanceSpecFromPod(pod *corev1.Pod, req provider.ProvisionRe // one whose (accelerator_type, gpu_count) pair matches, since the GPU count is // baked into the instance type (T4x1 = g4dn.xlarge, T4x8 = g4dn.metal) rather // than a free knob. MapAccelerator (from the embedded catalog.Base) resolves by - // that (type, count) key straight from the catalog, so no AWS-specific lookup is - // needed here. + // that (type, count) key straight from the catalog, returning the PRIMARY instance + // type first then any interchangeable alternates — the launch fleet spans them all + // so EC2 can land on whichever (type, AZ) pair has capacity. canonical, count, err := util.AcceleratorRequest(pod) if err != nil { return InstanceSpec{}, fmt.Errorf("aws: %w", err) @@ -688,19 +695,19 @@ func (p *Provider) instanceSpecFromPod(pod *corev1.Pod, req provider.ProvisionRe return InstanceSpec{}, errors.New( "aws: pod requests no accelerator; EC2 GPU provisioning needs an accelerator type and count") } - instanceType, ok := p.MapAccelerator(canonical, count) + instanceTypes, ok := p.MapAccelerator(canonical, count) if !ok { return InstanceSpec{}, fmt.Errorf("aws: no EC2 instance type for %s x%d", canonical, count) } return InstanceSpec{ - InstanceType: instanceType, - Image: c.Image, - Command: append(append([]string{}, c.Command...), c.Args...), - Env: env, - Spot: req.CapacityType == nebulav1alpha1.CapacitySpot, - Region: req.Region, - Tags: map[string]string{ClaimTagKey: req.ClaimName}, + InstanceTypes: instanceTypes, + Image: c.Image, + Command: append(append([]string{}, c.Command...), c.Args...), + Env: env, + Spot: req.CapacityType == nebulav1alpha1.CapacitySpot, + Region: req.Region, + Tags: map[string]string{ClaimTagKey: req.ClaimName}, }, nil } diff --git a/pkg/provider/aws/aws_test.go b/pkg/provider/aws/aws_test.go index 6e4186d..f4b9398 100644 --- a/pkg/provider/aws/aws_test.go +++ b/pkg/provider/aws/aws_test.go @@ -27,6 +27,7 @@ import ( awssdk "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/ec2" ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types" + smithy "github.com/aws/smithy-go" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -176,9 +177,10 @@ func TestProvision_MapsAcceleratorToInstanceType(t *testing.T) { if id != "i-1" { t.Fatalf("id = %q, want i-1", id) } - // AWS requests by instance type: H100 must resolve to its accelerator_id. - if f.lastSpec.InstanceType != "p5.48xlarge" { - t.Fatalf("InstanceType = %q, want p5.48xlarge", f.lastSpec.InstanceType) + // AWS requests by instance type: H100 must resolve to its accelerator_id as the + // primary (fleet) instance type. + if got := primaryType(f.lastSpec); got != "p5.48xlarge" { + t.Fatalf("primary InstanceType = %q, want p5.48xlarge", got) } if f.lastSpec.Image != "myimg:latest" { t.Fatalf("image = %q", f.lastSpec.Image) @@ -210,11 +212,20 @@ func TestProvision_LowercaseAcceleratorLabel(t *testing.T) { }); err != nil { t.Fatalf("Provision: %v", err) } - if f.lastSpec.InstanceType != "p5.48xlarge" { - t.Fatalf("InstanceType = %q, want p5.48xlarge from lowercase label", f.lastSpec.InstanceType) + if got := primaryType(f.lastSpec); got != "p5.48xlarge" { + t.Fatalf("primary InstanceType = %q, want p5.48xlarge from lowercase label", got) } } +// primaryType returns the fleet's primary (blocklist-keyed) instance type, or "" +// when the spec carries none. +func primaryType(spec InstanceSpec) string { + if len(spec.InstanceTypes) == 0 { + return "" + } + return spec.InstanceTypes[0] +} + func TestProvision_CountSelectsInstanceType(t *testing.T) { // The GPU count is a lookup key on AWS: the same accelerator at different // counts must resolve to DIFFERENT instance types (T4x1 = g4dn.xlarge, @@ -233,8 +244,8 @@ func TestProvision_CountSelectsInstanceType(t *testing.T) { if _, err := p.Provision(context.Background(), gpuPod("T4", tc.count), req); err != nil { t.Fatalf("Provision(T4 x%d): %v", tc.count, err) } - if f.lastSpec.InstanceType != tc.wantType { - t.Fatalf("T4 x%d -> InstanceType %q, want %q", tc.count, f.lastSpec.InstanceType, tc.wantType) + if got := primaryType(f.lastSpec); got != tc.wantType { + t.Fatalf("T4 x%d -> primary InstanceType %q, want %q", tc.count, got, tc.wantType) } } } @@ -448,15 +459,15 @@ func TestClassifyProvisionError(t *testing.T) { p := newTestProvider(&fakeClient{}) const accel = "H100" denyAll := provider.BlockScope{DenyAll: true} - // EC2 capacity/accelerator blocks are confined to the adapter's region (an + // EC2 capacity/accelerator/quota blocks are confined to the adapter's region (an // exact-match Region pointer) AND to the requested accelerator (passed in by the - // caller — the provider owns the whole scope). Auth/quota (DenyAll) ignores both. + // caller — the provider owns the whole scope). Only auth (DenyAll) ignores both. region, wantAccel := testRegion, accel onDemandRegional := provider.BlockScope{ - AcceleratorID: &wantAccel, CapacityType: nebulav1alpha1.CapacityOnDemand, Region: ®ion, + Accelerator: &wantAccel, CapacityType: nebulav1alpha1.CapacityOnDemand, Region: ®ion, } spotRegional := provider.BlockScope{ - AcceleratorID: &wantAccel, CapacityType: nebulav1alpha1.CapacitySpot, Region: ®ion, + Accelerator: &wantAccel, CapacityType: nebulav1alpha1.CapacitySpot, Region: ®ion, } // A Spot capacity failure the Client wraps with both sentinels: no-capacity @@ -470,12 +481,26 @@ func TestClassifyProvisionError(t *testing.T) { want provider.BlockScope }{ {"auth is provider-wide (no region)", provider.ErrAuth, denyAll}, - {"quota is provider-wide (no region)", provider.ErrQuota, denyAll}, + // A vCPU/instance-limit quota is a regional, per-family, per-tier ceiling, so + // it blocks only this accelerator + tier in this region — never the whole + // provider. Otherwise one region's quota would strand every other region. + {"quota is regional OnDemand", provider.ErrQuota, onDemandRegional}, {"capacity is regional OnDemand", provider.ErrNoCapacity, onDemandRegional}, {"wrapped capacity is regional", fmt.Errorf("run: %w", provider.ErrNoCapacity), onDemandRegional}, {"spot capacity blocks only Spot in region", spotNoCapacity, spotRegional}, {"string no-capacity is regional OnDemand", stringNoCapacity, onDemandRegional}, - {"unknown is provider-wide", fmt.Errorf("weird transient blip"), denyAll}, + // An unrecognized error is confined to this accelerator + tier + region, NOT + // DenyAll: a whole-provider block on an unidentifiable failure is too broad, so + // failover routes around the one failing candidate instead. + {"unknown is regional OnDemand", fmt.Errorf("weird transient blip"), onDemandRegional}, + // InvalidFleetConfiguration, in a CreateFleet per-override error, means the + // instance type is not offered in that subnet's AZ — a zone-local availability + // gap classifyEC2Error maps to no-capacity, so it blocks only this + // accelerator/tier/region (a sibling AZ or region may still serve it), never + // DenyAll. + {"invalid fleet config is regional OnDemand", + &smithy.GenericAPIError{Code: "InvalidFleetConfiguration", Message: "not supported in AZ"}, + onDemandRegional}, {"nil", nil, provider.BlockScope{}}, } for _, tt := range tests { diff --git a/pkg/provider/aws/client.go b/pkg/provider/aws/client.go index e31eaa2..74c73cc 100644 --- a/pkg/provider/aws/client.go +++ b/pkg/provider/aws/client.go @@ -20,6 +20,8 @@ import ( "context" "errors" "fmt" + "sync" + "time" awssdk "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/config" @@ -63,6 +65,30 @@ type ec2API interface { DescribeSubnets( ctx context.Context, in *ec2.DescribeSubnetsInput, optFns ...func(*ec2.Options), ) (*ec2.DescribeSubnetsOutput, error) + // CreateLaunchTemplate / DeleteLaunchTemplate manage the EPHEMERAL launch + // template a CreateFleet launch references: RunInstance creates one carrying the + // per-Pod fields (AMI, user-data, tags) that CreateFleet's overrides cannot + // express, fleets against it, then deletes it — so no launch template survives + // the call and the self-configuring model is preserved. + CreateLaunchTemplate( + ctx context.Context, in *ec2.CreateLaunchTemplateInput, optFns ...func(*ec2.Options), + ) (*ec2.CreateLaunchTemplateOutput, error) + DeleteLaunchTemplate( + ctx context.Context, in *ec2.DeleteLaunchTemplateInput, optFns ...func(*ec2.Options), + ) (*ec2.DeleteLaunchTemplateOutput, error) + // DescribeLaunchTemplates lists the Nebula-tagged ephemeral templates so + // RunInstance can sweep any leaked by a crash between create and delete (see + // sweepStaleLaunchTemplates). + DescribeLaunchTemplates( + ctx context.Context, in *ec2.DescribeLaunchTemplatesInput, optFns ...func(*ec2.Options), + ) (*ec2.DescribeLaunchTemplatesOutput, error) + // CreateFleet launches instances across candidate AZs in ONE server-side call. + // RunInstance uses an "instant" fleet (synchronous: instances/errors returned in + // the response) to collapse the former per-AZ RunInstances sweep into a single + // request — EC2 itself iterates the subnet overrides for capacity. + CreateFleet( + ctx context.Context, in *ec2.CreateFleetInput, optFns ...func(*ec2.Options), + ) (*ec2.CreateFleetOutput, error) } // sdkClient is the real Client, backed by aws-sdk-go-v2's EC2 API. All @@ -70,9 +96,14 @@ type ec2API interface { // SDK-free; the pure translation (user-data, error mapping) lives in translate.go. // // Execution model (see the package doc): a NodeClaim maps to one EC2 instance -// launched DIRECTLY from RunInstances — no Launch Template, no pre-created infra. -// Everything the launch needs is SELF-CONFIGURED from credentials alone at -// construction, so registering AWS in a new region requires nothing but access: +// launched via a CreateFleet "instant" request, which lets EC2 do the per-AZ +// capacity search server-side in ONE call. CreateFleet cannot take an inline AMI / +// user-data (its overrides only cover instance type / subnet / price), so the +// launch references a launch template — but an EPHEMERAL one, created for this +// provision and deleted before the call returns. No launch template or other +// pre-created infra survives, so the self-configuring model holds: everything the +// launch needs is derived from credentials alone at construction, and registering +// AWS in a new region requires nothing but access: // // - amiID: the region's GPU AMI (NVIDIA driver + container runtime baked in), // resolved live via DescribeImages against the Amazon-owned ECS GPU-optimized @@ -101,6 +132,35 @@ type sdkClient struct { // default VPC: RunInstance then makes a single attempt letting EC2 pick the // subnet (still valid, just no zone failover). subnets []subnet + + // azOfferings caches, per instance type, the AZ-offering lookup for that type in + // this region (DescribeInstanceTypeOfferings at AZ granularity). RunInstance + // consults it to prune fleet overrides that target an AZ where the type is not + // offered at all — e.g. g6.48xlarge in us-east-1e — which EC2 would otherwise + // reject per-override with InvalidFleetConfiguration, wasting a grid slot on a + // permanently-dead pair. Offerings are stable per region, so each type is looked + // up at most once (populated lazily on first use). + // + // offeringsMu guards ONLY the map bookkeeping — never a network call. Each entry + // is single-flight: the first caller for a type inserts an entry with an open + // `done` channel, releases the lock, runs the one Describe, fills the entry, and + // closes `done`; concurrent callers for the same type find the entry and wait on + // `done` without the lock and without issuing a duplicate Describe. So a burst of + // N pods for a new type costs ONE lookup, and cache hits never block on I/O. + offeringsMu sync.Mutex + azOfferings map[string]*offeringEntry +} + +// offeringEntry is a single-flight cache slot for one instance type's AZ offerings. +// done is closed once the lookup finishes; readers block on it, then read azs/err +// (which are written exactly once, before done closes, so no lock is needed to read +// them afterwards). azs is nil on a failed lookup (err set) — a fail-open signal +// fleetOverrides reads as "unknown, don't prune". A resolved-but-empty azs means the +// type is offered in no AZ of this region. +type offeringEntry struct { + done chan struct{} + azs map[string]bool + err error } // subnet is one candidate launch target: its id and the AZ it lives in (for @@ -206,126 +266,367 @@ const gpuAMINameFilter = "amzn2-ami-ecs-gpu-hvm-*-x86_64-ebs" // for the other providers. var ErrConfig = errors.New("aws: not configured") -// RunInstance implements Client. It launches exactly one instance directly (from -// the resolved GPU AMI into a default-VPC subnet), overriding only the -// workload-specific fields. +// RunInstance implements Client. It launches exactly one instance via a +// CreateFleet "instant" request, so EC2 does the per-availability-zone capacity +// search server-side in ONE call rather than the adapter issuing a RunInstances +// per zone and sweeping them serially. // -// EC2 capacity is per-availability-zone, so a launch is not a single shot: when a -// zone reports InsufficientInstanceCapacity (or a Spot-capacity variant), -// RunInstance retries in the next discovered zone before giving up. Only when -// every candidate zone is capacity-starved does it return the wrapped -// ErrNoCapacity, so the outer region-level failover (ClassifyProvisionError + -// NodePool regions) fires only after this cheaper inner failover is exhausted. A -// non-capacity failure (auth, quota, bad config) is terminal and returned -// immediately — retrying other zones would not help and only burns the deadline. -// The ctx deadline (set by the vnode handler from Capabilities.ProvisionTimeout) -// bounds the whole sweep. +// EC2 capacity is per-AZ, so the fleet is given one launch-template OVERRIDE per +// discovered default-VPC subnet: EC2 tries them itself and returns whichever it +// lands, collapsing the former N-call zone sweep into a single round-trip that no +// longer burns the provision deadline attempt-by-attempt. When no subnets were +// discovered (no default VPC) the fleet runs with no overrides, letting EC2 pick +// the subnet — a valid launch, just without cross-AZ capacity search. // -// When no subnets were discovered (no default VPC), the loop runs once with a -// zero-value target, letting EC2 pick the subnet — a valid single-shot launch, -// just without zone failover. +// Because CreateFleet's overrides cannot carry an AMI or user-data, the launch +// references an ephemeral launch template built here (createLaunchTemplate) and +// torn down before returning (best-effort; a stray template is swept by +// name/tag). An instant fleet reports capacity shortfalls in the response's +// Errors (not as a Go error), so a fleet that returns no instance is mapped back +// through classifyEC2Error to the wrapped ErrNoCapacity the region-level failover +// (ClassifyProvisionError + NodePool regions) expects. A non-capacity failure +// (auth, quota, bad config) surfaces the same way and is terminal. func (c *sdkClient) RunInstance(ctx context.Context, spec InstanceSpec) (string, error) { userData, err := buildUserData(spec) if err != nil { return "", err } - targets := c.subnets - if len(targets) == 0 { - targets = []subnet{{}} // single attempt, letting EC2 pick the subnet + log := logf.FromContext(ctx).WithName("aws-run").WithValues( + "region", c.region, "instanceTypes", spec.InstanceTypes, "spot", spec.Spot) + + // Opportunistically reap any ephemeral launch template a prior provision leaked + // (crash between create and delete). Best-effort and staleness-gated, so it + // cannot touch a concurrent provision's own template; see sweepStaleLaunchTemplates. + c.sweepStaleLaunchTemplates(ctx) + + ltName, err := c.createLaunchTemplate(ctx, spec, userData) + if err != nil { + return "", err // classifyEC2Error not needed: template creation is not a capacity path } + // Best-effort teardown: no launch template must survive the provision. A delete + // failure only leaves a cheap, uniquely-named artifact behind, so it is logged, + // not surfaced — failing the provision over a dangling template would be worse. + defer func() { + if _, derr := c.ec2.DeleteLaunchTemplate(ctx, &ec2.DeleteLaunchTemplateInput{ + LaunchTemplateName: awssdk.String(ltName), + }); derr != nil { + log.Error(derr, "deleting ephemeral launch template; it will need sweeping", "launchTemplate", ltName) + } + }() - log := logf.FromContext(ctx).WithName("aws-run").WithValues( - "region", c.region, "instanceType", spec.InstanceType, "spot", spec.Spot) - log.V(1).Info("starting zone sweep", "candidateZones", len(targets)) - - var lastErr error - for i, sn := range targets { - // A zone attempt logs its start and its outcome so a capacity sweep across AZs - // is traceable (which zones were tried, in order, and why each was abandoned). - // zone "" means no default VPC was discovered and EC2 is picking the subnet. - zone := sn.az - if zone == "" { - zone = "" - } - log.V(1).Info("zone attempt starting", "zone", zone, "subnet", sn.id, "attempt", i+1, "of", len(targets)) - - // Honor the deadline between attempts: a slow first zone must not let the - // sweep overrun the provision timeout. - if err := ctx.Err(); err != nil { - log.Info("zone sweep aborted before completion; provision deadline reached", - "zone", zone, "attempted", i, "of", len(targets)) - if lastErr != nil { - return "", lastErr + // Prune subnets whose AZ does not OFFER a requested type before building the grid, + // so we never emit a (type, subnet) override EC2 would reject outright with + // InvalidFleetConfiguration (e.g. g6.48xlarge in us-east-1e). Fail-open: a lookup + // error yields nil offerings and offeredAZs keeps every subnet, so a transient + // DescribeInstanceTypeOfferings failure can never shrink the grid below today's. + offerings := c.instanceTypeAZs(ctx, spec.InstanceTypes) + + // One override per (instance type, subnet) pair => EC2 searches that whole grid + // for capacity in a single call, landing on whichever pair is available. The + // instance types are interchangeable alternates the accelerator maps to (primary + // first); spanning them broadens the launch beyond a single type's per-AZ + // capacity. The launch template carries no instance type — it lives here in the + // overrides — so a template built once serves every type. No subnets (no default + // VPC) => one override per type with the subnet unset, letting EC2 pick it (still + // no cross-AZ search). + overrides := fleetOverrides(spec.InstanceTypes, c.subnets, offerings) + log.V(1).Info("launching via instant fleet", + "candidateInstanceTypes", len(spec.InstanceTypes), "candidateSubnets", len(c.subnets), + "overrides", len(overrides)) + + out, err := c.ec2.CreateFleet(ctx, &ec2.CreateFleetInput{ + Type: ec2types.FleetTypeInstant, // synchronous: instances/errors in the response + TargetCapacitySpecification: &ec2types.TargetCapacitySpecificationRequest{ + TotalTargetCapacity: awssdk.Int32(1), + DefaultTargetCapacityType: fleetCapacityType(spec.Spot), + }, + LaunchTemplateConfigs: []ec2types.FleetLaunchTemplateConfigRequest{{ + LaunchTemplateSpecification: &ec2types.FleetLaunchTemplateSpecificationRequest{ + LaunchTemplateName: awssdk.String(ltName), + Version: awssdk.String("$Latest"), + }, + Overrides: overrides, + }}, + }) + if err != nil { + // A top-level error is the whole request failing (throttle, malformed, auth + // on the fleet API itself); classify so a capacity-shaped error still fails + // over correctly. + return "", classifyEC2Error(err, spec.Spot) + } + + if id := firstFleetInstanceID(out); id != "" { + log.Info("instant fleet launched instance", "instanceID", id) + return id, nil + } + + // No instance: an instant fleet reports why in out.Errors. Map the reported + // error code back through the same classifier so a capacity shortfall carries + // ErrNoCapacity (and, for Spot, ErrSpotCapacity) exactly as the old per-zone + // sweep did, letting region/tier failover proceed. + err = fleetError(out, spec.Spot) + log.Info("instant fleet launched no instance", "error", err.Error()) + return "", err +} + +// staleLaunchTemplateAge bounds how old an ephemeral launch template must be +// before sweepStaleLaunchTemplates deletes it. It is far longer than any in-flight +// provision could take (the provision deadline is ProvisionTimeout, a couple of +// minutes), so the sweep can never race a live provision's own template: anything +// older than this was provably orphaned by a crash between create and delete. +const staleLaunchTemplateAge = 30 * time.Minute + +// sweepStaleLaunchTemplates deletes Nebula-tagged ephemeral launch templates older +// than staleLaunchTemplateAge. RunInstance's defer already deletes the template it +// created in the happy path; this reaps the rare leak — a crash (or a persistent +// DeleteLaunchTemplate failure) between create and delete — that the defer cannot. +// +// It runs opportunistically on the (infrequent) provision path rather than the +// per-tick poll loop, so a leak is bounded to "cleaned up by the next provision in +// this region" without adding a DescribeLaunchTemplates to every poll. It is +// entirely best-effort: any error is logged and swallowed so a sweep hiccup never +// fails the provision it is piggybacking on. The age gate is what keeps it safe — +// it never touches a template young enough to belong to a concurrent provision. +func (c *sdkClient) sweepStaleLaunchTemplates(ctx context.Context) { + log := logf.FromContext(ctx).WithName("aws-lt-sweep").WithValues("region", c.region) + + in := &ec2.DescribeLaunchTemplatesInput{ + Filters: []ec2types.Filter{{ + Name: awssdk.String("tag-key"), + Values: []string{ClaimTagKey}, + }}, + } + cutoff := time.Now().Add(-staleLaunchTemplateAge) + for { + page, err := c.ec2.DescribeLaunchTemplates(ctx, in) + if err != nil { + log.V(1).Info("stale launch-template sweep skipped; describe failed", "error", err.Error()) + return + } + for _, lt := range page.LaunchTemplates { + if lt.LaunchTemplateName == nil || lt.CreateTime == nil { + continue + } + if lt.CreateTime.After(cutoff) { + continue // young enough to belong to an in-flight provision; leave it + } + if _, err := c.ec2.DeleteLaunchTemplate(ctx, &ec2.DeleteLaunchTemplateInput{ + LaunchTemplateName: lt.LaunchTemplateName, + }); err != nil { + log.V(1).Info("failed to delete a stale launch template; will retry next sweep", + "launchTemplate", *lt.LaunchTemplateName, "error", err.Error()) + continue } - return "", err - } - - id, err := c.runInSubnet(ctx, spec, userData, sn) - if err == nil { - log.Info("zone attempt succeeded", "zone", zone, "subnet", sn.id, "instanceID", id) - return id, nil - } - classified := classifyEC2Error(err, spec.Spot) - // Only a ZONE-LOCAL capacity shortfall justifies trying the next zone: a - // sibling AZ's default subnet may still satisfy the launch. Everything else - // is terminal for this sweep and surfaced immediately — auth/quota/unknown - // (retrying other zones would not help), and crucially a region/account- - // scoped capacity error (a Spot price ceiling or per-region Spot limit), for - // which every AZ would fail identically. Stopping now hands such an error - // straight to region/tier failover instead of burning the deadline on a - // futile sweep. It still carries ErrNoCapacity, so ClassifyProvisionError - // blocks the region correctly. - if !errors.Is(classified, errZoneLocal) { - log.Info("zone attempt failed with a non-zone-local error; stopping sweep", - "zone", zone, "subnet", sn.id, "error", classified.Error()) - return "", classified - } - log.Info("zone attempt failed with a zone-local capacity shortfall; trying next zone", - "zone", zone, "subnet", sn.id, "remainingZones", len(targets)-i-1) - lastErr = classified - } - // Every zone was capacity-starved; return the last capacity error so - // ClassifyProvisionError can block the region and region-failover can proceed. - log.Info("zone sweep exhausted; every candidate zone was capacity-starved", - "zonesTried", len(targets)) - return "", lastErr + log.Info("swept a leaked ephemeral launch template", "launchTemplate", *lt.LaunchTemplateName) + } + if page.NextToken == nil { + break + } + in.NextToken = page.NextToken + } } -// runInSubnet issues one RunInstances attempt. sn.id empty => let EC2 pick the -// subnet (no default VPC discovered); otherwise launch into that subnet (and thus -// its AZ). SecurityGroupIds is deliberately unset: launching into a default-VPC -// subnet attaches that VPC's default SG (outbound-open, no inbound exposure), -// which is all a fire-and-forget container launch needs. -func (c *sdkClient) runInSubnet(ctx context.Context, spec InstanceSpec, userData string, sn subnet) (string, error) { - in := &ec2.RunInstancesInput{ - MinCount: awssdk.Int32(1), - MaxCount: awssdk.Int32(1), - ImageId: awssdk.String(c.amiID), - InstanceType: ec2types.InstanceType(spec.InstanceType), - UserData: awssdk.String(userData), - TagSpecifications: []ec2types.TagSpecification{{ +// createLaunchTemplate creates the ephemeral launch template a CreateFleet launch +// references, carrying the per-Pod fields the fleet's overrides cannot express +// (AMI, user-data, identity tags), and returns its name. The +// name is derived from the claim so a retried provision reuses a stable name; a +// leftover template from a crashed prior attempt collides on create, which is +// treated as reuse rather than an error. +// +// SecurityGroupIds is deliberately unset: launching into a default-VPC subnet +// attaches that VPC's default SG (outbound-open, no inbound exposure), which is +// all a fire-and-forget container launch needs. The subnet AND the instance type +// are chosen by the fleet overrides, not the template — one template serves every +// candidate instance type — so both are intentionally absent here. +func (c *sdkClient) createLaunchTemplate(ctx context.Context, spec InstanceSpec, userData string) (string, error) { + name := launchTemplateName(spec.Tags[ClaimTagKey]) + + // InstanceMarketOptions is deliberately NOT set here. With EC2 Fleet the Spot + // vs OnDemand market is selected by the fleet's DefaultTargetCapacityType (see + // RunInstance), and specifying it BOTH in the launch template and on the fleet + // is rejected with InvalidFleetConfiguration. Leaving it out also keeps the + // template tier-agnostic, so the stable per-claim template is safely reused when + // failover retries the same claim under the other tier. + ltData := &ec2types.RequestLaunchTemplateData{ + ImageId: awssdk.String(c.amiID), + UserData: awssdk.String(userData), + TagSpecifications: []ec2types.LaunchTemplateTagSpecificationRequest{{ ResourceType: ec2types.ResourceTypeInstance, Tags: ec2Tags(spec.Tags), }}, } - if sn.id != "" { - in.SubnetId = awssdk.String(sn.id) + + _, err := c.ec2.CreateLaunchTemplate(ctx, &ec2.CreateLaunchTemplateInput{ + LaunchTemplateName: awssdk.String(name), + LaunchTemplateData: ltData, + // Tag the template itself so a leaked one (delete failed / process crashed + // between create and delete) is identifiable for an out-of-band sweep. + TagSpecifications: []ec2types.TagSpecification{{ + ResourceType: ec2types.ResourceTypeLaunchTemplate, + Tags: ec2Tags(spec.Tags), + }}, + }) + if err != nil { + // A stable name means a crashed prior attempt may have left the template; + // AlreadyExists is reuse, not a failure — the fleet references it by name. + var apiErr smithy.APIError + if errors.As(err, &apiErr) && apiErr.ErrorCode() == "InvalidLaunchTemplateName.AlreadyExistsException" { + return name, nil + } + return "", err } - if spec.Spot { - in.InstanceMarketOptions = &ec2types.InstanceMarketOptionsRequest{ - MarketType: ec2types.MarketTypeSpot, + return name, nil +} + +// launchTemplateName derives the ephemeral template's name from the claim so it is +// stable across a retried provision (create then collides => reuse) and greppable +// for an out-of-band sweep of leaked templates. +func launchTemplateName(claim string) string { + return "nebula-" + claim +} + +// fleetOverrides builds the CreateFleet launch-template overrides: one per +// (instance type, subnet) pair so EC2 searches that whole grid for capacity in a +// single call and lands on whichever pair is available. The instance types are the +// interchangeable alternates the accelerator resolves to (primary first); spanning +// them widens the launch past any single type's per-AZ capacity. When no subnets +// were discovered (no default VPC) it emits one override per instance type with the +// subnet unset, so EC2 still tries every type — it just cannot search across AZs. +// +// offerings maps a type to the set of AZs that OFFER it (nil = unknown, don't prune; +// see instanceTypeAZs): a (type, subnet) pair whose AZ is not in the type's offered +// set is dropped so a permanently-unavailable pair (e.g. g6.48xlarge in us-east-1e) +// never becomes an override EC2 would reject with InvalidFleetConfiguration. Pruning +// is skipped for a type whose offered set is unknown (fail-open) OR when it would +// leave the type with no subnet at all — in that last case the type keeps its full +// subnet list so a stale/empty offerings map can never zero out the launch. +func fleetOverrides( + instanceTypes []string, subnets []subnet, offerings map[string]map[string]bool, +) []ec2types.FleetLaunchTemplateOverridesRequest { + var overrides []ec2types.FleetLaunchTemplateOverridesRequest + for _, it := range instanceTypes { + if len(subnets) == 0 { + overrides = append(overrides, ec2types.FleetLaunchTemplateOverridesRequest{ + InstanceType: ec2types.InstanceType(it), + }) + continue + } + usable := offeredSubnets(it, subnets, offerings) + for _, sn := range usable { + overrides = append(overrides, ec2types.FleetLaunchTemplateOverridesRequest{ + InstanceType: ec2types.InstanceType(it), + SubnetId: awssdk.String(sn.id), + }) } } + return overrides +} - out, err := c.ec2.RunInstances(ctx, in) - if err != nil { - return "", err // caller classifies (needs the raw error to detect capacity) +// offeredSubnets filters subnets to those in an AZ that offers it, per offerings. +// It fails open: an unknown offered set (nil — lookup skipped or failed) keeps every +// subnet, and a filter that would remove ALL subnets also keeps every subnet, so +// pruning only ever narrows a launch that still has at least one usable AZ left. +func offeredSubnets(it string, subnets []subnet, offerings map[string]map[string]bool) []subnet { + azs := offerings[it] + if azs == nil { + return subnets // offered set unknown: do not prune + } + kept := make([]subnet, 0, len(subnets)) + for _, sn := range subnets { + if azs[sn.az] { + kept = append(kept, sn) + } + } + if len(kept) == 0 { + return subnets // pruning would zero the launch: keep all, let EC2 decide + } + return kept +} + +// fleetCapacityType maps the Spot flag to the fleet's default target-capacity +// type. CreateFleet takes the tier here rather than a per-instance market option. +func fleetCapacityType(spot bool) ec2types.DefaultTargetCapacityType { + if spot { + return ec2types.DefaultTargetCapacityTypeSpot + } + return ec2types.DefaultTargetCapacityTypeOnDemand +} + +// firstFleetInstanceID digs the launched instance id out of an instant fleet's +// response, or "" when the fleet placed nothing (every candidate was starved / +// rejected — the reason is in out.Errors, read by fleetError). +func firstFleetInstanceID(out *ec2.CreateFleetOutput) string { + if out == nil { + return "" + } + for _, inst := range out.Instances { + for _, id := range inst.InstanceIds { + return id + } + } + return "" +} + +// fleetError turns an instant fleet that launched nothing into the classified +// error region/tier failover expects. An instant fleet does not fail the API call +// on a capacity shortfall — it succeeds and reports a PER-OVERRIDE reason for every +// (instance type, subnet) pair it could not place in out.Errors — so the reported +// codes are run back through classifyEC2Error to recover ErrNoCapacity / +// ErrSpotCapacity / ErrQuota / ErrAuth. An empty Errors list (no instance and no +// reason) is treated as a generic no-capacity so the caller still fails over rather +// than wedging. +// +// The grid's errors are usually a MIX — one subnet's AZ may not offer the type +// (InvalidFleetConfiguration) while the rest are genuinely capacity-starved +// (InsufficientInstanceCapacity), and a terminal reason (auth/quota) can hide +// behind them. Picking out.Errors[0] blindly would surface whichever AZ EC2 listed +// first, so the most AUTHORITATIVE error is chosen instead (auth > quota > capacity +// > unknown): a real terminal reason is never masked by a per-AZ availability gap, +// and its true message is preserved for the log/Event. +func fleetError(out *ec2.CreateFleetOutput, spot bool) error { + if out != nil { + var best *ec2types.CreateFleetError + bestRank := -1 + for i := range out.Errors { + e := &out.Errors[i] + if e.ErrorCode == nil { + continue + } + if r := fleetErrorRank(*e.ErrorCode); r > bestRank { + bestRank, best = r, e + } + } + if best != nil { + msg := "" + if best.ErrorMessage != nil { + msg = *best.ErrorMessage + } + return classifyEC2Error(apiErrorFromFleet(*best.ErrorCode, msg), spot) + } } - if len(out.Instances) == 0 || out.Instances[0].InstanceId == nil { - return "", errors.New("aws: RunInstances returned no instance id") + // No reason reported: assume capacity so the region is blocked and failover runs. + return wrapNoCapacity(errors.New("aws: instant fleet returned no instance"), spot, false) +} + +// fleetErrorRank orders CreateFleet per-override error codes by how authoritative +// the reason is, so fleetError surfaces the one that best explains the whole +// launch: auth (fails everywhere) > quota (region/account ceiling) > capacity +// (per-AZ shortfall, incl. an AZ that doesn't offer the type) > unknown. A higher +// rank wins. This never changes the derived BlockScope for a uniform grid; it only +// picks the representative error when the grid's reasons differ. +func fleetErrorRank(code string) int { + switch code { + case "UnauthorizedOperation", "AuthFailure", "Blocked", "OptInRequired", "PendingVerification": + return 3 + case "InstanceLimitExceeded", "VcpuLimitExceeded", "RequestLimitExceeded", + "SpotMaxPriceTooLow", "MaxSpotInstanceCountExceeded": + return 2 + case "InsufficientInstanceCapacity", "InsufficientHostCapacity", "Unsupported", "InvalidFleetConfiguration": + return 1 + default: + return 0 } - return *out.Instances[0].InstanceId, nil } // resolveGPUAMI finds the region's GPU AMI: the newest Amazon-owned image whose @@ -445,12 +746,25 @@ func (c *sdkClient) DescribeInstance(ctx context.Context, id string) (*EC2Instan // carrying the ClaimTagKey tag — the tag every Nebula instance is launched with — // so instances Nebula does not own are never returned, and pages through all // results. This is the engine of the poll loop. +// +// The instance-state filter excludes terminated/shutting-down instances, which EC2 +// keeps visible in DescribeInstances for ~1h after teardown. Without it a torn-down +// instance would still be counted as "live" every tick (inflating the poll count) +// and — worse, since claim names are reused across pod restarts — findByClaim could +// match a corpse and mis-report a freshly re-created Pod off a dead instance. Only +// states that represent a live (or resumable) instance are returned. func (c *sdkClient) ListInstances(ctx context.Context) ([]EC2Instance, error) { in := &ec2.DescribeInstancesInput{ - Filters: []ec2types.Filter{{ - Name: awssdk.String("tag-key"), - Values: []string{ClaimTagKey}, - }}, + Filters: []ec2types.Filter{ + { + Name: awssdk.String("tag-key"), + Values: []string{ClaimTagKey}, + }, + { + Name: awssdk.String("instance-state-name"), + Values: []string{"pending", "running", "stopping", "stopped"}, + }, + }, } var out []EC2Instance var ids []string @@ -567,6 +881,98 @@ func (c *sdkClient) AvailableInstanceTypes(ctx context.Context) (map[string]bool return out, nil } +// instanceTypeAZs returns, for each of the requested types, the set of AZs in this +// region that offer it — used by RunInstance to prune fleet overrides targeting an +// AZ that does not offer the type. Offerings are stable per region, so a resolved +// type is cached on the client (azOfferings) and never re-queried. +// +// It is fail-open by construction: a type whose DescribeInstanceTypeOfferings lookup +// fails is left OUT of the returned map (its entry stays nil), which fleetOverrides +// reads as "unknown, don't prune". A transient API error therefore degrades to +// today's behavior (full grid), never to a shrunk or empty launch. Each miss is one +// AZ-scoped Describe (filtered to the single type), cached thereafter. +func (c *sdkClient) instanceTypeAZs(ctx context.Context, instanceTypes []string) map[string]map[string]bool { + log := logf.FromContext(ctx).WithName("aws-offerings").WithValues("region", c.region) + + out := map[string]map[string]bool{} + for _, it := range instanceTypes { + e, mine := c.offeringSlot(it) + if mine { + // This goroutine owns the single-flight lookup for it: run the one Describe + // OUTSIDE the lock, fill the entry, and release everyone waiting on done. + e.azs, e.err = c.describeTypeAZs(ctx, it) + if e.err != nil { + // Do NOT cache a failure: drop the slot so a later provision retries + // rather than fail-open forever on a transient throttle. + c.offeringsMu.Lock() + delete(c.azOfferings, it) + c.offeringsMu.Unlock() + } + close(e.done) + } else { + <-e.done // another goroutine is (or has) resolved it; wait without the lock. + } + + if e.err != nil { + // Fail-open: leave this type unresolved so fleetOverrides does not prune it. + log.V(1).Info("offering lookup failed; not pruning this type", "instanceType", it, "error", e.err.Error()) + continue + } + out[it] = e.azs // resolved set (may be empty: offered in no AZ of this region) + } + return out +} + +// offeringSlot returns the single-flight cache slot for an instance type, creating +// it if absent. The bool is true for the goroutine that CREATED the slot — the one +// responsible for running the lookup and closing e.done; all others receive the +// existing slot (mine=false) and must wait on e.done. Holds offeringsMu only for the +// map read/insert, never across the network call. +func (c *sdkClient) offeringSlot(it string) (*offeringEntry, bool) { + c.offeringsMu.Lock() + defer c.offeringsMu.Unlock() + if c.azOfferings == nil { + c.azOfferings = map[string]*offeringEntry{} + } + if e, ok := c.azOfferings[it]; ok { + return e, false + } + e := &offeringEntry{done: make(chan struct{})} + c.azOfferings[it] = e + return e, true +} + +// describeTypeAZs lists the AZs in this region that offer one instance type, via +// DescribeInstanceTypeOfferings at AZ granularity (LocationType=availability-zone), +// paging through all results. An empty result means the type is offered in no AZ of +// this region. +func (c *sdkClient) describeTypeAZs(ctx context.Context, instanceType string) (map[string]bool, error) { + in := &ec2.DescribeInstanceTypeOfferingsInput{ + LocationType: ec2types.LocationTypeAvailabilityZone, + Filters: []ec2types.Filter{{ + Name: awssdk.String("instance-type"), + Values: []string{instanceType}, + }}, + } + azs := map[string]bool{} + for { + page, err := c.ec2.DescribeInstanceTypeOfferings(ctx, in) + if err != nil { + return nil, err + } + for _, o := range page.InstanceTypeOfferings { + if o.Location != nil { + azs[*o.Location] = true + } + } + if page.NextToken == nil { + break + } + in.NextToken = page.NextToken + } + return azs, nil +} + // observe normalizes a live EC2 SDK instance into the adapter-level EC2Instance // view: id, tags, state name, region (from the placement AZ), a best-effort // public endpoint, and whether it is Spot. It is a point-in-time read and never diff --git a/pkg/provider/aws/client_test.go b/pkg/provider/aws/client_test.go index 3a27cb3..781ea1a 100644 --- a/pkg/provider/aws/client_test.go +++ b/pkg/provider/aws/client_test.go @@ -20,8 +20,11 @@ import ( "context" "encoding/base64" "errors" + "reflect" "strings" + "sync" "testing" + "time" awssdk "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/service/ec2" @@ -44,6 +47,7 @@ type fakeEC2 struct { describePages []*ec2.DescribeInstancesOutput describeErr error describeIdx int + lastDescribe *ec2.DescribeInstancesInput // statusPages seeds DescribeInstanceStatus (reachability checks); statusErr // injects a failure. Left nil, the fake reports no status rows — every instance @@ -58,6 +62,12 @@ type fakeEC2 struct { offeringPages []*ec2.DescribeInstanceTypeOfferingsOutput offeringErr error offeringIdx int + // offeringMu guards the offering* fields so a concurrency test can drive the + // single-flight lookup from many goroutines without racing the fake's counters. + offeringMu sync.Mutex + // offeringCalls counts DescribeInstanceTypeOfferings invocations (distinct calls, + // not pages) so a test can assert single-flight coalescing. + offeringCalls int // lastOfferingIn records the last DescribeInstanceTypeOfferings input so a test // can assert the region filter was applied. lastOfferingIn *ec2.DescribeInstanceTypeOfferingsInput @@ -66,6 +76,23 @@ type fakeEC2 struct { imagesErr error subnetsOut *ec2.DescribeSubnetsOutput subnetsErr error + + // fleet path. lastFleet records the last CreateFleet input; fleetOut/fleetErr + // seed its result. createdLTs/deletedLTs record ephemeral launch-template names + // so a test can assert one is created and torn down; ltCreateErr injects a + // CreateLaunchTemplate failure. + lastFleet *ec2.CreateFleetInput + fleetOut *ec2.CreateFleetOutput + fleetErr error + fleetCalls int + createdLTs []string + lastLTData ec2types.RequestLaunchTemplateData // last CreateLaunchTemplate data + deletedLTs []string + ltCreateErr error + + // describeLTsOut/Err seed DescribeLaunchTemplates for the stale-template sweep. + describeLTsOut *ec2.DescribeLaunchTemplatesOutput + describeLTsErr error } func (f *fakeEC2) DescribeImages( @@ -103,6 +130,53 @@ func (f *fakeEC2) RunInstances( return f.runOut, nil } +func (f *fakeEC2) CreateLaunchTemplate( + _ context.Context, in *ec2.CreateLaunchTemplateInput, _ ...func(*ec2.Options), +) (*ec2.CreateLaunchTemplateOutput, error) { + if f.ltCreateErr != nil { + return nil, f.ltCreateErr + } + if in.LaunchTemplateName != nil { + f.createdLTs = append(f.createdLTs, *in.LaunchTemplateName) + } + if in.LaunchTemplateData != nil { + f.lastLTData = *in.LaunchTemplateData + } + return &ec2.CreateLaunchTemplateOutput{}, nil +} + +func (f *fakeEC2) DeleteLaunchTemplate( + _ context.Context, in *ec2.DeleteLaunchTemplateInput, _ ...func(*ec2.Options), +) (*ec2.DeleteLaunchTemplateOutput, error) { + if in.LaunchTemplateName != nil { + f.deletedLTs = append(f.deletedLTs, *in.LaunchTemplateName) + } + return &ec2.DeleteLaunchTemplateOutput{}, nil +} + +func (f *fakeEC2) CreateFleet( + _ context.Context, in *ec2.CreateFleetInput, _ ...func(*ec2.Options), +) (*ec2.CreateFleetOutput, error) { + f.lastFleet = in + f.fleetCalls++ + if f.fleetErr != nil { + return nil, f.fleetErr + } + return f.fleetOut, nil +} + +func (f *fakeEC2) DescribeLaunchTemplates( + _ context.Context, _ *ec2.DescribeLaunchTemplatesInput, _ ...func(*ec2.Options), +) (*ec2.DescribeLaunchTemplatesOutput, error) { + if f.describeLTsErr != nil { + return nil, f.describeLTsErr + } + if f.describeLTsOut != nil { + return f.describeLTsOut, nil + } + return &ec2.DescribeLaunchTemplatesOutput{}, nil +} + func (f *fakeEC2) TerminateInstances( _ context.Context, in *ec2.TerminateInstancesInput, _ ...func(*ec2.Options), ) (*ec2.TerminateInstancesOutput, error) { @@ -114,8 +188,9 @@ func (f *fakeEC2) TerminateInstances( } func (f *fakeEC2) DescribeInstances( - _ context.Context, _ *ec2.DescribeInstancesInput, _ ...func(*ec2.Options), + _ context.Context, in *ec2.DescribeInstancesInput, _ ...func(*ec2.Options), ) (*ec2.DescribeInstancesOutput, error) { + f.lastDescribe = in if f.describeErr != nil { return nil, f.describeErr } @@ -145,7 +220,10 @@ func (f *fakeEC2) DescribeInstanceStatus( func (f *fakeEC2) DescribeInstanceTypeOfferings( _ context.Context, in *ec2.DescribeInstanceTypeOfferingsInput, _ ...func(*ec2.Options), ) (*ec2.DescribeInstanceTypeOfferingsOutput, error) { + f.offeringMu.Lock() + defer f.offeringMu.Unlock() f.lastOfferingIn = in + f.offeringCalls++ if f.offeringErr != nil { return nil, f.offeringErr } @@ -277,16 +355,185 @@ func TestClassifyEC2Error(t *testing.T) { } } +// fleetWith builds an instant-fleet output that launched a single instance id. +func fleetWith(id string) *ec2.CreateFleetOutput { + return &ec2.CreateFleetOutput{ + Instances: []ec2types.CreateFleetInstance{{InstanceIds: []string{id}}}, + } +} + +// fleetNoCapacity builds an instant-fleet output that launched NOTHING and reports +// the reason in Errors — how CreateFleet(instant) surfaces a capacity shortfall +// (the API call itself succeeds). +func fleetNoCapacity(code string) *ec2.CreateFleetOutput { + return &ec2.CreateFleetOutput{ + Errors: []ec2types.CreateFleetError{{ErrorCode: awssdk.String(code)}}, + } +} + +// offeringPage builds a DescribeInstanceTypeOfferings page listing the AZs (Location +// values) that offer a type — the AZ-granularity shape instanceTypeAZs reads. +func offeringPage(azs ...string) *ec2.DescribeInstanceTypeOfferingsOutput { + out := &ec2.DescribeInstanceTypeOfferingsOutput{} + for _, az := range azs { + out.InstanceTypeOfferings = append(out.InstanceTypeOfferings, ec2types.InstanceTypeOffering{ + Location: awssdk.String(az), + }) + } + return out +} + +func TestFleetOverrides_PrunesUnofferedAZ(t *testing.T) { + subnets := []subnet{ + {id: "sn-a", az: "us-east-1a"}, + {id: "sn-e", az: "us-east-1e"}, // g6.48xlarge not offered here + } + // Offerings say the type lives only in us-east-1a. + offerings := map[string]map[string]bool{"g6.48xlarge": {"us-east-1a": true}} + got := fleetOverrides([]string{"g6.48xlarge"}, subnets, offerings) + if len(got) != 1 || got[0].SubnetId == nil || *got[0].SubnetId != "sn-a" { + t.Fatalf("expected only the us-east-1a subnet to survive pruning, got %+v", got) + } +} + +func TestFleetOverrides_UnknownOfferingsDoesNotPrune(t *testing.T) { + subnets := []subnet{{id: "sn-a", az: "us-east-1a"}, {id: "sn-e", az: "us-east-1e"}} + // nil offered set for the type => unknown => fail-open, keep every subnet. + got := fleetOverrides([]string{"g6.48xlarge"}, subnets, nil) + if len(got) != 2 { + t.Fatalf("unknown offerings must not prune; want 2 overrides, got %d", len(got)) + } +} + +func TestFleetOverrides_EmptyOfferedSetKeepsAll(t *testing.T) { + subnets := []subnet{{id: "sn-a", az: "us-east-1a"}, {id: "sn-e", az: "us-east-1e"}} + // Resolved-but-empty offered set (type offered in no listed AZ) would zero the + // launch; fail-open keeps all subnets so a stale map can never wedge provisioning. + offerings := map[string]map[string]bool{"g6.48xlarge": {}} + got := fleetOverrides([]string{"g6.48xlarge"}, subnets, offerings) + if len(got) != 2 { + t.Fatalf("an empty offered set must keep all subnets (fail-open); got %d", len(got)) + } +} + +// RunInstance prunes the dead (type, AZ) pair using DescribeInstanceTypeOfferings, +// so the fleet grid never carries a subnet whose AZ does not offer the type; the +// resolved offering set is cached so a second provision issues no further lookup. +func TestSDKRunInstance_PrunesUnofferedAZAndCaches(t *testing.T) { + f := &fakeEC2{ + fleetOut: fleetWith("i-1"), + offeringPages: []*ec2.DescribeInstanceTypeOfferingsOutput{offeringPage("us-east-1a")}, + } + c := newSDKClient(f) + c.subnets = []subnet{{id: "sn-a", az: "us-east-1a"}, {id: "sn-e", az: "us-east-1e"}} + + if _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"g6.48xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c"}, + }); err != nil { + t.Fatalf("RunInstance: %v", err) + } + ovs := f.lastFleet.LaunchTemplateConfigs[0].Overrides + if len(ovs) != 1 || ovs[0].SubnetId == nil || *ovs[0].SubnetId != "sn-a" { + t.Fatalf("expected the us-east-1e override pruned, got %+v", ovs) + } + // The AZ-scoped offerings query filters to the single type. + if f.offeringCalls != 1 { + t.Fatalf("expected one offerings lookup, got %d", f.offeringCalls) + } + + // A second provision reuses the cached offered set — no further lookup. + if _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"g6.48xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c2"}, + }); err != nil { + t.Fatalf("second RunInstance: %v", err) + } + if f.offeringCalls != 1 { + t.Fatalf("offerings must be cached across provisions; got %d lookups", f.offeringCalls) + } +} + +// A burst of concurrent provisions for the same NEW type must coalesce into ONE +// DescribeInstanceTypeOfferings (single-flight): the first caller runs the lookup +// while the rest wait on it, so the shared cache is populated with a single call +// rather than one per goroutine. +func TestInstanceTypeAZs_SingleFlightCoalesces(t *testing.T) { + f := &fakeEC2{offeringPages: []*ec2.DescribeInstanceTypeOfferingsOutput{offeringPage("us-east-1a")}} + c := newSDKClient(f) + + const n = 8 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + got := c.instanceTypeAZs(context.Background(), []string{"g6.48xlarge"}) + if !got["g6.48xlarge"]["us-east-1a"] { + t.Errorf("expected us-east-1a in the resolved offered set, got %+v", got) + } + }() + } + wg.Wait() + + if f.offeringCalls != 1 { + t.Fatalf("single-flight must coalesce %d concurrent lookups into 1, got %d", n, f.offeringCalls) + } +} + +// A failed lookup must NOT be cached: the slot is dropped so a later provision +// retries the Describe rather than fail-open forever on a transient throttle. +func TestInstanceTypeAZs_ErrorNotCachedRetries(t *testing.T) { + f := &fakeEC2{ + offeringErr: errors.New("throttled"), + offeringPages: []*ec2.DescribeInstanceTypeOfferingsOutput{offeringPage("us-east-1a")}, + } + c := newSDKClient(f) + + // First lookup fails -> unresolved (fail-open) and not cached. + if got := c.instanceTypeAZs(context.Background(), []string{"g6.48xlarge"}); got["g6.48xlarge"] != nil { + t.Fatalf("a failed lookup must be unresolved (nil), got %+v", got["g6.48xlarge"]) + } + // Clear the injected error; the retry must re-issue the Describe and now succeed. + f.offeringMu.Lock() + f.offeringErr = nil + f.offeringMu.Unlock() + + got := c.instanceTypeAZs(context.Background(), []string{"g6.48xlarge"}) + if !got["g6.48xlarge"]["us-east-1a"] { + t.Fatalf("a retry after a transient error must resolve the offered set, got %+v", got) + } + if f.offeringCalls != 2 { + t.Fatalf("expected the failed lookup to be retried (2 calls), got %d", f.offeringCalls) + } +} + +// A DescribeInstanceTypeOfferings failure must fail open: the launch keeps every +// subnet (today's behavior) rather than shrinking or wedging. +func TestSDKRunInstance_OfferingLookupErrorFailsOpen(t *testing.T) { + f := &fakeEC2{fleetOut: fleetWith("i-1"), offeringErr: errors.New("throttled")} + c := newSDKClient(f) + c.subnets = []subnet{{id: "sn-a", az: "us-east-1a"}, {id: "sn-e", az: "us-east-1e"}} + + if _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"g6.48xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c"}, + }); err != nil { + t.Fatalf("RunInstance: %v", err) + } + if ovs := f.lastFleet.LaunchTemplateConfigs[0].Overrides; len(ovs) != 2 { + t.Fatalf("an offerings lookup error must not prune; want 2 overrides, got %d", len(ovs)) + } +} + func TestSDKRunInstance_OnDemand(t *testing.T) { - f := &fakeEC2{runOut: &ec2.RunInstancesOutput{ - Instances: []ec2types.Instance{{InstanceId: awssdk.String("i-42")}}, - }} + f := &fakeEC2{fleetOut: fleetWith("i-42")} c := newSDKClient(f) id, err := c.RunInstance(context.Background(), InstanceSpec{ - InstanceType: "p5.48xlarge", - Image: "img", - Tags: map[string]string{ClaimTagKey: "claim-a"}, + InstanceTypes: []string{"p5.48xlarge"}, + Image: "img", + Tags: map[string]string{ClaimTagKey: "claim-a"}, }) if err != nil { t.Fatalf("RunInstance: %v", err) @@ -294,100 +541,318 @@ func TestSDKRunInstance_OnDemand(t *testing.T) { if id != "i-42" { t.Fatalf("id = %q, want i-42", id) } - if f.lastRun.InstanceType != ec2types.InstanceType("p5.48xlarge") { - t.Fatalf("InstanceType = %q", f.lastRun.InstanceType) + // The launch goes through an instant fleet, not RunInstances. + if f.fleetCalls != 1 { + t.Fatalf("fleetCalls = %d, want 1", f.fleetCalls) + } + if f.lastFleet.Type != ec2types.FleetTypeInstant { + t.Fatalf("fleet Type = %q, want instant", f.lastFleet.Type) + } + // OnDemand: the fleet's default target-capacity type is on-demand. + if f.lastFleet.TargetCapacitySpecification.DefaultTargetCapacityType != ec2types.DefaultTargetCapacityTypeOnDemand { + t.Fatalf("capacity type = %q, want on-demand", f.lastFleet.TargetCapacitySpecification.DefaultTargetCapacityType) } - if f.lastRun.ImageId == nil || *f.lastRun.ImageId != "ami-123" { - t.Fatalf("AMI not set from resolved GPU AMI: %+v", f.lastRun.ImageId) + // Exactly one instance requested. + if got := f.lastFleet.TargetCapacitySpecification.TotalTargetCapacity; got == nil || *got != 1 { + t.Fatalf("TotalTargetCapacity = %v, want 1", got) } - // Security groups are deliberately unset: the default-VPC subnet supplies the - // default SG (outbound-open, no inbound exposure). - if len(f.lastRun.SecurityGroupIds) != 0 { - t.Fatalf("SecurityGroupIds must be unset, got %+v", f.lastRun.SecurityGroupIds) + // The ephemeral launch template is created and torn down: none survives. + if len(f.createdLTs) != 1 { + t.Fatalf("createdLTs = %v, want exactly one ephemeral template", f.createdLTs) } - // OnDemand: no spot market option. - if f.lastRun.InstanceMarketOptions != nil { - t.Fatalf("OnDemand request must not set market options") + if len(f.deletedLTs) != 1 || f.deletedLTs[0] != f.createdLTs[0] { + t.Fatalf("deletedLTs = %v, want the created template %v torn down", f.deletedLTs, f.createdLTs) } - // Claim tag rides on the instance tag spec. - if len(f.lastRun.TagSpecifications) == 0 || *f.lastRun.TagSpecifications[0].Tags[0].Value != "claim-a" { - t.Fatalf("claim tag not set: %+v", f.lastRun.TagSpecifications) + // The fleet references that template by name. + cfg := f.lastFleet.LaunchTemplateConfigs[0] + if cfg.LaunchTemplateSpecification.LaunchTemplateName == nil || + *cfg.LaunchTemplateSpecification.LaunchTemplateName != f.createdLTs[0] { + t.Fatalf("fleet references %+v, want template %q", cfg.LaunchTemplateSpecification, f.createdLTs[0]) } } -func TestSDKRunInstance_SpotSetsMarketOption(t *testing.T) { - f := &fakeEC2{runOut: &ec2.RunInstancesOutput{ - Instances: []ec2types.Instance{{InstanceId: awssdk.String("i-spot")}}, - }} +func TestSDKRunInstance_SpotSetsCapacityType(t *testing.T) { + f := &fakeEC2{fleetOut: fleetWith("i-spot")} c := newSDKClient(f) if _, err := c.RunInstance(context.Background(), InstanceSpec{ - InstanceType: "p5.48xlarge", Image: "img", Spot: true, + InstanceTypes: []string{"p5.48xlarge"}, Image: "img", Spot: true, Tags: map[string]string{ClaimTagKey: "c"}, }); err != nil { t.Fatalf("RunInstance: %v", err) } - if f.lastRun.InstanceMarketOptions == nil || - f.lastRun.InstanceMarketOptions.MarketType != ec2types.MarketTypeSpot { - t.Fatalf("Spot request must set spot market option: %+v", f.lastRun.InstanceMarketOptions) + // Spot is requested via the fleet's default target-capacity type. + if f.lastFleet.TargetCapacitySpecification.DefaultTargetCapacityType != ec2types.DefaultTargetCapacityTypeSpot { + t.Fatalf("Spot request must set spot capacity type: %q", + f.lastFleet.TargetCapacitySpecification.DefaultTargetCapacityType) } } -func TestSDKRunInstance_CapacityErrorWrapped(t *testing.T) { - f := &fakeEC2{runErr: apiErr("InsufficientInstanceCapacity")} - c := newSDKClient(f) +// One override per discovered subnet: EC2 does the per-AZ capacity search inside +// the single fleet call rather than the adapter sweeping RunInstances per zone. +func TestSDKRunInstance_OneOverridePerSubnet(t *testing.T) { + f := &fakeEC2{fleetOut: fleetWith("i-1")} + c := &sdkClient{ec2: f, region: testRegion, amiID: "ami-123", subnets: []subnet{ + {id: "subnet-a", az: "us-east-1a"}, + {id: "subnet-b", az: "us-east-1b"}, + {id: "subnet-c", az: "us-east-1c"}, + }} - _, err := c.RunInstance(context.Background(), InstanceSpec{ - InstanceType: "p5.48xlarge", Image: "img", Spot: true, + if _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p5.48xlarge"}, Image: "img", Tags: map[string]string{ClaimTagKey: "c"}, - }) - if !errors.Is(err, provider.ErrNoCapacity) || !errors.Is(err, ErrSpotCapacity) { - t.Fatalf("err = %v, want wrapped ErrNoCapacity + ErrSpotCapacity", err) + }); err != nil { + t.Fatalf("RunInstance: %v", err) + } + // Still ONE fleet call, regardless of subnet count. + if f.fleetCalls != 1 { + t.Fatalf("fleetCalls = %d, want 1 (server-side sweep, not per-zone)", f.fleetCalls) + } + overrides := f.lastFleet.LaunchTemplateConfigs[0].Overrides + if len(overrides) != 3 { + t.Fatalf("overrides = %d, want one per subnet (3)", len(overrides)) + } + for i, want := range []string{"subnet-a", "subnet-b", "subnet-c"} { + if overrides[i].SubnetId == nil || *overrides[i].SubnetId != want { + t.Fatalf("override[%d] subnet = %v, want %q", i, overrides[i].SubnetId, want) + } } } -// A zone-local capacity shortfall must make RunInstance sweep EVERY discovered -// subnet before giving up — a sibling AZ may still have capacity. -func TestSDKRunInstance_ZoneLocalSweepsAllSubnets(t *testing.T) { - f := &fakeEC2{runErr: apiErr("InsufficientInstanceCapacity")} +// Several interchangeable instance types are spanned in ONE fleet: the overrides +// are the (instance type, subnet) grid, so EC2 lands on whichever pair has +// capacity. The launch template carries no instance type — it lives in the +// overrides — so one template serves every candidate type. +func TestSDKRunInstance_SpansInstanceTypesAndSubnets(t *testing.T) { + f := &fakeEC2{fleetOut: fleetWith("i-1")} c := &sdkClient{ec2: f, region: testRegion, amiID: "ami-123", subnets: []subnet{ {id: "subnet-a", az: "us-east-1a"}, {id: "subnet-b", az: "us-east-1b"}, - {id: "subnet-c", az: "us-east-1c"}, }} + if _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p5.48xlarge", "p4d.24xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c"}, + }); err != nil { + t.Fatalf("RunInstance: %v", err) + } + overrides := f.lastFleet.LaunchTemplateConfigs[0].Overrides + // 2 types x 2 subnets = 4 overrides, primary type first. + type pair struct{ it, sn string } + got := make([]pair, 0, len(overrides)) + for _, o := range overrides { + sn := "" + if o.SubnetId != nil { + sn = *o.SubnetId + } + got = append(got, pair{string(o.InstanceType), sn}) + } + want := []pair{ + {"p5.48xlarge", "subnet-a"}, {"p5.48xlarge", "subnet-b"}, + {"p4d.24xlarge", "subnet-a"}, {"p4d.24xlarge", "subnet-b"}, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("overrides = %v, want the (type, subnet) grid %v", got, want) + } + // The launch template must NOT pin an instance type (the overrides do). + if it := f.lastLTData.InstanceType; it != "" { + t.Fatalf("launch template InstanceType = %q, want unset (overrides carry it)", it) + } +} + +// With no discovered subnets (no default VPC) the fleet still spans every +// instance type — one override per type, subnet unset for EC2 to pick. +func TestSDKRunInstance_NoSubnetsStillSpansTypes(t *testing.T) { + f := &fakeEC2{fleetOut: fleetWith("i-1")} + c := &sdkClient{ec2: f, region: testRegion, amiID: "ami-123"} // no subnets + + if _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p5.48xlarge", "p4d.24xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c"}, + }); err != nil { + t.Fatalf("RunInstance: %v", err) + } + overrides := f.lastFleet.LaunchTemplateConfigs[0].Overrides + if len(overrides) != 2 { + t.Fatalf("overrides = %d, want one per instance type (2) when no subnets", len(overrides)) + } + for i, want := range []string{"p5.48xlarge", "p4d.24xlarge"} { + if string(overrides[i].InstanceType) != want { + t.Fatalf("override[%d] type = %q, want %q", i, overrides[i].InstanceType, want) + } + if overrides[i].SubnetId != nil { + t.Fatalf("override[%d] subnet = %v, want unset (no default VPC)", i, *overrides[i].SubnetId) + } + } +} + +// A fleet that launches nothing reports the reason in out.Errors (the API call +// succeeds). A capacity code must map back to the wrapped ErrNoCapacity (+ Spot) +// so region/tier failover still fires — same contract the old per-zone sweep had. +func TestSDKRunInstance_FleetNoCapacityWrapped(t *testing.T) { + f := &fakeEC2{fleetOut: fleetNoCapacity("InsufficientInstanceCapacity")} + c := newSDKClient(f) + _, err := c.RunInstance(context.Background(), InstanceSpec{ - InstanceType: "p5.48xlarge", Image: "img", + InstanceTypes: []string{"p5.48xlarge"}, Image: "img", Spot: true, Tags: map[string]string{ClaimTagKey: "c"}, }) - if !errors.Is(err, provider.ErrNoCapacity) { - t.Fatalf("err = %v, want ErrNoCapacity after exhausting all zones", err) + if !errors.Is(err, provider.ErrNoCapacity) || !errors.Is(err, ErrSpotCapacity) { + t.Fatalf("err = %v, want wrapped ErrNoCapacity + ErrSpotCapacity", err) } - if f.runCalls != 3 { - t.Fatalf("runCalls = %d, want 3 (one attempt per subnet)", f.runCalls) + // The ephemeral template is torn down even on the no-capacity path. + if len(f.deletedLTs) != 1 { + t.Fatalf("deletedLTs = %v, want the template torn down on failure too", f.deletedLTs) } } -// A region/account-scoped Spot limit must STOP the sweep at the first subnet: -// every AZ would fail identically, so iterating them wastes the deadline. It still -// bubbles up ErrNoCapacity so region-level failover fires. -func TestSDKRunInstance_RegionScopedStopsSweep(t *testing.T) { - f := &fakeEC2{runErr: apiErr("MaxSpotInstanceCountExceeded")} - c := &sdkClient{ec2: f, region: testRegion, amiID: "ami-123", subnets: []subnet{ - {id: "subnet-a", az: "us-east-1a"}, - {id: "subnet-b", az: "us-east-1b"}, - {id: "subnet-c", az: "us-east-1c"}, - }} +// A CreateFleet grid returns a MIX of per-override reasons: one AZ does not offer +// the type (InvalidFleetConfiguration) while the rest are capacity-starved. Both are +// capacity-class, so the launch is still ErrNoCapacity (never a whole-provider block) +// and the AZ-availability error does not derail failover. +func TestSDKRunInstance_FleetMixedCapacityErrors(t *testing.T) { + f := &fakeEC2{fleetOut: &ec2.CreateFleetOutput{Errors: []ec2types.CreateFleetError{ + {ErrorCode: awssdk.String("InvalidFleetConfiguration"), + ErrorMessage: awssdk.String("g6.48xlarge is not supported in us-east-1e")}, + {ErrorCode: awssdk.String("InsufficientInstanceCapacity"), + ErrorMessage: awssdk.String("insufficient capacity in us-east-1d")}, + }}} + c := newSDKClient(f) _, err := c.RunInstance(context.Background(), InstanceSpec{ - InstanceType: "p5.48xlarge", Image: "img", Spot: true, + InstanceTypes: []string{"g6.48xlarge"}, Image: "img", Spot: true, Tags: map[string]string{ClaimTagKey: "c"}, }) if !errors.Is(err, provider.ErrNoCapacity) || !errors.Is(err, ErrSpotCapacity) { - t.Fatalf("err = %v, want ErrNoCapacity + ErrSpotCapacity", err) + t.Fatalf("err = %v, want wrapped ErrNoCapacity + ErrSpotCapacity for a mixed capacity grid", err) } - if f.runCalls != 1 { - t.Fatalf("runCalls = %d, want 1 (region-scoped error must not sweep)", f.runCalls) +} + +// A terminal reason (auth) hiding among per-AZ capacity errors must win: it fails +// everywhere, so fleetError surfaces it rather than the more numerous capacity +// errors, and RunInstance reports ErrAuth (a whole-provider block) not no-capacity. +func TestSDKRunInstance_FleetAuthOutranksCapacity(t *testing.T) { + f := &fakeEC2{fleetOut: &ec2.CreateFleetOutput{Errors: []ec2types.CreateFleetError{ + {ErrorCode: awssdk.String("InsufficientInstanceCapacity")}, + {ErrorCode: awssdk.String("UnauthorizedOperation"), + ErrorMessage: awssdk.String("not authorized to launch")}, + }}} + c := newSDKClient(f) + + _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"g6.48xlarge"}, Image: "img", Spot: true, + Tags: map[string]string{ClaimTagKey: "c"}, + }) + if !errors.Is(err, provider.ErrAuth) { + t.Fatalf("err = %v, want ErrAuth to outrank the capacity errors in the grid", err) + } +} + +// A fleet that launches nothing AND reports no reason must still fail over: map it +// to a generic no-capacity rather than wedging on a nil error. +func TestSDKRunInstance_FleetEmptyIsNoCapacity(t *testing.T) { + f := &fakeEC2{fleetOut: &ec2.CreateFleetOutput{}} + c := newSDKClient(f) + + _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p5.48xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c"}, + }) + if !errors.Is(err, provider.ErrNoCapacity) { + t.Fatalf("err = %v, want ErrNoCapacity for an empty fleet result", err) + } +} + +// A top-level CreateFleet error (throttle/malformed/auth on the fleet API) is +// classified too, so a capacity-shaped top-level error also fails over. +func TestSDKRunInstance_FleetTopLevelErrorClassified(t *testing.T) { + f := &fakeEC2{fleetErr: apiErr("InsufficientInstanceCapacity")} + c := newSDKClient(f) + + _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p5.48xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c"}, + }) + if !errors.Is(err, provider.ErrNoCapacity) { + t.Fatalf("err = %v, want classified ErrNoCapacity", err) + } + // Template is still cleaned up after a top-level fleet failure. + if len(f.deletedLTs) != 1 { + t.Fatalf("deletedLTs = %v, want cleanup after top-level fleet error", f.deletedLTs) + } +} + +// A launch-template create failure aborts the provision before any fleet call and +// leaves nothing to clean up. +func TestSDKRunInstance_LaunchTemplateCreateFails(t *testing.T) { + f := &fakeEC2{ltCreateErr: errors.New("boom")} + c := newSDKClient(f) + + if _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p5.48xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c"}, + }); err == nil { + t.Fatal("expected the launch-template create error to abort the provision") + } + if f.fleetCalls != 0 { + t.Fatalf("fleetCalls = %d, want 0 (no fleet after template create failed)", f.fleetCalls) + } + if len(f.deletedLTs) != 0 { + t.Fatalf("deletedLTs = %v, want none (nothing was created)", f.deletedLTs) + } +} + +// RunInstance opportunistically sweeps leaked ephemeral templates: one older than +// the staleness cutoff is deleted; a young one (a possible concurrent provision) +// is left alone. The sweep must never fail the provision it piggybacks on. +func TestSDKRunInstance_SweepsStaleLaunchTemplates(t *testing.T) { + old := time.Now().Add(-2 * staleLaunchTemplateAge) // provably orphaned + young := time.Now().Add(-1 * time.Minute) // could be in-flight + f := &fakeEC2{ + fleetOut: fleetWith("i-1"), + describeLTsOut: &ec2.DescribeLaunchTemplatesOutput{ + LaunchTemplates: []ec2types.LaunchTemplate{ + {LaunchTemplateName: awssdk.String("nebula-stale"), CreateTime: &old}, + {LaunchTemplateName: awssdk.String("nebula-young"), CreateTime: &young}, + }, + }, + } + c := newSDKClient(f) + + if _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p5.48xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "claim-a"}, + }); err != nil { + t.Fatalf("RunInstance: %v", err) + } + // deletedLTs holds the swept stale template plus this provision's own template. + // The young one must NOT be deleted. + deleted := map[string]bool{} + for _, n := range f.deletedLTs { + deleted[n] = true + } + if !deleted["nebula-stale"] { + t.Fatalf("stale template not swept; deletedLTs = %v", f.deletedLTs) + } + if deleted["nebula-young"] { + t.Fatalf("young template was swept but may be in-flight; deletedLTs = %v", f.deletedLTs) + } +} + +// A sweep describe failure must be swallowed: the provision still succeeds. +func TestSDKRunInstance_SweepFailureIsNonFatal(t *testing.T) { + f := &fakeEC2{ + fleetOut: fleetWith("i-1"), + describeLTsErr: errors.New("throttled"), + } + c := newSDKClient(f) + + id, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p5.48xlarge"}, Image: "img", + Tags: map[string]string{ClaimTagKey: "c"}, + }) + if err != nil || id != "i-1" { + t.Fatalf("RunInstance = (%q, %v), want a successful launch despite the sweep failing", id, err) } } @@ -527,6 +992,36 @@ func TestSDKList_PagesAndFiltersByClaimTag(t *testing.T) { if len(list) != 2 || list[0].ID != "i-1" || list[1].ID != "i-2" { t.Fatalf("list = %+v, want both pages", list) } + + // The query must be scoped server-side to (a) the Nebula claim tag and (b) live + // instance states — terminated/shutting-down are excluded so torn-down instances + // (which EC2 keeps visible for ~1h) are neither counted nor matched by claim. + filters := map[string][]string{} + for _, fl := range f.lastDescribe.Filters { + filters[awssdk.ToString(fl.Name)] = fl.Values + } + if got := filters["tag-key"]; len(got) != 1 || got[0] != ClaimTagKey { + t.Fatalf("tag-key filter = %v, want [%s]", got, ClaimTagKey) + } + states := filters["instance-state-name"] + if len(states) == 0 { + t.Fatal("expected an instance-state-name filter to exclude terminated instances") + } + for _, s := range states { + if s == "terminated" || s == "shutting-down" { + t.Fatalf("instance-state-name filter must not include %q; got %v", s, states) + } + } + wantStates := map[string]bool{"pending": true, "running": true, "stopping": true, "stopped": true} + for _, s := range states { + if !wantStates[s] { + t.Fatalf("unexpected instance-state-name value %q; got %v", s, states) + } + delete(wantStates, s) + } + if len(wantStates) != 0 { + t.Fatalf("instance-state-name filter missing states %v; got %v", wantStates, states) + } } func TestSDKList_FoldsInStatusChecks(t *testing.T) { diff --git a/pkg/provider/aws/translate.go b/pkg/provider/aws/translate.go index e3db901..4347ae0 100644 --- a/pkg/provider/aws/translate.go +++ b/pkg/provider/aws/translate.go @@ -98,6 +98,22 @@ func shellQuote(s string) string { // distinguishes the two. var errZoneLocal = errors.New("aws: zone-local capacity shortfall") +// apiErrorFromFleet reconstructs a smithy.APIError from a CreateFleet per-attempt +// error so an instant fleet's failure (reported in out.Errors, not as a returned Go +// error) can be run through classifyEC2Error alongside the RunInstances-style +// errors. The fleet reuses the same EC2 error codes (InsufficientInstanceCapacity, +// SpotMaxPriceTooLow, ...), so the same classifier applies once the code is wrapped +// back into the smithy.APIError shape it reads. AWS's human-readable message is +// preserved (falling back to the code when absent) so the surfaced error names the +// ACTUAL reason — e.g. "g6.48xlarge is not supported in us-east-1e" — instead of +// echoing the bare code twice. +func apiErrorFromFleet(code, message string) error { + if message == "" { + message = code + } + return &smithy.GenericAPIError{Code: code, Message: message} +} + // classifyEC2Error maps a raw EC2 API error onto the shared provider sentinels so // the adapter's ClassifyProvisionError can derive a BlockScope uniformly. It // recognizes EC2's error codes (smithy.APIError) first, falling back to the @@ -130,7 +146,14 @@ func classifyEC2Error(err error, spot bool) error { } switch apiErr.ErrorCode() { - case "InsufficientInstanceCapacity", "InsufficientHostCapacity", "Unsupported": + case "InsufficientInstanceCapacity", "InsufficientHostCapacity", "Unsupported", + // InvalidFleetConfiguration, in a CreateFleet per-override error, is how EC2 + // reports "this instance type is not offered in the AZ of this subnet" (e.g. + // g6.48xlarge in us-east-1e) — an availability gap in ONE zone, exactly like + // Unsupported. A sibling AZ may still satisfy the launch, so it is zone-local + // no-capacity, not a malformed request: treat it as capacity so region/tier + // failover routes around it rather than surfacing an opaque config error. + "InvalidFleetConfiguration": return wrapNoCapacity(err, spot, true /* zoneLocal */) case "SpotMaxPriceTooLow", "MaxSpotInstanceCountExceeded": // Region/account-scoped Spot limits: NOT zone-local, so the sweep stops. diff --git a/pkg/provider/catalog/base.go b/pkg/provider/catalog/base.go index 1d78392..acbd742 100644 --- a/pkg/provider/catalog/base.go +++ b/pkg/provider/catalog/base.go @@ -77,25 +77,35 @@ func (b Base) Offerings(context.Context) ([]provider.Offering, error) { } // MapAccelerator translates a canonical accelerator request (type + count) into -// this provider's own accelerator id using the catalog as the mapping table. It -// finds the offering row whose AcceleratorType matches (case-insensitively) and -// whose GPUCount matches the request, and returns that row's AcceleratorID. When -// AcceleratorID is blank (a provider whose id equals the canonical name) it falls -// back to the canonical name, so an identity-mapped provider needs no per-name -// data. Because the mapping lives entirely in the CSV, a provider whose ids -// diverge (AWS instance types) does NOT need to override this — it just populates -// the accelerator_id and gpu_count columns. +// this provider's own accelerator ids using the catalog as the mapping table. It +// finds the offering rows whose AcceleratorType matches (case-insensitively) and +// whose GPUCount matches the request, and returns their AcceleratorIDs in catalog +// order — the PRIMARY (first matching row) first, then any interchangeable +// alternates, de-duplicated. When a row's AcceleratorID is blank (a provider whose +// id equals the canonical name) it falls back to the canonical name, so an +// identity-mapped provider needs no per-name data. Because the mapping lives +// entirely in the CSV, a provider whose ids diverge (AWS instance types) does NOT +// need to override this — it just populates the accelerator_id and gpu_count +// columns, and adds a row per alternate to widen the launch. // // Count matching honours the two catalog shapes (see Offering.GPUCount): a // provider that bakes the count into the offering (AWS: T4x1=g4dn.xlarge, // T4x8=g4dn.metal) emits one row per count, so the row whose GPUCount equals the -// request is the match — this is what keeps (L4, 1) and (L4, 8) on DISTINCT ids so -// one's capacity block does not exclude the other. A provider that takes the count -// as a free parameter (Modal) leaves GPUCount 0 on its single row; a 0-count row -// matches any requested count, since count is not a lookup dimension there. -// Returns ok=false when the provider offers no row for that (type, count). -func (b Base) MapAccelerator(canonical string, count int32) (providerAcceleratorID string, ok bool) { - for _, o := range b.Catalog.Offerings(b.ProviderName) { +// request is the match — this is what keeps (L4, 1) and (L4, 8) on DISTINCT primary +// ids so one's capacity block does not exclude the other. A provider that takes the +// count as a free parameter (Modal) leaves GPUCount 0 on its single row; a 0-count +// row matches any requested count, since count is not a lookup dimension there. +// +// Dedup collapses the catalog's per-capacity-type/per-region row duplicates (an +// instance type is the same object whether the row prices Spot or OnDemand). The +// primary — ids[0] — is the identity failover blocks on; alternates broaden a +// single launch but never the blocklist. Returns ok=false when the provider offers +// no row for that (type, count). +func (b Base) MapAccelerator(canonical string, count int32) (providerAcceleratorIDs []string, ok bool) { + offerings := b.Catalog.Offerings(b.ProviderName) + ids := make([]string, 0, len(offerings)) + seen := make(map[string]bool) + for _, o := range offerings { if !strings.EqualFold(o.AcceleratorType, canonical) { continue } @@ -104,10 +114,15 @@ func (b Base) MapAccelerator(canonical string, count int32) (providerAccelerator if o.GPUCount != 0 && o.GPUCount != count { continue } - if o.AcceleratorID != "" { - return o.AcceleratorID, true + id := o.AcceleratorID + if id == "" { + id = o.AcceleratorType + } + if seen[id] { + continue } - return o.AcceleratorType, true + seen[id] = true + ids = append(ids, id) } - return "", false + return ids, len(ids) > 0 } diff --git a/pkg/provider/catalog/catalog_test.go b/pkg/provider/catalog/catalog_test.go index f082af1..458f1bf 100644 --- a/pkg/provider/catalog/catalog_test.go +++ b/pkg/provider/catalog/catalog_test.go @@ -19,6 +19,7 @@ package catalog import ( "os" "path/filepath" + "reflect" "testing" nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" @@ -190,7 +191,7 @@ func TestBaseMapAccelerator_CaseInsensitive(t *testing.T) { } // A canonical accelerator supplied in any case must resolve, and the returned - // id is always the catalog's canonical casing — so a lowercase + // primary id is always the catalog's canonical casing — so a lowercase // accelerator-type label ("h100") maps cleanly to the provider's accelerator ("H100"). cases := map[string]struct { in string @@ -205,13 +206,44 @@ func TestBaseMapAccelerator_CaseInsensitive(t *testing.T) { for name, tc := range cases { t.Run(name, func(t *testing.T) { got, ok := base.MapAccelerator(tc.in, 1) - if ok != tc.wantOK || got != tc.want { - t.Fatalf("MapAccelerator(%q) = (%q, %v), want (%q, %v)", tc.in, got, ok, tc.want, tc.wantOK) + primary := "" + if len(got) > 0 { + primary = got[0] + } + if ok != tc.wantOK || primary != tc.want { + t.Fatalf("MapAccelerator(%q) = (%v, %v), want (%q, %v)", tc.in, got, ok, tc.want, tc.wantOK) } }) } } +// Several catalog rows for the SAME (type, count) but different instance types are +// interchangeable alternates: MapAccelerator returns them all, primary (first row) +// first, so an AWS launch fleet can span them and land on whichever has capacity. +func TestBaseMapAccelerator_AlternatesPrimaryFirst(t *testing.T) { + base := Base{ + ProviderName: "aws", + Catalog: fakeLookup{rows: []provider.Offering{ + // Primary first, then two alternates for L4 x1. The per-capacity-type + // duplicate (the second p4d row) must be deduped, not returned twice. + {AcceleratorType: "L4", AcceleratorID: "g6.xlarge", GPUCount: 1, CapacityType: nebulav1alpha1.CapacityOnDemand}, + {AcceleratorType: "L4", AcceleratorID: "g6.xlarge", GPUCount: 1, CapacityType: nebulav1alpha1.CapacitySpot}, + {AcceleratorType: "L4", AcceleratorID: "gr6.4xlarge", GPUCount: 1, CapacityType: nebulav1alpha1.CapacityOnDemand}, + // A different count must NOT bleed into the x1 result. + {AcceleratorType: "L4", AcceleratorID: "g6.48xlarge", GPUCount: 8, CapacityType: nebulav1alpha1.CapacityOnDemand}, + }}, + } + + got, ok := base.MapAccelerator("L4", 1) + if !ok { + t.Fatal("expected L4 x1 to resolve") + } + want := []string{"g6.xlarge", "gr6.4xlarge"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("MapAccelerator(L4, 1) = %v, want %v (primary first, deduped, count-filtered)", got, want) + } +} + func TestOfferings_CopyIsolation(t *testing.T) { c, err := Load() if err != nil { diff --git a/pkg/provider/errors.go b/pkg/provider/errors.go index 031a84c..38fa804 100644 --- a/pkg/provider/errors.go +++ b/pkg/provider/errors.go @@ -40,54 +40,63 @@ var ( // ErrAuth: credentials/authorization failed. Whole-provider — nothing will // succeed until it is fixed. ErrAuth = errors.New("provider: authentication failed") - // ErrQuota: account quota/limit reached. Whole-provider until quota frees up. + // ErrQuota: a resource quota/limit was reached. Scoped like a capacity failure + // (accelerator + tier), NOT whole-provider: cloud quotas are per-resource and, + // for a multi-region adapter, per-region — e.g. an AWS vCPU limit is a regional, + // per-instance-family, per-tier ceiling, so exhausting it in one region says + // nothing about the same request in another. The adapter confines it to the + // failing region (see aws.ClassifyProvisionError). Transient until quota frees up. ErrQuota = errors.New("provider: quota exceeded") ) // ClassifyError maps a provision error to the BlockScope it should be // blocklisted at, using the shared sentinels first and a string-heuristic // fallback for raw API messages that were not wrapped. It encodes the rule that -// a narrow failure (this accelerator has no capacity) must not disqualify other -// accelerators on the same provider, while auth/quota widen to the whole -// provider; an unrecognized error is treated conservatively as whole-provider so -// a failure does not hot-loop (the blocklist TTL bounds the blast radius). +// a narrow failure (this accelerator has no capacity, or its quota is exhausted) +// must not disqualify other accelerators — or, for a multi-region adapter, other +// regions — on the same provider. Only auth (which fails everywhere) widens to the +// whole provider via DenyAll; every other case — including an unrecognized error — +// is scoped to the failing accelerator/tier/region so failover can route around it +// rather than fencing off the entire provider (the blocklist TTL bounds it either +// way). // // It is the single place the SHARED part of a scope is derived — category -// (DenyAll vs capacity-scoped), the capacity tier, and the accelerator id — so +// (DenyAll vs capacity-scoped), the capacity tier, and the accelerator pool — so // every adapter's ClassifyProvisionError delegates here and then only decorates // with what is provider-specific (AWS adds its region). No scope is assembled -// anywhere else: the vnode handler resolves the accelerator id off the Pod and +// anywhere else: the vnode handler resolves the accelerator pool off the Pod and // passes it in, rather than mutating the returned scope. // // capacityType is the tier the failing request used; it is stamped onto // accelerator-scoped scopes so the block is precise (a Spot failure does not -// block OnDemand). acceleratorID is the provider's RESOLVED id for what served the -// failing request (the (type, count) mapped through MapAccelerator — an EC2 -// instance type on AWS, the canonical name on Modal; "" for a CPU-only Pod): on a -// capacity-scoped block a non-empty id becomes an exact-match pointer (narrowing -// the block to the instance type / capacity pool that actually failed), while "" -// leaves AcceleratorID nil ("not applicable") so the block does not widen across -// every accelerator. Auth/quota (DenyAll) ignores both, since bad credentials fail -// for every accelerator and tier. -func ClassifyError(err error, capacityType nebulav1alpha1.CapacityType, acceleratorID string) BlockScope { +// block OnDemand). accelerator is the request's POOL identity (type:count, e.g. +// "H100:8"; "" for a CPU-only Pod), NOT the provider's SKU id — so the block stays +// truthful when a launch spans several interchangeable instance types: on a +// capacity-scoped block a non-empty pool becomes an exact-match pointer (narrowing +// the block to the (type, count) pool that actually failed), while "" leaves +// Accelerator nil ("not applicable") so the block does not widen across every +// accelerator. Auth (DenyAll) ignores both, since bad credentials fail for +// every accelerator and tier; quota is scoped like capacity (per accelerator and +// tier, and per region once the adapter confines it). +func ClassifyError(err error, capacityType nebulav1alpha1.CapacityType, accelerator string) BlockScope { if err == nil { return BlockScope{} } // capacityScope builds an accelerator/tier-scoped block, promoting a non-empty - // accelerator id to an exact-match pointer and leaving it nil otherwise. + // accelerator pool to an exact-match pointer and leaving it nil otherwise. capacityScope := func() BlockScope { s := BlockScope{CapacityType: capacityType} - if acceleratorID != "" { - s.AcceleratorID = &acceleratorID + if accelerator != "" { + s.Accelerator = &accelerator } return s } switch { - case errors.Is(err, ErrAuth), errors.Is(err, ErrQuota): + case errors.Is(err, ErrAuth): return BlockScope{DenyAll: true} - case errors.Is(err, ErrNoCapacity), errors.Is(err, ErrUnsupportedAccelerator): + case errors.Is(err, ErrNoCapacity), errors.Is(err, ErrUnsupportedAccelerator), errors.Is(err, ErrQuota): return capacityScope() } @@ -97,11 +106,17 @@ func ClassifyError(err error, capacityType nebulav1alpha1.CapacityType, accelera case containsAny(msg, "unauthorized", "forbidden", "authentication", "invalid token", "api key"): return BlockScope{DenyAll: true} case containsAny(msg, "quota", "limit exceeded", "rate limit"): - return BlockScope{DenyAll: true} + return capacityScope() case containsAny(msg, "no capacity", "capacity", "unavailable", "out of", "no gpu"): return capacityScope() default: - return BlockScope{DenyAll: true} + // An unrecognized error is scoped like capacity (this accelerator + tier, and + // per region once the adapter confines it), NOT DenyAll. A DenyAll on an + // unknown error fences off the WHOLE provider — every region and accelerator — + // which is far too broad a blast radius for a failure we can't even identify + // (e.g. a transient malformed-request blip in one region). Failover past the + // one failing candidate is the safer default; the TTL still bounds it. + return capacityScope() } } diff --git a/pkg/provider/errors_test.go b/pkg/provider/errors_test.go index 6c52ade..44a993e 100644 --- a/pkg/provider/errors_test.go +++ b/pkg/provider/errors_test.go @@ -29,7 +29,7 @@ func TestClassifyError(t *testing.T) { const accel = "H100" // capacityScope is the expected accelerator-scoped block: the accelerator id is // promoted to an exact-match pointer. - capacityScope := BlockScope{CapacityType: tier, AcceleratorID: ptrStr(accel)} + capacityScope := BlockScope{CapacityType: tier, Accelerator: ptrStr(accel)} tests := []struct { name string err error @@ -37,14 +37,20 @@ func TestClassifyError(t *testing.T) { }{ {"nil", nil, BlockScope{}}, {"auth sentinel", ErrAuth, BlockScope{DenyAll: true}}, - {"quota sentinel", ErrQuota, BlockScope{DenyAll: true}}, + // Quota is scoped like capacity (per accelerator + tier), NOT DenyAll: cloud + // quotas are per-resource and, for a multi-region adapter, per-region, so one + // exhausted quota must not fence off other regions/accelerators. + {"quota sentinel", ErrQuota, capacityScope}, {"no-capacity sentinel", ErrNoCapacity, capacityScope}, {"unsupported sentinel", ErrUnsupportedAccelerator, capacityScope}, {"wrapped sentinel", fmt.Errorf("provision failed: %w", ErrNoCapacity), capacityScope}, {"string unauthorized", fmt.Errorf("HTTP 401 unauthorized"), BlockScope{DenyAll: true}}, - {"string quota", fmt.Errorf("account limit exceeded"), BlockScope{DenyAll: true}}, + {"string quota", fmt.Errorf("account limit exceeded"), capacityScope}, {"string capacity", fmt.Errorf("no capacity available"), capacityScope}, - {"unknown conservative", fmt.Errorf("weird transient blip"), BlockScope{DenyAll: true}}, + // An unrecognized error is scoped like capacity (this accelerator + tier), NOT + // DenyAll: a DenyAll would fence off the whole provider on a failure we can't + // even identify, so failover past the one failing candidate is the safer default. + {"unknown capacity-scoped", fmt.Errorf("weird transient blip"), capacityScope}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -64,12 +70,12 @@ func TestClassifyError_TierStamped(t *testing.T) { } } -// An empty accelerator (CPU-only Pod) must leave AcceleratorID nil ("not +// An empty accelerator (CPU-only Pod) must leave Accelerator nil ("not // applicable"), never a wildcard that would widen the block across every GPU. func TestClassifyError_EmptyAcceleratorStaysNil(t *testing.T) { got := ClassifyError(ErrNoCapacity, nebulav1alpha1.CapacityOnDemand, "") - if got.AcceleratorID != nil { - t.Fatalf("expected nil AcceleratorID for a CPU-only request, got %+v", got.AcceleratorID) + if got.Accelerator != nil { + t.Fatalf("expected nil Accelerator for a CPU-only request, got %+v", got.Accelerator) } if got.DenyAll || got.CapacityType != nebulav1alpha1.CapacityOnDemand { t.Fatalf("expected an OnDemand capacity scope, got %+v", got) diff --git a/pkg/provider/modal/modal.go b/pkg/provider/modal/modal.go index 4be7d6c..b719bf9 100644 --- a/pkg/provider/modal/modal.go +++ b/pkg/provider/modal/modal.go @@ -288,11 +288,13 @@ func (p *Provider) sandboxSpecFromPod(pod *corev1.Pod, req provider.ProvisionReq return SandboxSpec{}, fmt.Errorf("modal: %w", err) } if canonical != "" { - modalGPU, ok := p.MapAccelerator(canonical, count) + // Modal takes the count as a free parameter and has no interchangeable + // alternates, so it always maps to a single id; take the primary (ids[0]). + ids, ok := p.MapAccelerator(canonical, count) if !ok { return SandboxSpec{}, fmt.Errorf("modal: unsupported accelerator %q", canonical) } - spec.GPU = modalGPU + spec.GPU = ids[0] spec.GPUCount = count } // No annotation => CPU-only sandbox (GPU/"" GPUCount 0), handled naturally. diff --git a/pkg/provider/modal/modal_test.go b/pkg/provider/modal/modal_test.go index b3912ec..4885ece 100644 --- a/pkg/provider/modal/modal_test.go +++ b/pkg/provider/modal/modal_test.go @@ -268,12 +268,16 @@ func TestClassifyProvisionError(t *testing.T) { want provider.BlockScope }{ {"auth sentinel", provider.ErrAuth, denyAll}, - {"quota sentinel", provider.ErrQuota, denyAll}, + // Quota is scoped like capacity (accelerator + tier), not DenyAll: an + // exhausted quota for one accelerator must not fence off the whole provider. + {"quota sentinel", provider.ErrQuota, onDemand}, {"capacity sentinel", provider.ErrNoCapacity, onDemand}, {"wrapped capacity", fmt.Errorf("provision: %w", provider.ErrNoCapacity), onDemand}, {"string unauthorized", fmt.Errorf("401 unauthorized"), denyAll}, {"string no capacity", fmt.Errorf("no capacity available in region"), onDemand}, - {"unknown", fmt.Errorf("weird transient blip"), denyAll}, + // An unrecognized error is scoped like capacity, not DenyAll: a whole-provider + // block on an unidentifiable failure is too broad; failover routes around it. + {"unknown capacity-scoped", fmt.Errorf("weird transient blip"), onDemand}, {"nil", nil, provider.BlockScope{}}, } for _, tt := range tests { diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go index e50f339..d2a677c 100644 --- a/pkg/provider/provider.go +++ b/pkg/provider/provider.go @@ -87,17 +87,24 @@ type Provider interface { // --- Translation ----------------------------------------------------- // MapAccelerator translates a canonical Nebula accelerator request (type + - // count) into this provider's identifier for what actually serves it (e.g. - // RunPod's "NVIDIA H100 80GB HBM3", or AWS's "g6.48xlarge"). Count is part of - // the key because on some providers the GPU count is baked into the offering - // rather than a free knob: on AWS (L4, 1) and (L4, 8) resolve to DIFFERENT - // instance types (g6.xlarge vs g6.48xlarge) — distinct capacity pools — so the - // id must distinguish them. Providers that attach an arbitrary count to a single - // offering (Modal) ignore count and return the same id for every count. Returns - // ok=false if the provider does not offer that (type, count). The returned id is - // what failover keys a capacity block on, so two requests that share a capacity - // pool must return the same id and two that do not must differ. - MapAccelerator(canonical string, count int32) (providerAcceleratorID string, ok bool) + // count) into the provider ids that can serve it — the PRIMARY first, then any + // interchangeable alternates. Count is part of the key because on some providers + // the GPU count is baked into the offering rather than a free knob: on AWS + // (L4, 1) and (L4, 8) resolve to DIFFERENT instance types (g6.xlarge vs + // g6.48xlarge) — distinct capacity pools — so the id must distinguish them. + // Providers that attach an arbitrary count to a single offering (Modal) ignore + // count and return the same id for every count. Returns ok=false if the provider + // offers no id for that (type, count). + // + // Most of the system uses ids[0], the PRIMARY: it is what failover keys a + // capacity block on, so two requests that share a capacity pool must return the + // same primary and two that do not must differ. Any further ids are equivalent + // alternates a provider may try within a SINGLE provisioning attempt (AWS's + // launch fleet spans several instance types so EC2 lands on whichever has + // capacity) — they broaden one launch, they do NOT widen the blocklist: an + // alternate running dry never disables the primary. A provider with no alternates + // returns a single-element slice. + MapAccelerator(canonical string, count int32) (providerAcceleratorIDs []string, ok bool) // ClassifyProvisionError maps a Provision error to the granularity at which // the failing placement should be blocklisted. This keeps failover precise: @@ -266,18 +273,17 @@ type Offering struct { // mixes the two: a NodePool candidate is resolved to fully-concrete values before // it is matched against these patterns. type BlockScope struct { - // AcceleratorID: nil => not applicable (CPU-only Pod, or a DenyAll scope); - // &"" => wildcard, matches every accelerator; &"g6.48xlarge" => exactly that - // one. It is the provider's RESOLVED id for what serves the failing request — - // the (type, count) mapped through MapAccelerator (an EC2 instance type on AWS, - // the canonical GPU name on Modal) — NOT the bare accelerator type. Keying on - // the resolved id is what confines a capacity block to the pool that actually - // ran out: an L4x8 (g6.48xlarge) Spot shortage must not exclude L4x1 - // (g6.xlarge), a different instance type and capacity pool, though both are - // "L4". A capacity error carries no accelerator (it is classified from the error - // alone), so the adapter leaves this nil and the vnode handler fills it in from - // the failing Pod's resolved request. - AcceleratorID *string + // Accelerator: nil => not applicable (CPU-only Pod, or a DenyAll scope); + // &"" => wildcard, matches every accelerator; &"H100:8" => exactly that one + // pool. It is the request's POOL identity (type:count), NOT the provider's SKU + // id. Keying on the pool is what confines a capacity block to what actually ran + // out: an L4:8 Spot shortage must not exclude L4:1, a different pool, though both + // are "L4" — and it stays stable when a launch spans several interchangeable + // provider instance types (so a post-launch SKU choice never desyncs the key). A + // capacity error carries no accelerator (it is classified from the error alone), + // so the adapter leaves this nil and the vnode handler fills it in from the + // failing Pod's request. + Accelerator *string // CapacityType empty => blocks all capacity types. CapacityType nebulav1alpha1.CapacityType // Region: nil => the provider has no region axis (a region-simple provider like diff --git a/pkg/util/accelerator.go b/pkg/util/accelerator.go index 421bf2f..c56798c 100644 --- a/pkg/util/accelerator.go +++ b/pkg/util/accelerator.go @@ -24,6 +24,10 @@ import ( nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" ) +// acceleratorSep joins the accelerator type and count in the canonical +// "type:count" identity (e.g. "H100:8"). +const acceleratorSep = ":" + // NvidiaGPUResource is the extended-resource key a GPU count is expressed under. // It mirrors the ecosystem-standard nvidia.com/gpu so a Pod's accelerator count // drives both the scheduler's fit check (against the virtual node's advertised @@ -60,6 +64,21 @@ func AcceleratorRequest(pod *corev1.Pod) (accelerator string, count int32, err e return typ, n, nil } +// AcceleratorPool is the canonical identity of the accelerator POOL a request +// targets: the type and count joined as "type:count" (e.g. "H100:8"). It is the +// key the failover blocklist records/queries and the value the NodeClaim reports, +// chosen over the provider's resolved SKU id because a single launch may span +// several interchangeable instance types (AWS's fleet) — the pool identity stays +// truthful whichever alternate lands, and it keeps distinct (type, count) pairs on +// distinct keys so an H100:8 shortage never disqualifies H100:1. Returns "" for a +// CPU-only request (empty type), which has no accelerator pool. +func AcceleratorPool(accelerator string, count int32) string { + if accelerator == "" { + return "" + } + return fmt.Sprintf("%s%s%d", accelerator, acceleratorSep, count) +} + // gpuCount returns the largest nvidia.com/gpu quantity requested across the // Pod's containers, preferring limits and falling back to requests (Kubernetes // treats an extended resource's request and limit as equal, but a Pod may set diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index 3b8571d..a878ef4 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -53,7 +53,7 @@ const defaultPollInterval = 15 * time.Second // carries no BlocklistTTLAnnotation (an out-of-band Pod, or one placed before the // annotation existed). It mirrors FailoverPolicy.BlocklistTTL's own default so the // behaviour is identical whether or not the pool policy reached the handler. -const defaultBlocklistTTL = 10 * time.Minute +const defaultBlocklistTTL = 3 * time.Minute // Blocklister records a failed placement so the placement controller can fail over // to the next candidate instead of hot-looping against a provider that just @@ -596,22 +596,24 @@ func (h *Handler) recordBlock(ctx context.Context, pod *corev1.Pod, err error) { if h.blocklist == nil { return } - // The accelerator id and region are properties of the REQUEST, not the error; the - // provider cannot see them, so we resolve them off the Pod and hand them over. The - // accelerator (type + count from the Pod) is mapped through the provider to its - // RESOLVED id (an EC2 instance type on AWS), so the block is keyed on the concrete - // capacity pool that failed — an L4x8 shortage must not exclude L4x1. "" for either - // means "not applicable" (a CPU-only Pod, or a region-simple provider), which the - // provider treats accordingly. + // The accelerator pool and region are properties of the REQUEST, not the error; + // the provider cannot see them, so we resolve them off the Pod and hand them over. + // The block is keyed on the POOL identity (type:count, e.g. "H100:8") — the SAME + // key placement resolves and queries a candidate by (selectPlacement uses + // util.AcceleratorPool). A block filed under any other key would never be queried, + // and failover would silently re-place onto the candidate that just failed. We key + // on the pool, NOT the provider's resolved SKU, because a single launch may span + // several interchangeable instance types (AWS's fleet): the pool stays truthful + // whichever alternate lands, and a block is only written when the WHOLE launch + // failed (it succeeds if any alternate had capacity), so the pool key correctly + // names a request whose every equivalent option is dry. It keeps distinct + // (type, count) pools on distinct keys, so an H100:8 shortage never excludes + // H100:1. "" for either means "not applicable" (a CPU-only Pod, or a region-simple + // provider), which the provider treats accordingly. accel, count, _ := util.AcceleratorRequest(pod) - acceleratorID := "" - if accel != "" { - // ok=false only if the provider stopped offering the (type, count) since - // provisioning; leave the id "" and let the provider scope from the error. - acceleratorID, _ = h.prov.MapAccelerator(accel, count) - } + accelerator := util.AcceleratorPool(accel, count) region := pod.Annotations[nebulav1alpha1.RegionAnnotation] - scope := h.prov.ClassifyProvisionError(err, acceleratorID, region) + scope := h.prov.ClassifyProvisionError(err, accelerator, region) if scope == (provider.BlockScope{}) { // An empty scope means the error is not one we know how to blocklist; do not diff --git a/pkg/vnode/handler_test.go b/pkg/vnode/handler_test.go index 1c5afe4..7242196 100644 --- a/pkg/vnode/handler_test.go +++ b/pkg/vnode/handler_test.go @@ -81,7 +81,7 @@ func (f *fakeProvider) List(context.Context) ([]provider.Instance, error) { return f.list, f.listErr } func (f *fakeProvider) Offerings(context.Context) ([]provider.Offering, error) { return nil, nil } -func (f *fakeProvider) MapAccelerator(c string, _ int32) (string, bool) { return c, true } +func (f *fakeProvider) MapAccelerator(c string, _ int32) ([]string, bool) { return []string{c}, true } func (f *fakeProvider) ClassifyProvisionError(_ error, accel, region string) provider.BlockScope { f.classifyAccel = accel f.classifyRegion = region @@ -160,16 +160,18 @@ func TestCreatePod_ProvisionErrorSurfaces(t *testing.T) { func TestCreatePod_ProvisionFailureRecordsBlock(t *testing.T) { accel, region := "H100", "us-east-1" scope := provider.BlockScope{ - AcceleratorID: &accel, - CapacityType: nebulav1alpha1.CapacitySpot, - Region: ®ion, + Accelerator: &accel, + CapacityType: nebulav1alpha1.CapacitySpot, + Region: ®ion, } fp := &fakeProvider{provisionErr: errors.New("no capacity"), classifyScope: scope} bl := &recordingBlocklist{} h := NewHandler(fp, nil, bl) pod := testPod("default", "p1") - pod.Annotations = map[string]string{nebulav1alpha1.BlocklistTTLAnnotation: "3m"} + // A non-default TTL so this asserts the annotation is honored, not that it + // happens to equal defaultBlocklistTTL. + pod.Annotations = map[string]string{nebulav1alpha1.BlocklistTTLAnnotation: "7m"} pod.Labels = map[string]string{nebulav1alpha1.AcceleratorTypeLabel: "H100"} if err := h.CreatePod(context.Background(), pod); err == nil { @@ -178,10 +180,11 @@ func TestCreatePod_ProvisionFailureRecordsBlock(t *testing.T) { if bl.calls != 1 { t.Fatalf("expected exactly one Record call, got %d", bl.calls) } - // The handler must resolve the accelerator off the Pod and hand it to the - // provider (which owns scope assembly), not mutate the returned scope itself. - if fp.classifyAccel != "H100" { - t.Fatalf("handler passed accelerator %q to ClassifyProvisionError, want H100", fp.classifyAccel) + // The handler must resolve the accelerator POOL identity (type:count) off the Pod + // and hand it to the provider (which owns scope assembly), not mutate the returned + // scope itself. The Pod labels H100 with no explicit count, so the pool is "H100:1". + if fp.classifyAccel != "H100:1" { + t.Fatalf("handler passed accelerator %q to ClassifyProvisionError, want H100:1", fp.classifyAccel) } if bl.prov != "fake" { t.Fatalf("recorded provider = %q, want fake", bl.prov) @@ -190,8 +193,8 @@ func TestCreatePod_ProvisionFailureRecordsBlock(t *testing.T) { t.Fatalf("recorded scope = %+v, want %+v", bl.scope, scope) } // TTL comes from the Pod annotation the placement controller stamped. - if bl.ttl != 3*time.Minute { - t.Fatalf("recorded ttl = %v, want 3m from the annotation", bl.ttl) + if bl.ttl != 7*time.Minute { + t.Fatalf("recorded ttl = %v, want 7m from the annotation", bl.ttl) } } diff --git a/pkg/vnode/node.go b/pkg/vnode/node.go index 570b343..3f94295 100644 --- a/pkg/vnode/node.go +++ b/pkg/vnode/node.go @@ -41,6 +41,14 @@ import ( // informerResync is the full-resync period for the virtual node's informers. const informerResync = time.Minute +// podSyncWorkers is the number of concurrent workers the pod controller runs per +// queue (create/delete/status-sync). VK serializes work per pod key, so distinct +// pods provision in parallel while the same key never runs twice at once — this +// removes the head-of-line blocking where one slow (cold-region) provision stalled +// pods that would otherwise succeed instantly. Kept modest to bound concurrent +// CreateFleet bursts against the provider's API rate limits. +const podSyncWorkers = 8 + // NodeName returns the virtual node name for a provider: "nebula-". // One static node per provider (see docs/architecture.md §3); the scheduler // routes an ungated Pod to it via the ProviderLabel nodeSelector. @@ -166,7 +174,7 @@ func (r *Runner) run( ctx context.Context, pc *vknode.PodController, nc *vknode.NodeController, nodeSpec *corev1.Node, np *vknode.NaiveNodeProviderV2, ) error { - go pc.Run(ctx, 1) //nolint:errcheck + go pc.Run(ctx, podSyncWorkers) //nolint:errcheck select { case <-ctx.Done():