diff --git a/api/v1alpha1/nodepool_types.go b/api/v1alpha1/nodepool_types.go index cac8bca..779de1b 100644 --- a/api/v1alpha1/nodepool_types.go +++ b/api/v1alpha1/nodepool_types.go @@ -134,9 +134,12 @@ const ( // type, so a failed H100 request does not block A100 requests on the same // provider) for BlocklistTTL, then reconsiders it. This spec only tunes that. type FailoverPolicy struct { - // BlocklistTTL is how long a failed placement is excluded before the - // provider becomes a candidate for it again. - // +kubebuilder:default="3m" + // BlocklistTTL is the BASE duration a failed placement is excluded before the + // provider becomes a candidate for it again. The controller adds a random jitter + // (up to a minute) on top so Pods that failed for the same reason do not all + // retry the just-freed candidate in lockstep, so the effective exclusion is this + // value plus that jitter. + // +kubebuilder:default="30s" BlocklistTTL metav1.Duration `json:"blocklistTTL,omitempty"` } diff --git a/config/crd/bases/nebula.inftyai.com_nodepools.yaml b/config/crd/bases/nebula.inftyai.com_nodepools.yaml index 5d4586b..61e03c6 100644 --- a/config/crd/bases/nebula.inftyai.com_nodepools.yaml +++ b/config/crd/bases/nebula.inftyai.com_nodepools.yaml @@ -98,10 +98,13 @@ spec: RunPod reports no capacity) is temporarily excluded and re-tried. properties: blocklistTTL: - default: 3m + default: 30s description: |- - BlocklistTTL is how long a failed placement is excluded before the - provider becomes a candidate for it again. + BlocklistTTL is the BASE duration a failed placement is excluded before the + provider becomes a candidate for it again. The controller adds a random jitter + (up to a minute) on top so Pods that failed for the same reason do not all + retry the just-freed candidate in lockstep, so the effective exclusion is this + value plus that jitter. type: string type: object providers: diff --git a/config/samples/deployment.yaml b/config/samples/deployment.yaml index 82b9400..f45239b 100644 --- a/config/samples/deployment.yaml +++ b/config/samples/deployment.yaml @@ -38,7 +38,7 @@ spec: app: gpu-workload-sample nebula.inftyai.com/enabled: "true" nebula.inftyai.com/nodepool: sample - nebula.inftyai.com/accelerator-type: l4 + nebula.inftyai.com/accelerator-type: a100-40gb spec: # Do NOT set nodeName or a provider nodeSelector yourself — the placement # controller fills the nodeSelector in when it ungates the Pod. Setting diff --git a/pkg/provider/aws/aws.go b/pkg/provider/aws/aws.go index 1ffc669..c061ec4 100644 --- a/pkg/provider/aws/aws.go +++ b/pkg/provider/aws/aws.go @@ -126,8 +126,13 @@ type InstanceSpec struct { // 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 - // Command is the container command+args, from the Pod. + // Command is the Pod container's command, kept SEPARATE from Args to preserve + // Kubernetes container semantics: command overrides the image ENTRYPOINT (Docker + // --entrypoint), not its CMD. Empty means "use the image's own ENTRYPOINT". Command []string + // Args is the Pod container's args, appended after the entrypoint just as CMD + // arguments. Empty means "use the image's own CMD". + Args []string // Env is the environment, flattened from the Pod's container env. Env map[string]string // Spot requests interruptible capacity when true (OnDemand otherwise). @@ -703,7 +708,8 @@ func (p *Provider) instanceSpecFromPod(pod *corev1.Pod, req provider.ProvisionRe return InstanceSpec{ InstanceTypes: instanceTypes, Image: c.Image, - Command: append(append([]string{}, c.Command...), c.Args...), + Command: append([]string{}, c.Command...), + Args: append([]string{}, c.Args...), Env: env, Spot: req.CapacityType == nebulav1alpha1.CapacitySpot, Region: req.Region, diff --git a/pkg/provider/aws/aws_test.go b/pkg/provider/aws/aws_test.go index f4b9398..2083394 100644 --- a/pkg/provider/aws/aws_test.go +++ b/pkg/provider/aws/aws_test.go @@ -195,8 +195,13 @@ func TestProvision_MapsAcceleratorToInstanceType(t *testing.T) { if got := f.lastSpec.Tags[ClaimTagKey]; got != "claim-a" { t.Fatalf("claim tag = %q, want claim-a", got) } - if len(f.lastSpec.Command) != 2 || f.lastSpec.Command[0] != "run" { - t.Fatalf("command = %v", f.lastSpec.Command) + // Command and Args ride through SEPARATELY (no flattening), preserving the + // Pod's Kubernetes container semantics for the user-data renderer. + if len(f.lastSpec.Command) != 1 || f.lastSpec.Command[0] != "run" { + t.Fatalf("command = %v, want [run]", f.lastSpec.Command) + } + if len(f.lastSpec.Args) != 1 || f.lastSpec.Args[0] != "--flag" { + t.Fatalf("args = %v, want [--flag]", f.lastSpec.Args) } } diff --git a/pkg/provider/aws/client.go b/pkg/provider/aws/client.go index 74c73cc..0ee0629 100644 --- a/pkg/provider/aws/client.go +++ b/pkg/provider/aws/client.go @@ -295,6 +295,42 @@ func (c *sdkClient) RunInstance(ctx context.Context, spec InstanceSpec) (string, log := logf.FromContext(ctx).WithName("aws-run").WithValues( "region", c.region, "instanceTypes", spec.InstanceTypes, "spot", spec.Spot) + // 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 offeredSubnets keeps every subnet, so a transient + // DescribeInstanceTypeOfferings failure can never shrink the grid below today's. + // Done BEFORE any launch template is created so a region that offers none of the + // requested types costs neither the template create/delete pair nor a CreateFleet. + 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("preparing instant fleet", + "candidateInstanceTypes", len(spec.InstanceTypes), "candidateSubnets", len(c.subnets), + "overrides", len(overrides)) + + // No override survived the prune: every requested type is offered in NO configured + // AZ of this region (a known-empty offered set — offerings are stable and we just + // queried them). CreateFleet would only bounce with InvalidFleetConfiguration, so + // short-circuit to a region-scoped no-capacity (NOT zone-local: no sibling AZ can + // help) and let region/tier failover route elsewhere — without the doomed API call + // or its Warning event. A fail-open (unknown) offered set never lands here: it + // keeps every subnet, so overrides is non-empty. + if len(overrides) == 0 { + log.Info("skipping instant fleet: no requested type offered in any configured AZ of this region") + return "", wrapNoCapacity( + fmt.Errorf("aws: none of %v offered in any availability zone of %s", spec.InstanceTypes, c.region), + spec.Spot, false /* zoneLocal */) + } + // 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. @@ -315,25 +351,7 @@ func (c *sdkClient) RunInstance(ctx context.Context, spec InstanceSpec) (string, } }() - // 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)) + log.V(1).Info("launching via instant fleet", "overrides", len(overrides)) out, err := c.ec2.CreateFleet(ctx, &ec2.CreateFleetInput{ Type: ec2types.FleetTypeInstant, // synchronous: instances/errors in the response @@ -524,9 +542,16 @@ func fleetOverrides( } // 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. +// The two outcomes turn on whether the offered set is KNOWN: +// - unknown (nil — lookup skipped or failed): fail open, keep every subnet, so a +// transient DescribeInstanceTypeOfferings failure never shrinks the launch. +// - known: keep only the subnets whose AZ offers the type. When that intersection +// is EMPTY the type is offered in no configured AZ of this region, so return nil +// (no usable subnet) rather than the old fail-open "keep all". RunInstance reads +// an all-types-empty override set as a region no-capacity and skips the doomed +// CreateFleet — this is the source of the InvalidFleetConfiguration event churn +// (e.g. p4d.24xlarge in us-west-1, offered in neither AZ). A known-empty result +// is trustworthy: offerings are stable and we just queried them. func offeredSubnets(it string, subnets []subnet, offerings map[string]map[string]bool) []subnet { azs := offerings[it] if azs == nil { @@ -538,10 +563,7 @@ func offeredSubnets(it string, subnets []subnet, offerings map[string]map[string kept = append(kept, sn) } } - if len(kept) == 0 { - return subnets // pruning would zero the launch: keep all, let EC2 decide - } - return kept + return kept // may be empty: type offered in no configured AZ of this region } // fleetCapacityType maps the Spot flag to the fleet's default target-capacity diff --git a/pkg/provider/aws/client_test.go b/pkg/provider/aws/client_test.go index 781ea1a..544a161 100644 --- a/pkg/provider/aws/client_test.go +++ b/pkg/provider/aws/client_test.go @@ -247,6 +247,7 @@ func TestBuildUserData(t *testing.T) { spec := InstanceSpec{ Image: "myrepo/train:v1", Command: []string{"python", "train.py"}, + Args: []string{"--epochs", "3"}, Env: map[string]string{"B": "2", "A": "1"}, } encoded, err := buildUserData(spec) @@ -269,8 +270,55 @@ func TestBuildUserData(t *testing.T) { if idxA, idxB := strings.Index(script, "'A=1'"), strings.Index(script, "'B=2'"); idxA < 0 || idxB < 0 || idxA > idxB { t.Fatalf("env not rendered in sorted order:\n%s", script) } - if !strings.Contains(script, "'python' 'train.py'") { - t.Fatalf("script missing command:\n%s", script) + // Kubernetes semantics: Command[0] overrides the image ENTRYPOINT (--entrypoint), + // Command[1:] are its leading args, and Args follow as CMD arguments — the whole + // tail rendered after the image, in that order. + if !strings.Contains(script, "--entrypoint 'python'") { + t.Fatalf("command[0] not mapped to --entrypoint:\n%s", script) + } + if !strings.Contains(script, "'myrepo/train:v1' 'train.py' '--epochs' '3'") { + t.Fatalf("command tail/args not rendered after the image in order:\n%s", script) + } +} + +// TestBuildUserData_ArgsWithoutCommand covers a Pod that sets args but not command: +// the image's own ENTRYPOINT must run (no --entrypoint emitted) with args appended +// as CMD, exactly as a kubelet would launch it. +func TestBuildUserData_ArgsWithoutCommand(t *testing.T) { + spec := InstanceSpec{ + Image: "srv:v2", + Args: []string{"--port", "8080"}, + } + encoded, err := buildUserData(spec) + if err != nil { + t.Fatalf("buildUserData: %v", err) + } + raw, _ := base64.StdEncoding.DecodeString(encoded) + script := string(raw) + + if strings.Contains(script, "--entrypoint") { + t.Fatalf("no command was set; --entrypoint must not be emitted:\n%s", script) + } + if !strings.Contains(script, "'srv:v2' '--port' '8080'") { + t.Fatalf("args not appended after the image as CMD:\n%s", script) + } +} + +// TestBuildUserData_NoCommandOrArgs covers a bare image: the image's own +// ENTRYPOINT/CMD run, so the run line is just the image with no trailing tokens. +func TestBuildUserData_NoCommandOrArgs(t *testing.T) { + encoded, err := buildUserData(InstanceSpec{Image: "bare:v1"}) + if err != nil { + t.Fatalf("buildUserData: %v", err) + } + raw, _ := base64.StdEncoding.DecodeString(encoded) + script := string(raw) + + if strings.Contains(script, "--entrypoint") { + t.Fatalf("no command was set; --entrypoint must not be emitted:\n%s", script) + } + if !strings.Contains(script, "--gpus all 'bare:v1'\n") { + t.Fatalf("bare image must run with no trailing command/args:\n%s", script) } } @@ -405,14 +453,16 @@ func TestFleetOverrides_UnknownOfferingsDoesNotPrune(t *testing.T) { } } -func TestFleetOverrides_EmptyOfferedSetKeepsAll(t *testing.T) { +func TestFleetOverrides_KnownEmptyOfferedSetPrunesAll(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. + // Resolved-but-empty offered set: the type is offered in NO configured AZ. Unlike + // an unknown (nil) set, this is trustworthy, so it prunes to nothing rather than + // failing open — RunInstance reads the empty grid as a region no-capacity and skips + // the doomed CreateFleet instead of letting EC2 bounce it. 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)) + if len(got) != 0 { + t.Fatalf("a known-empty offered set must prune every subnet; got %d", len(got)) } } @@ -454,6 +504,38 @@ func TestSDKRunInstance_PrunesUnofferedAZAndCaches(t *testing.T) { } } +// When a requested type is offered in NO configured AZ of the region (a known-empty +// offered set, e.g. p4d.24xlarge in us-west-1), RunInstance must short-circuit to a +// region-scoped no-capacity WITHOUT creating a launch template or calling CreateFleet +// — the fleet would only bounce with InvalidFleetConfiguration. The error must carry +// ErrNoCapacity (so failover proceeds) but NOT errZoneLocal (no sibling AZ can help). +func TestSDKRunInstance_NoOfferedAZShortCircuits(t *testing.T) { + f := &fakeEC2{ + fleetOut: fleetWith("i-1"), // would succeed IF a fleet were launched + offeringPages: []*ec2.DescribeInstanceTypeOfferingsOutput{offeringPage()}, + } + c := newSDKClient(f) + c.subnets = []subnet{{id: "sn-a", az: "us-west-1a"}, {id: "sn-b", az: "us-west-1b"}} + + _, err := c.RunInstance(context.Background(), InstanceSpec{ + InstanceTypes: []string{"p4d.24xlarge"}, Image: "img", Spot: true, + Tags: map[string]string{ClaimTagKey: "c"}, + }) + if !errors.Is(err, provider.ErrNoCapacity) { + t.Fatalf("want ErrNoCapacity, got %v", err) + } + if errors.Is(err, errZoneLocal) { + t.Fatal("a region-wide unavailability must not be zone-local (no sibling AZ helps)") + } + // No doomed API calls: neither a launch template nor a fleet. + if f.fleetCalls != 0 { + t.Fatalf("CreateFleet must be skipped when no AZ offers the type; got %d calls", f.fleetCalls) + } + if len(f.createdLTs) != 0 { + t.Fatalf("no launch template must be created on the short-circuit; got %v", f.createdLTs) + } +} + // 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 @@ -591,7 +673,10 @@ func TestSDKRunInstance_SpotSetsCapacityType(t *testing.T) { // 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")} + f := &fakeEC2{ + fleetOut: fleetWith("i-1"), + offeringPages: []*ec2.DescribeInstanceTypeOfferingsOutput{offeringPage("us-east-1a", "us-east-1b", "us-east-1c")}, + } 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"}, @@ -624,7 +709,14 @@ func TestSDKRunInstance_OneOverridePerSubnet(t *testing.T) { // 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")} + f := &fakeEC2{ + fleetOut: fleetWith("i-1"), + // One offering page per type lookup (p5 then p4d), each offered in both AZs. + offeringPages: []*ec2.DescribeInstanceTypeOfferingsOutput{ + offeringPage("us-east-1a", "us-east-1b"), + offeringPage("us-east-1a", "us-east-1b"), + }, + } 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"}, diff --git a/pkg/provider/aws/translate.go b/pkg/provider/aws/translate.go index 4347ae0..ccb0444 100644 --- a/pkg/provider/aws/translate.go +++ b/pkg/provider/aws/translate.go @@ -55,15 +55,32 @@ func buildUserData(spec InstanceSpec) (string, error) { // Pull first so an image error surfaces before we try to run. fmt.Fprintf(&b, "docker pull %s\n", shellQuote(spec.Image)) - // docker run --gpus all -d, with env and command appended. Env keys are sorted - // so the rendered script is deterministic (stable across reconciles and easy to - // assert in tests). + // docker run --rm --gpus all, with env, then the image, then the workload's + // command/args. Kubernetes container semantics are preserved by mapping the two + // Pod fields the way the kubelet does, NOT by concatenating them: + // - Command (Pod command) overrides the image ENTRYPOINT => Docker --entrypoint, + // which takes a single program; Command[0] is the entrypoint and Command[1:] + // become leading arguments (Docker only lets --entrypoint carry the program). + // - Args (Pod args) are appended after as CMD arguments. + // When Command is empty the image's own ENTRYPOINT runs; when Args is empty the + // image's own CMD runs — so an image with a baked-in entrypoint launches exactly + // as it would under a kubelet. Env keys are sorted so the rendered script is + // deterministic (stable across reconciles and easy to assert in tests). b.WriteString("docker run --rm --gpus all") for _, k := range sortedKeys(spec.Env) { fmt.Fprintf(&b, " -e %s", shellQuote(k+"="+spec.Env[k])) } + // runArgs are everything that follows the image: the entrypoint's own arguments + // (Command[1:]) then the CMD arguments (Args), in that order. + var runArgs []string + if len(spec.Command) > 0 { + fmt.Fprintf(&b, " --entrypoint %s", shellQuote(spec.Command[0])) + runArgs = append(runArgs, spec.Command[1:]...) + } + runArgs = append(runArgs, spec.Args...) + b.WriteString(" " + shellQuote(spec.Image)) - for _, arg := range spec.Command { + for _, arg := range runArgs { b.WriteString(" " + shellQuote(arg)) } b.WriteString("\n") diff --git a/pkg/provider/catalog/base.go b/pkg/provider/catalog/base.go index acbd742..485b171 100644 --- a/pkg/provider/catalog/base.go +++ b/pkg/provider/catalog/base.go @@ -101,6 +101,13 @@ func (b Base) Offerings(context.Context) ([]provider.Offering, error) { // 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). +// +// Availability gates the mapping: a row whose Available is false (the catalog's +// availability column, e.g. a type we no longer wish to schedule onto) contributes +// no id, so a (type, count) whose every row is unavailable maps to ok=false. This +// is the single seam placement (servability check), AWS Provision and Modal +// Provision all consult, so flipping a CSV row's availability off removes it from +// scheduling everywhere without touching Go — no launch is attempted for it. func (b Base) MapAccelerator(canonical string, count int32) (providerAcceleratorIDs []string, ok bool) { offerings := b.Catalog.Offerings(b.ProviderName) ids := make([]string, 0, len(offerings)) @@ -109,6 +116,11 @@ func (b Base) MapAccelerator(canonical string, count int32) (providerAccelerator if !strings.EqualFold(o.AcceleratorType, canonical) { continue } + // An unavailable row cannot serve the request: skip it so its id is not + // offered up for a launch that the provider would only reject. + if !o.Available { + continue + } // GPUCount 0 => count is not a lookup dimension for this provider (matches any // requested count); otherwise the row's count must equal the request. if o.GPUCount != 0 && o.GPUCount != count { diff --git a/pkg/provider/catalog/catalog_test.go b/pkg/provider/catalog/catalog_test.go index 458f1bf..35b9c4e 100644 --- a/pkg/provider/catalog/catalog_test.go +++ b/pkg/provider/catalog/catalog_test.go @@ -185,8 +185,8 @@ func TestBaseMapAccelerator_CaseInsensitive(t *testing.T) { base := Base{ ProviderName: "modal", Catalog: fakeLookup{rows: []provider.Offering{ - {AcceleratorType: "H100", CapacityType: nebulav1alpha1.CapacityOnDemand}, - {AcceleratorType: "A100-80GB", CapacityType: nebulav1alpha1.CapacityOnDemand}, + {AcceleratorType: "H100", CapacityType: nebulav1alpha1.CapacityOnDemand, Available: true}, + {AcceleratorType: "A100-80GB", CapacityType: nebulav1alpha1.CapacityOnDemand, Available: true}, }}, } @@ -221,16 +221,17 @@ func TestBaseMapAccelerator_CaseInsensitive(t *testing.T) { // 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) { + od, sp := nebulav1alpha1.CapacityOnDemand, nebulav1alpha1.CapacitySpot 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}, + {AcceleratorType: "L4", AcceleratorID: "g6.xlarge", GPUCount: 1, CapacityType: od, Available: true}, + {AcceleratorType: "L4", AcceleratorID: "g6.xlarge", GPUCount: 1, CapacityType: sp, Available: true}, + {AcceleratorType: "L4", AcceleratorID: "gr6.4xlarge", GPUCount: 1, CapacityType: od, Available: true}, // A different count must NOT bleed into the x1 result. - {AcceleratorType: "L4", AcceleratorID: "g6.48xlarge", GPUCount: 8, CapacityType: nebulav1alpha1.CapacityOnDemand}, + {AcceleratorType: "L4", AcceleratorID: "g6.48xlarge", GPUCount: 8, CapacityType: od, Available: true}, }}, } @@ -244,6 +245,32 @@ func TestBaseMapAccelerator_AlternatesPrimaryFirst(t *testing.T) { } } +// An offering whose availability column is false must not contribute an id, and a +// (type, count) whose every matching row is unavailable must map to ok=false so +// placement/Provision skip it — the CSV's availability flag is the scheduling gate. +func TestBaseMapAccelerator_SkipsUnavailable(t *testing.T) { + base := Base{ + ProviderName: "aws", + Catalog: fakeLookup{rows: []provider.Offering{ + // A100-80GB has one available alternate and one unavailable primary: only + // the available id survives. + {AcceleratorType: "A100-80GB", AcceleratorID: "p4d.24xlarge", GPUCount: 8, Available: false}, + {AcceleratorType: "A100-80GB", AcceleratorID: "p4de.24xlarge", GPUCount: 8, Available: true}, + // H100 has only unavailable rows: it maps to nothing. + {AcceleratorType: "H100", AcceleratorID: "p5.48xlarge", GPUCount: 8, Available: false}, + }}, + } + + got, ok := base.MapAccelerator("A100-80GB", 8) + if !ok || !reflect.DeepEqual(got, []string{"p4de.24xlarge"}) { + t.Fatalf("MapAccelerator(A100-80GB, 8) = (%v, %v), want p4de.24xlarge — unavailable primary dropped", got, ok) + } + + if got, ok := base.MapAccelerator("H100", 8); ok || len(got) > 0 { + t.Fatalf("MapAccelerator(H100, 8) = (%v, %v), want (nil, false) — every row unavailable", got, ok) + } +} + func TestOfferings_CopyIsolation(t *testing.T) { c, err := Load() if err != nil { diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index a878ef4..22fda2d 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -20,6 +20,7 @@ import ( "context" "encoding/json" "io" + "math/rand/v2" "sync" "time" @@ -49,11 +50,24 @@ import ( // reclaims sooner; an OnDemand-only one can poll lazily). const defaultPollInterval = 15 * time.Second -// defaultBlocklistTTL bounds how long a failed placement is excluded when the Pod +// defaultBlocklistTTL is the BASE exclusion for a failed placement when the Pod // 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 = 3 * time.Minute +// behaviour is identical whether or not the pool policy reached the handler. The +// effective TTL is this base plus a random jitter of up to blocklistJitter (see +// recordBlock), so it is deliberately short: the jitter, not a long fixed floor, +// is what spreads retries. +const defaultBlocklistTTL = 30 * time.Second + +// blocklistJitter is the upper bound on the random delay ADDED to the base TTL for +// every recorded block. Without it, all Pods that failed for one (accelerator, +// tier, region) scope re-block in lockstep and their exclusions lapse at the same +// instant, so they stampede the same just-freed candidate together — and, because +// records coalesce on the LATEST expiry (see Blocklist.Record), the shared entry +// would otherwise carry an identical deadline for all of them. A per-record jitter +// in [0, blocklistJitter) decorrelates those wake-ups so freed capacity is sampled +// across a spread of moments instead of one thundering retry. +const blocklistJitter = 1 * 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 @@ -109,6 +123,11 @@ type Handler struct { // nowFn and pollEvery are seams for tests. nowFn func() metav1.Time pollEvery time.Duration + + // jitterFn returns the random delay added to a block's base TTL (see + // recordBlock). A seam so tests can pin it (e.g. to 0) and assert an exact TTL; + // production wires it to a uniform draw in [0, blocklistJitter). + jitterFn func() time.Duration } // trackedPod is the virtual node's local record of one pod it provisioned for. @@ -144,6 +163,9 @@ func NewHandler(prov provider.Provider, client kubernetes.Interface, blocklist B tracked: make(map[string]*trackedPod), nowFn: metav1.Now, pollEvery: poll, + // rand/v2's top-level source is auto-seeded and safe for concurrent use, so + // every handler draws an independent jitter without shared seeding. + jitterFn: func() time.Duration { return time.Duration(rand.Int64N(int64(blocklistJitter))) }, } } @@ -621,7 +643,11 @@ func (h *Handler) recordBlock(ctx context.Context, pod *corev1.Pod, err error) { return } - ttl := blocklistTTL(pod) + // Effective TTL = base (pool policy or default) + a per-record jitter, so Pods + // that failed for the SAME scope do not all re-probe the just-freed candidate at + // the same instant. Coalescing keeps the latest expiry, so distinct jittered + // records spread the shared entry's deadline instead of pinning it. + ttl := blocklistTTL(pod) + h.jitterFn() // Use the request-scoped logger from ctx (it carries the virtualNode/provider // values controller-runtime attached upstream); a logf.FromContext on a fresh // context.Background() would fall back to the global delegate and could be diff --git a/pkg/vnode/handler_test.go b/pkg/vnode/handler_test.go index 7242196..e1ff35c 100644 --- a/pkg/vnode/handler_test.go +++ b/pkg/vnode/handler_test.go @@ -167,6 +167,7 @@ func TestCreatePod_ProvisionFailureRecordsBlock(t *testing.T) { fp := &fakeProvider{provisionErr: errors.New("no capacity"), classifyScope: scope} bl := &recordingBlocklist{} h := NewHandler(fp, nil, bl) + h.jitterFn = func() time.Duration { return 0 } // pin jitter so the base TTL is asserted exactly pod := testPod("default", "p1") // A non-default TTL so this asserts the annotation is honored, not that it @@ -217,6 +218,7 @@ func TestCreatePod_MissingTTLAnnotationUsesDefault(t *testing.T) { fp := &fakeProvider{provisionErr: errors.New("no capacity"), classifyScope: provider.BlockScope{DenyAll: true}} bl := &recordingBlocklist{} h := NewHandler(fp, nil, bl) + h.jitterFn = func() time.Duration { return 0 } // pin jitter so the base default is asserted exactly // No BlocklistTTLAnnotation on the Pod => the handler's built-in default. if err := h.CreatePod(context.Background(), testPod("default", "p1")); err == nil { @@ -227,6 +229,36 @@ func TestCreatePod_MissingTTLAnnotationUsesDefault(t *testing.T) { } } +// The recorded TTL is the base (annotation or default) PLUS the handler's jitter, +// so Pods failing for one scope do not re-probe the freed candidate in lockstep. +func TestCreatePod_BlocklistTTLAddsJitter(t *testing.T) { + fp := &fakeProvider{provisionErr: errors.New("no capacity"), classifyScope: provider.BlockScope{DenyAll: true}} + bl := &recordingBlocklist{} + h := NewHandler(fp, nil, bl) + h.jitterFn = func() time.Duration { return 20 * time.Second } // deterministic jitter + + pod := testPod("default", "p1") + pod.Annotations = map[string]string{nebulav1alpha1.BlocklistTTLAnnotation: "30s"} + if err := h.CreatePod(context.Background(), pod); err == nil { + t.Fatal("expected CreatePod to return the provision error") + } + if want := 30*time.Second + 20*time.Second; bl.ttl != want { + t.Fatalf("recorded ttl = %v, want base+jitter %v", bl.ttl, want) + } +} + +// The production jitter draw stays within [0, blocklistJitter) — never negative +// (which would shorten the block below its base) and never at/over the bound. +func TestProductionJitterInRange(t *testing.T) { + h := NewHandler(&fakeProvider{}, nil, nil) + for i := 0; i < 1000; i++ { + j := h.jitterFn() + if j < 0 || j >= blocklistJitter { + t.Fatalf("jitter %v out of [0, %v)", j, blocklistJitter) + } + } +} + func TestDeletePod_TerminatesAndUntracks(t *testing.T) { fp := &fakeProvider{provisionID: "inst-1"} h := NewHandler(fp, nil, nil)