fix: use live spot prices

docs/deploy.md

@@ -172,6 +172,26 @@ identity, the `system:nodes` organization, and the current manager Pod IP as the
 
 ---
 
+## AWS IAM permissions
+
+The AWS adapter uses the SDK's default credential chain (IRSA or an instance role in production).
+That identity needs these EC2 actions:
+
+| Action | Used for |
+|---|---|
+| `ec2:CreateFleet` | Launching an instance (an *instant* fleet across the region's AZs) |
+| `ec2:RunInstances`, `ec2:CreateTags` | Required by `CreateFleet` to launch and tag the instance |
+| `ec2:CreateLaunchTemplate`, `ec2:DeleteLaunchTemplate`, `ec2:DescribeLaunchTemplates` | The ephemeral launch template each launch creates, and sweeping leaked ones |
+| `ec2:TerminateInstances` | Deleting an instance |
+| `ec2:DescribeInstances` | Finding and listing Nebula's instances |
+| `ec2:DescribeInstanceStatus` | The 2/2 status checks that gate `Running` (a separate grant from `DescribeInstances`) |
+| `ec2:DescribeInstanceTypeOfferings` | Which instance types a region and each AZ offer |
+| `ec2:DescribeImages`, `ec2:DescribeSubnets` | The region's GPU AMI and default-VPC subnets |
+| `ec2:DescribeSpotPriceHistory` | The live Spot price a Spot claim is priced at |
+
+Without `ec2:DescribeSpotPriceHistory`, Spot claims are priced at the catalog's seed rate (a
+us-east-1 snapshot) instead of the region's current price. Placement and launches still work.
+
 ## Modal Environments
 
 Optional, and off unless you set it. A Modal **Environment** is a named partition of

internal/controller/nodeclaim_controller.go

@@ -409,6 +409,7 @@ func (r *NodeClaimReconciler) recordPrice(ctx context.Context, nc *nebulav1alpha
 		AcceleratorType: accelerator,
 		Count:           count,
 		CapacityType:    nc.Spec.CapacityType,
+		Region:          nc.Spec.Region,
 		CPUCores:        cpuCores,
 		MemoryMiB:       memoryMiB,
 		DiskGiB:         util.PodEphemeralStorageGiB(pod),

internal/controller/nodeclaim_price_test.go

@@ -75,6 +75,7 @@ func TestRecordPrice_WritesRateAndRequest(t *testing.T) {
 	pod := gpuPod("H100", 2, "4", "8Gi")
 	claim := newClaim("c1", "p1", "default", "uid-1", "fake")
 	claim.Spec.CapacityType = nebulav1alpha1.CapacityOnDemand
+	claim.Spec.Region = "us-east-1"
 	pp := &pricedProvider{fakeProvider: &fakeProvider{name: "fake"}, rate: 7.9}
 	r, c := newPricedReconciler(t, []client.Object{pod, claim}, pp)
 
@@ -90,6 +91,8 @@ func TestRecordPrice_WritesRateAndRequest(t *testing.T) {
 	want := provider.PriceRequest{
 		AcceleratorType: "H100", Count: 2,
 		CapacityType: nebulav1alpha1.CapacityOnDemand,
+		// The claim's region, for a provider whose price varies by region (AWS Spot).
+		Region: "us-east-1",
 		// The Pod's own reservation, which only a provider metering CPU/memory apart from
 		// the accelerator charges for — but the request carries it either way.
 		CPUCores: 4, MemoryMiB: 8192,

pkg/provider/aws/aws.go

@@ -134,6 +134,11 @@ type Client interface {
 	// region offers is AWS-authoritative and changes over time, so it is queried
 	// live rather than hand-maintained in the catalog.
 	AvailableInstanceTypes(ctx context.Context) (map[string]bool, error)
+	// SpotPrice returns the current Spot price (USD/hour) of one instance type in the
+	// client's region: the LOWEST across its AZs, since the launch fleet's default Spot
+	// allocation strategy is lowest-price. Spot prices move hourly and differ per region
+	// and AZ, so they are read live rather than from the catalog's seed rows.
+	SpotPrice(ctx context.Context, instanceType string) (float64, error)
 }
 
 // InstanceSpec is the resolved, EC2-shaped request the Client turns into a
@@ -236,8 +241,34 @@ type Provider struct {
 
 	mu      sync.Mutex
 	clients map[string]Client // region -> Client, populated lazily by clientFor
+
+	// spotMu guards spotPrices, the live Spot price cache PricePerHour reads (see spotRate).
+	spotMu     sync.Mutex
+	spotPrices map[spotPriceKey]spotPriceEntry
+}
+
+// spotPriceKey is one cached Spot price: an instance type in a region.
+type spotPriceKey struct{ region, instanceType string }
+
+// spotPriceEntry is a cached Spot price, or the failure to read one (so a missing IAM grant
+// or throttle is not retried on every reconcile).
+type spotPriceEntry struct {
+	price float64
+	err   error
+	at    time.Time
 }
 
+const (
+	// spotPriceTTL is how long a live Spot price is reused. Spot prices move gradually, at
+	// most every hour or so, so this bounds staleness without a call per reconcile.
+	spotPriceTTL = time.Hour
+	// spotPriceErrTTL is how long a failed lookup is remembered before retrying.
+	spotPriceErrTTL = 5 * time.Minute
+	// spotPriceTimeout caps one DescribeSpotPriceHistory lookup. PricePerHour has no context
+	// of its own, and a slow lookup only delays a cost figure.
+	spotPriceTimeout = 10 * time.Second
+)
+
 // New returns an EC2 Provider backed by a client factory and price catalog.
 // regionSource supplies the NodePool declarations the List/Offerings fan-out
 // sweeps (nil is tolerated — the sweep then uses only the regions already
@@ -254,6 +285,7 @@ func New(newClient ClientFactory, cat catalog.Lookup, regionSource RegionSource)
 		newClient:    newClient,
 		regionSource: regionSource,
 		clients:      make(map[string]Client),
+		spotPrices:   make(map[spotPriceKey]spotPriceEntry),
 	}
 }
 
@@ -454,7 +486,7 @@ func (p *Provider) Offerings(ctx context.Context) ([]provider.Offering, error) {
 			log.Error(err, "probe instance-type availability failed; skipping region", "region", region)
 			continue
 		}
-		for _, o := range rows {
+		for _, o := range regionRows(rows, region) {
 			o.Region = region
 			o.Available = o.Available && avail[o.AcceleratorID]
 			out = append(out, o)
@@ -466,6 +498,36 @@ func (p *Provider) Offerings(ctx context.Context) ([]provider.Offering, error) {
 	return out, nil
 }
 
+// regionRows returns the catalog rows that price region: its own rows, plus a blank-region
+// (default) row for each offering the region has no row of its own for. Rows for other
+// regions are dropped.
+func regionRows(rows []provider.Offering, region string) []provider.Offering {
+	type offeringKey struct {
+		id    string
+		count int32
+		tier  nebulav1alpha1.CapacityType
+	}
+	key := func(o provider.Offering) offeringKey {
+		return offeringKey{id: o.AcceleratorID, count: o.GPUCount, tier: o.CapacityType}
+	}
+	own := map[offeringKey]bool{}
+	for _, o := range rows {
+		if o.Region == region {
+			own[key(o)] = true
+		}
+	}
+	out := make([]provider.Offering, 0, len(rows))
+	for _, o := range rows {
+		switch {
+		case o.Region == region:
+			out = append(out, o)
+		case o.Region == "" && !own[key(o)]:
+			out = append(out, o)
+		}
+	}
+	return out
+}
+
 // Provision implements provider.Provider. The Pod is the source of truth for the
 // workload; req carries the claim identity, capacity tier, and region.
 //
@@ -760,9 +822,62 @@ func (p *Provider) PricePerHour(req provider.PriceRequest) (float64, error) {
 	if err != nil {
 		return 0, err
 	}
+	if req.CapacityType == nebulav1alpha1.CapacitySpot && req.Region != "" {
+		rate = p.spotRate(req, rate)
+	}
 	return rate + data.AWSRootVolumeCostPerHour(awsAMIRootGiB+req.DiskGiB), nil
 }
 
+// spotRate is the live Spot price of the request's primary instance type in its region, or
+// seed (the catalog's Spot row) when the live price cannot be read. The catalog's Spot rows
+// are a single us-east-1 snapshot; the live price is what the region charges now.
+func (p *Provider) spotRate(req provider.PriceRequest, seed float64) float64 {
+	ids, ok := p.MapAccelerator(req.AcceleratorType, req.Count)
+	if !ok || len(ids) == 0 {
+		return seed
+	}
+	key := spotPriceKey{region: req.Region, instanceType: ids[0]}
+	log := logf.Log.WithName("aws-spot-price").WithValues("region", key.region, "instanceType", key.instanceType)
+
+	p.spotMu.Lock()
+	e, cached := p.spotPrices[key]
+	p.spotMu.Unlock()
+	if !cached || e.expired(time.Now()) {
+		e = p.fetchSpotPrice(key)
+		p.spotMu.Lock()
+		p.spotPrices[key] = e
+		p.spotMu.Unlock()
+	}
+	if e.err != nil {
+		log.V(1).Info("live spot price unavailable; using the catalog seed", "seed", seed, "error", e.err.Error())
+		return seed
+	}
+	return e.price
+}
+
+// fetchSpotPrice reads one live Spot price through the region's client.
+func (p *Provider) fetchSpotPrice(key spotPriceKey) spotPriceEntry {
+	ctx, cancel := context.WithTimeout(context.Background(), spotPriceTimeout)
+	defer cancel()
+	now := time.Now()
+	client, err := p.clientFor(ctx, key.region)
+	if err != nil {
+		return spotPriceEntry{err: err, at: now}
+	}
+	price, err := client.SpotPrice(ctx, key.instanceType)
+	return spotPriceEntry{price: price, err: err, at: now}
+}
+
+// expired reports whether the entry must be re-read: a price after spotPriceTTL, a failure
+// after the shorter spotPriceErrTTL.
+func (e spotPriceEntry) expired(now time.Time) bool {
+	ttl := spotPriceTTL
+	if e.err != nil {
+		ttl = spotPriceErrTTL
+	}
+	return now.Sub(e.at) > ttl
+}
+
 // instanceSpecFromPod reads the workload off the Pod (source of truth) and the
 // accelerator type (from the AcceleratorTypeLabel), maps it to an EC2 instance
 // type via the catalog, and stamps the claim tag, capacity tier, and region.

pkg/provider/aws/aws_test.go

@@ -55,6 +55,11 @@
 	// Offerings call on a bare fake reports every row unavailable but still present.
 	available map[string]bool
 	availErr  error
+	// spotPrices is what SpotPrice returns per instance type; spotErr injects a failure.
+	// spotCalls counts lookups so a test can assert the cache.
+	spotPrices map[string]float64
+	spotErr    error
+	spotCalls  int
 }
 
 func (f *fakeClient) RunInstance(_ context.Context, spec InstanceSpec) (string, error) {
@@ -107,6 +112,18 @@
 	return nil, nil
 }
 
+func (f *fakeClient) SpotPrice(_ context.Context, instanceType string) (float64, error) {
+	f.spotCalls++
+	if f.spotErr != nil {
+		return 0, f.spotErr
+	}
+	price, ok := f.spotPrices[instanceType]
+	if !ok {
+		return 0, errNoSpotPrice
+	}
+	return price, nil
+}
+
 func (f *fakeClient) AvailableInstanceTypes(_ context.Context) (map[string]bool, error) {
 	if f.availErr != nil {
 		return nil, f.availErr
@@ -123,9 +140,10 @@
 
 const testRegion = "us-east-1"
 
-// offering is a compact constructor for a region-aware test catalog row (the
-// full struct literal is too wide to read inline). count is the gpu_count, the
-// AWS lookup key that (with the accelerator type) selects the instance type.
+// offering is a compact constructor for a test catalog row (the full struct literal
+// is too wide to read inline). count is the gpu_count, the AWS lookup key that (with
+// the accelerator type) selects the instance type. The region is blank, as in the real
+// catalog's default rows; Offerings stamps it per region.
 func offering(accel, id string, count int32, tier nebulav1alpha1.CapacityType, price float64) provider.Offering {
 	return provider.Offering{
 		AcceleratorType: accel,
@@ -134,7 +152,6 @@
 		CapacityType:    tier,
 		PricePerHour:    price,
 		Available:       true,
-		Region:          testRegion,
 	}
 }
 
@@ -1126,3 +1143,94 @@
 		t.Fatalf("RunInstance called %d times for a refused disk", f.runCnt)
 	}
 }
+
+func TestPricePerHour_SpotUsesLivePrice(t *testing.T) {
+	f := &fakeClient{spotPrices: map[string]float64{"p5.48xlarge": 30.692}}
+	p := newTestProvider(f)
+	req := provider.PriceRequest{AcceleratorType: "H100", Count: 8, CapacityType: nebulav1alpha1.CapacitySpot, Region: testRegion}
+	want := 30.692 + data.AWSRootVolumeCostPerHour(awsAMIRootGiB)
+
+	got, err := p.PricePerHour(req)
+	if err != nil || got != want {
+		t.Fatalf("PricePerHour(spot) = %v, %v; want the live price + root volume %v, not the 34.41 seed", got, err, want)
+	}
+	if _, _ = p.PricePerHour(req); f.spotCalls != 1 {
+		t.Fatalf("spot lookups = %d after two calls, want 1 (cached)", f.spotCalls)
+	}
+
+	// Past the TTL the price is read again.
+	key := spotPriceKey{region: testRegion, instanceType: "p5.48xlarge"}
+	e := p.spotPrices[key]
+	e.at = time.Now().Add(-spotPriceTTL - time.Minute)
+	p.spotPrices[key] = e
+	f.spotPrices["p5.48xlarge"] = 31.0
+	if got, _ := p.PricePerHour(req); f.spotCalls != 2 || got != 31.0+data.AWSRootVolumeCostPerHour(awsAMIRootGiB) {
+		t.Fatalf("after TTL: price %v, lookups %d; want the refreshed 31.0 from a second lookup", got, f.spotCalls)
+	}
+}
+
+func TestPricePerHour_SpotFallsBackToSeed(t *testing.T) {
+	seed := 34.41 + data.AWSRootVolumeCostPerHour(awsAMIRootGiB)
+	req := provider.PriceRequest{AcceleratorType: "H100", Count: 8, CapacityType: nebulav1alpha1.CapacitySpot, Region: testRegion}
+
+	f := &fakeClient{spotErr: errors.New("UnauthorizedOperation")}
+	p := newTestProvider(f)
+	if got, err := p.PricePerHour(req); err != nil || got != seed {
+		t.Fatalf("PricePerHour(lookup fails) = %v, %v; want the seed %v, never an error", got, err, seed)
+	}
+	if _, _ = p.PricePerHour(req); f.spotCalls != 1 {
+		t.Fatalf("spot lookups = %d after two failing calls, want 1 (the failure is cached)", f.spotCalls)
+	}
+
+	// No live price for the type (e.g. not sold on Spot in the region) is the same fallback.
+	p = newTestProvider(&fakeClient{spotPrices: map[string]float64{}})
+	if got, err := p.PricePerHour(req); err != nil || got != seed {
+		t.Fatalf("PricePerHour(no live price) = %v, %v; want the seed %v", got, err, seed)
+	}
+}
+
+func TestPricePerHour_LivePriceOnlyForSpotWithRegion(t *testing.T) {
+	f := &fakeClient{spotPrices: map[string]float64{"p5.48xlarge": 30.692}}
+	p := newTestProvider(f)
+	root := data.AWSRootVolumeCostPerHour(awsAMIRootGiB)
+
+	// Spot with no region: nothing to look up.
+	got, err := p.PricePerHour(provider.PriceRequest{AcceleratorType: "H100", Count: 8, CapacityType: nebulav1alpha1.CapacitySpot})
+	if err != nil || got != 34.41+root {
+		t.Fatalf("PricePerHour(spot, no region) = %v, %v; want the seed %v", got, err, 34.41+root)
+	}
+	// On-demand: the catalog's list price, untouched.
+	got, err = p.PricePerHour(provider.PriceRequest{AcceleratorType: "H100", Count: 8, CapacityType: nebulav1alpha1.CapacityOnDemand, Region: testRegion})
+	if err != nil || got != 98.32+root {
+		t.Fatalf("PricePerHour(on-demand) = %v, %v; want the catalog price %v", got, err, 98.32+root)
+	}
+	if f.spotCalls != 0 {
+		t.Fatalf("spot lookups = %d, want 0 for these requests", f.spotCalls)
+	}
+}
+
+func TestRegionRows(t *testing.T) {
+	od, spot := nebulav1alpha1.CapacityOnDemand, nebulav1alpha1.CapacitySpot
+	withRegion := func(o provider.Offering, region string) provider.Offering { o.Region = region; return o }
+	rows := []provider.Offering{
+		offering("H100", "p5.48xlarge", 8, od, 55.040),
+		offering("H100", "p5.48xlarge", 8, spot, 29.168),
+		withRegion(offering("H100", "p5.48xlarge", 8, od, 58.893), "eu-north-1"),
+		withRegion(offering("H100", "p5.48xlarge", 8, od, 71.552), "eu-west-2"),
+	}
+	got := regionRows(rows, "eu-north-1")
+	if len(got) != 2 {
+		t.Fatalf("regionRows(eu-north-1) = %d rows, want 2 (its own on-demand + the default spot): %+v", len(got), got)
+	}
+	for _, o := range got {
+		switch {
+		case o.CapacityType == od && (o.Region != "eu-north-1" || o.PricePerHour != 58.893):
+			t.Errorf("on-demand row = %+v, want eu-north-1's own 58.893, not the default or eu-west-2's", o)
+		case o.CapacityType == spot && o.Region != "":
+			t.Errorf("spot row = %+v, want the default (eu-north-1 has none of its own)", o)
+		}
+	}
+	if got := regionRows(rows, "us-east-2"); len(got) != 2 || got[0].Region != "" || got[1].Region != "" {
+		t.Fatalf("regionRows(us-east-2) = %+v, want just the two defaults", got)
+	}
+}