diff --git a/README.md b/README.md index f0d236f..e88781b 100644 --- a/README.md +++ b/README.md @@ -2,12 +2,12 @@ # Nebula -**The control plane for GPUaaS** +**The Control Plane for GPUaaS** +[![Discord](https://img.shields.io/badge/Discord-Join%20us-5865F2?logo=discord&logoColor=white)](https://discord.gg/7WTUuFqyS6) +![Go Version](https://img.shields.io/badge/go-1.24-00ADD8?logo=go&logoColor=white) [![Go Reference](https://pkg.go.dev/badge/github.com/InftyAI/Nebula.svg)](https://pkg.go.dev/github.com/InftyAI/Nebula) [![License](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](LICENSE) -![Go Version](https://img.shields.io/badge/go-1.24-00ADD8?logo=go&logoColor=white) -[![Discord](https://img.shields.io/badge/Discord-Join%20us-5865F2?logo=discord&logoColor=white)](https://discord.gg/7WTUuFqyS6) Run GPU workloads on any NeoCloud or hyperscaler or your own infrastructure through one Kubernetes API. diff --git a/api/v1alpha1/groupversion_info.go b/api/v1alpha1/groupversion_info.go index 0f93c10..97f8c1a 100644 --- a/api/v1alpha1/groupversion_info.go +++ b/api/v1alpha1/groupversion_info.go @@ -128,15 +128,19 @@ const ( // default TTL. BlocklistTTLAnnotation = "nebula.inftyai.com/blocklist-ttl" - // EndpointAnnotation carries the reachable address of the external instance - // once it is running (a public DNS name or IP, in the provider's own form). + // EndpointAnnotation carries the reachable address of the external instance (a + // public DNS name, an IP, or a URL, in the provider's own form). // It is the ONLY way to reach the workload, so it must be visible on the Pod: // PodIP cannot hold it because the API server validates PodIP as a literal IP // and rejects a DNS name (the common AWS case), so the endpoint rides an - // annotation instead. Written by the virtual kubelet when it first observes the - // instance running; absent until then. Unlike the provisioning-input - // annotations above (which the placement controller stamps and VK reads), this - // flows the other way — VK writes it for operators/tooling to read. + // annotation instead. Written by the virtual kubelet as soon as it knows the + // address, which is provider-dependent and NOT tied to the phase: a provider + // that mints a connect URL at create time (Modal) publishes it from CreatePod, + // before the instance is Running; one whose address only exists after boot (AWS) + // publishes it from the poll loop. Absent until then, and never cleared once + // written. Unlike the provisioning-input annotations above (which the placement + // controller stamps and VK reads), this flows the other way — VK writes it for + // operators/tooling to read. EndpointAnnotation = "nebula.inftyai.com/endpoint" // TerminateInstanceFinalizer is held by every NodeClaim to guarantee teardown. diff --git a/docs/status.md b/docs/status.md index 9599838..f624fb3 100644 --- a/docs/status.md +++ b/docs/status.md @@ -48,11 +48,19 @@ teardown. | `Failed` | `Terminated` | `applyState` ← `InstanceTerminated` | gone | `Terminated` | | `Succeeded` | `Terminated` | `DeletePod` | gone | `Terminated` | -`Running` also sets `Ready=True` and the endpoint annotation. A Pod carrying a -`DeletionTimestamp` maps to `Terminating` from any non-terminal phase. When the -served Pod is ABSENT: after `Bound`/`Terminating`, the claim deletes itself and -the terminate finalizer runs; before `Bound`, it waits `placementGracePeriod` -first. +`Running` also sets `Ready=True`. A Pod carrying a `DeletionTimestamp` maps to +`Terminating` from any non-terminal phase. When the served Pod is ABSENT: after +`Bound`/`Terminating`, the claim deletes itself and the terminate finalizer runs; +before `Bound`, it waits `placementGracePeriod` first. + +The endpoint annotation is written on whichever path first knows the address, which +differs by provider and is independent of phase. A provider that mints a connect URL +at create time (Modal) publishes it from `CreatePod`, alongside the Secret holding +the matching bearer token — so the endpoint is on the Pod before it is `Running`. A +provider whose address only exists once the instance boots (AWS's public DNS name) +reports it through `List()`, so the poll loop publishes it. Nothing ever clears the +annotation: once written it stays for the Pod's life, which is what lets the two +paths coexist and what makes the address survive a manager restart. Note the two writers. `CreatePod` writes the rows around the `Provision` call; every other non-terminal row comes from `applyState`, driven by the poll loop, and diff --git a/internal/controller/nodeclaim_controller_test.go b/internal/controller/nodeclaim_controller_test.go index 447349c..6872fac 100644 --- a/internal/controller/nodeclaim_controller_test.go +++ b/internal/controller/nodeclaim_controller_test.go @@ -50,8 +50,8 @@ type fakeProvider struct { func (f *fakeProvider) Name() string { return f.name } func (f *fakeProvider) Capabilities() provider.Capabilities { return provider.Capabilities{} } -func (f *fakeProvider) Provision(context.Context, *corev1.Pod, provider.ProvisionRequest) (string, bool, error) { - return "", false, nil +func (f *fakeProvider) Provision(context.Context, *corev1.Pod, provider.ProvisionRequest) (provider.ProvisionResult, error) { + return provider.ProvisionResult{}, nil } func (f *fakeProvider) Terminate(_ context.Context, id string) error { f.terminated = append(f.terminated, id) diff --git a/pkg/provider/aws/aws.go b/pkg/provider/aws/aws.go index 8268d1f..b8b430c 100644 --- a/pkg/provider/aws/aws.go +++ b/pkg/provider/aws/aws.go @@ -388,20 +388,26 @@ func (p *Provider) Offerings(ctx context.Context) ([]provider.Offering, error) { // is booting on real hardware — or the reason it could not launch, which becomes an // error driving AZ/region/tier failover. There is no queued state for an EC2 // instance to sit in, so the reserved return is unconditionally true on success. +// +// No connect credential is returned. EC2 has nothing to mint: an instance is reached +// at its public DNS name or IP, which EC2 does not know until the instance boots, so +// the address is reported the level-triggered way — observed by List/Get into +// Instance.Endpoint — and access is authenticated by the key pair and security group, +// not a bearer token. func (p *Provider) Provision( ctx context.Context, pod *corev1.Pod, req provider.ProvisionRequest, -) (string, bool, error) { +) (provider.ProvisionResult, error) { if pod == nil { - return "", false, errors.New("aws: nil pod") + return provider.ProvisionResult{}, errors.New("aws: nil pod") } if req.ClaimName == "" { - return "", false, errors.New("aws: empty ClaimName in ProvisionRequest") + return provider.ProvisionResult{}, errors.New("aws: empty ClaimName in ProvisionRequest") } region := req.Region client, err := p.clientFor(ctx, region) if err != nil { - return "", false, err + return provider.ProvisionResult{}, err } // Idempotency: if an instance already carries this claim tag IN THIS REGION, @@ -410,23 +416,23 @@ func (p *Provider) Provision( // target region's client is sufficient. It is reserved for the same reason a // fresh launch is: it only exists because some earlier instant fleet succeeded. if existing, err := findByClaim(ctx, client, req.ClaimName); err != nil { - return "", false, err + return provider.ProvisionResult{}, err } else if existing != nil { - return existing.ID, true, nil + return provider.ProvisionResult{InstanceID: existing.ID, Reserved: true}, nil } spec, err := p.instanceSpecFromPod(pod, req) if err != nil { - return "", false, err + return provider.ProvisionResult{}, err } // The Provision deadline is enforced generically by the vnode handler (from // Capabilities.ProvisionTimeout), so RunInstance simply honors ctx as it fails // over across zones — no adapter-local WithTimeout here. id, err := client.RunInstance(ctx, spec) if err != nil { - return "", false, err + return provider.ProvisionResult{}, err } - return id, true, nil + return provider.ProvisionResult{InstanceID: id, Reserved: true}, nil } // Terminate implements provider.Provider. Idempotent by the Client contract. The diff --git a/pkg/provider/aws/aws_test.go b/pkg/provider/aws/aws_test.go index e4a9c8e..639de64 100644 --- a/pkg/provider/aws/aws_test.go +++ b/pkg/provider/aws/aws_test.go @@ -164,7 +164,7 @@ func TestProvision_MapsAcceleratorToInstanceType(t *testing.T) { f := &fakeClient{runID: "i-1"} p := newTestProvider(f) - id, reserved, err := p.Provision(context.Background(), gpuPod("H100", 8), provider.ProvisionRequest{ + res, err := p.Provision(context.Background(), gpuPod("H100", 8), provider.ProvisionRequest{ ClaimName: "claim-a", CapacityType: nebulav1alpha1.CapacityOnDemand, Region: "us-west-2", @@ -172,6 +172,14 @@ func TestProvision_MapsAcceleratorToInstanceType(t *testing.T) { if err != nil { t.Fatalf("Provision: %v", err) } + id, reserved := res.InstanceID, res.Reserved + // EC2 mints no bearer credential: an instance is reached at the public DNS name or + // IP that only exists once it boots, so the address comes from the observed + // endpoint, not from this call. + if res.ConnectURL != "" || res.ConnectToken != "" { + t.Fatalf("expected no credential from AWS, got url=%q token set=%t", + res.ConnectURL, res.ConnectToken != "") + } // The returned id is the raw EC2 id (no region prefix); Terminate/Get re-locate // it by sweeping regions. if id != "i-1" { @@ -216,7 +224,7 @@ func TestProvision_LowercaseAcceleratorLabel(t *testing.T) { // A user may write the accelerator-type label in any case; it must resolve to // the canonical catalog row (and thus the right instance type). - if _, _, err := p.Provision(context.Background(), gpuPod("h100", 8), provider.ProvisionRequest{ + if _, err := p.Provision(context.Background(), gpuPod("h100", 8), provider.ProvisionRequest{ ClaimName: "claim-lc", Region: testRegion, }); err != nil { @@ -251,7 +259,7 @@ func TestProvision_CountSelectsInstanceType(t *testing.T) { f := &fakeClient{runID: "i-t4"} p := newTestProvider(f) req := provider.ProvisionRequest{ClaimName: "claim-t4", Region: testRegion} - if _, _, err := p.Provision(context.Background(), gpuPod("T4", tc.count), req); err != nil { + if _, err := p.Provision(context.Background(), gpuPod("T4", tc.count), req); err != nil { t.Fatalf("Provision(T4 x%d): %v", tc.count, err) } if got := primaryType(f.lastSpec); got != tc.wantType { @@ -266,7 +274,7 @@ func TestProvision_UnsupportedCountIsError(t *testing.T) { // T4 x2 has no instance type (there is no 2-GPU T4 shape): must error rather // than silently picking the x1 or x8 row. req := provider.ProvisionRequest{ClaimName: "claim-t4x2", Region: testRegion} - if _, _, err := p.Provision(context.Background(), gpuPod("T4", 2), req); err == nil { + if _, err := p.Provision(context.Background(), gpuPod("T4", 2), req); err == nil { t.Fatal("expected an error for an unsupported (accelerator, count) pair") } if f.runCnt != 0 { @@ -278,7 +286,7 @@ func TestProvision_SpotSetsMarketOption(t *testing.T) { f := &fakeClient{runID: "i-spot"} p := newTestProvider(f) - if _, _, err := p.Provision(context.Background(), gpuPod("H100", 8), provider.ProvisionRequest{ + if _, err := p.Provision(context.Background(), gpuPod("H100", 8), provider.ProvisionRequest{ ClaimName: "claim-spot", CapacityType: nebulav1alpha1.CapacitySpot, Region: testRegion, @@ -298,7 +306,7 @@ func TestProvision_EmptyRegionIsError(t *testing.T) { // Provision errors rather than silently guessing. In production every request // carries a region (admission requires each aws pool to list ≥1; placement stamps // it), so this only guards a malformed request. - if _, _, err := p.Provision(context.Background(), gpuPod("H100", 8), provider.ProvisionRequest{ + if _, err := p.Provision(context.Background(), gpuPod("H100", 8), provider.ProvisionRequest{ ClaimName: "claim-def", }); err == nil { t.Fatal("expected an error for a request with no region") @@ -315,7 +323,7 @@ func TestProvision_NoAcceleratorIsError(t *testing.T) { // EC2 GPU provisioning is by instance type; a Pod with no accelerator has no // instance type to launch, so it must error rather than silently guessing. req := provider.ProvisionRequest{ClaimName: "claim-cpu", Region: testRegion} - if _, _, err := p.Provision(context.Background(), gpuPod("", 0), req); err == nil { + if _, err := p.Provision(context.Background(), gpuPod("", 0), req); err == nil { t.Fatal("expected an error for a Pod requesting no accelerator") } if f.runCnt != 0 { @@ -334,11 +342,12 @@ func TestProvision_Idempotent(t *testing.T) { } p := newTestProvider(f) - id, _, err := p.Provision(context.Background(), gpuPod("H100", 8), + res, err := p.Provision(context.Background(), gpuPod("H100", 8), provider.ProvisionRequest{ClaimName: "claim-a", Region: testRegion}) if err != nil { t.Fatalf("Provision: %v", err) } + id := res.InstanceID // Raw EC2 id, since idempotent reuse returns the same clean id a fresh launch would. if id != "i-existing" { t.Fatalf("id = %q, want i-existing (idempotent reuse)", id) @@ -352,7 +361,7 @@ func TestProvision_UnsupportedAccelerator(t *testing.T) { f := &fakeClient{} p := newTestProvider(f) req := provider.ProvisionRequest{ClaimName: "claim-x", Region: testRegion} - if _, _, err := p.Provision(context.Background(), gpuPod("TPU-v4", 1), req); err == nil { + if _, err := p.Provision(context.Background(), gpuPod("TPU-v4", 1), req); err == nil { t.Fatal("expected error for unsupported accelerator") } } diff --git a/pkg/provider/fake/fake.go b/pkg/provider/fake/fake.go index a4e7f0d..1be4282 100644 --- a/pkg/provider/fake/fake.go +++ b/pkg/provider/fake/fake.go @@ -88,12 +88,17 @@ func (p *Provider) Capabilities() provider.Capabilities { // It reports the instance RESERVED, which is honest for an in-memory backend: the // instance is Running the moment it is recorded, so there is no queueing to model // (and none of the capacity the reserved flag exists to describe). -func (p *Provider) Provision(_ context.Context, pod *corev1.Pod, req provider.ProvisionRequest) (string, bool, error) { +// +// No connect credential is minted: there is nothing to authenticate against. The +// endpoint is reported the level-triggered way, through the recorded instance. +func (p *Provider) Provision( + _ context.Context, pod *corev1.Pod, req provider.ProvisionRequest, +) (provider.ProvisionResult, error) { if pod == nil { - return "", false, fmt.Errorf("fake: nil pod") + return provider.ProvisionResult{}, fmt.Errorf("fake: nil pod") } if req.ClaimName == "" { - return "", false, fmt.Errorf("fake: empty ClaimName in ProvisionRequest") + return provider.ProvisionResult{}, fmt.Errorf("fake: empty ClaimName in ProvisionRequest") } p.mu.Lock() @@ -101,7 +106,8 @@ func (p *Provider) Provision(_ context.Context, pod *corev1.Pod, req provider.Pr for _, inst := range p.instances { if inst.ClaimName == req.ClaimName { - return inst.ID, true, nil // idempotent reuse + // Idempotent reuse. + return provider.ProvisionResult{InstanceID: inst.ID, Reserved: true}, nil } } @@ -114,7 +120,7 @@ func (p *Provider) Provision(_ context.Context, pod *corev1.Pod, req provider.Pr Endpoint: fmt.Sprintf("fake://%s", id), CapacityType: req.CapacityType, } - return id, true, nil + return provider.ProvisionResult{InstanceID: id, Reserved: true}, nil } // Terminate forgets the instance. Idempotent: terminating an already-gone (or diff --git a/pkg/provider/fake/fake_test.go b/pkg/provider/fake/fake_test.go index 2c7312c..9d15153 100644 --- a/pkg/provider/fake/fake_test.go +++ b/pkg/provider/fake/fake_test.go @@ -38,20 +38,27 @@ func TestProvisionReportsRunningAndLists(t *testing.T) { p := New() ctx := context.Background() - id, reserved, err := p.Provision(ctx, testPod(), provider.ProvisionRequest{ + res, err := p.Provision(ctx, testPod(), provider.ProvisionRequest{ ClaimName: "claim-a", CapacityType: nebulav1alpha1.CapacityOnDemand, }) if err != nil { t.Fatalf("Provision: %v", err) } + id := res.InstanceID if id == "" { t.Fatal("expected a non-empty instance id") } + // The fake authenticates nothing, so it mints no credential; its address is + // reported the level-triggered way, through the observed instance. + if res.ConnectURL != "" || res.ConnectToken != "" { + t.Fatalf("expected no credential from the fake, got url=%q token set=%t", + res.ConnectURL, res.ConnectToken != "") + } // The fake has no queueing to model — an instance is Running the moment it is // created — so it always reserves. Reporting false would make the fake exercise // the Modal-shaped path and leave Pods at Provisioning forever. - if !reserved { + if !res.Reserved { t.Fatal("reserved = false; the fake allocates synchronously and is Running immediately") } @@ -82,14 +89,15 @@ func TestProvisionIdempotentOnClaim(t *testing.T) { ctx := context.Background() req := provider.ProvisionRequest{ClaimName: "claim-a"} - id1, _, err := p.Provision(ctx, testPod(), req) + res1, err := p.Provision(ctx, testPod(), req) if err != nil { t.Fatalf("Provision #1: %v", err) } - id2, _, err := p.Provision(ctx, testPod(), req) + res2, err := p.Provision(ctx, testPod(), req) if err != nil { t.Fatalf("Provision #2: %v", err) } + id1, id2 := res1.InstanceID, res2.InstanceID if id1 != id2 { t.Fatalf("ids differ (%q vs %q); Provision must be idempotent on ClaimName", id1, id2) } @@ -102,10 +110,11 @@ func TestTerminateIsIdempotent(t *testing.T) { p := New() ctx := context.Background() - id, _, err := p.Provision(ctx, testPod(), provider.ProvisionRequest{ClaimName: "claim-a"}) + res, err := p.Provision(ctx, testPod(), provider.ProvisionRequest{ClaimName: "claim-a"}) if err != nil { t.Fatalf("Provision: %v", err) } + id := res.InstanceID if err := p.Terminate(ctx, id); err != nil { t.Fatalf("Terminate: %v", err) } diff --git a/pkg/provider/modal/client.go b/pkg/provider/modal/client.go index 9f755b7..fdc0d77 100644 --- a/pkg/provider/modal/client.go +++ b/pkg/provider/modal/client.go @@ -45,10 +45,6 @@ type sdkClient struct { mc *modal.Client appName string - // endpointTimeout bounds the best-effort tunnel lookup used to report an - // instance's reachable address; tunnels only exist once the sandbox is up. - endpointTimeout time.Duration - // readyTimeout is the budget one background readiness waiter gets. It is a real // budget, not a hint: WaitUntilReady BLOCKS and returns early only to say // "ready" — never to say "not ready" — so a budget below the call's own setup @@ -105,12 +101,11 @@ func NewSDKClient(ctx context.Context, appName string) (*Provider, error) { return nil, fmt.Errorf("modal: load price catalog: %w", err) } return New(&sdkClient{ - mc: mc, - appName: appName, - endpointTimeout: 5 * time.Second, - readyTimeout: 30 * time.Second, - ready: make(map[string]bool), - waiting: make(map[string]struct{}), + mc: mc, + appName: appName, + readyTimeout: 30 * time.Second, + ready: make(map[string]bool), + waiting: make(map[string]struct{}), }, cat), nil } @@ -120,19 +115,19 @@ func (c *sdkClient) app(ctx context.Context) (*modal.App, error) { } // CreateSandbox implements Client. -func (c *sdkClient) CreateSandbox(ctx context.Context, spec SandboxSpec) (string, error) { +func (c *sdkClient) CreateSandbox(ctx context.Context, spec SandboxSpec) (string, Credential, error) { app, err := c.app(ctx) if err != nil { - return "", fmt.Errorf("modal: resolve app: %w", err) + return "", Credential{}, fmt.Errorf("modal: resolve app: %w", err) } if spec.Image == "" { - return "", fmt.Errorf("modal: empty image in sandbox spec") + return "", Credential{}, fmt.Errorf("modal: empty image in sandbox spec") } image := c.mc.Images.FromRegistry(spec.Image, nil) probe, err := modalProbe(spec.ReadinessProbe) if err != nil { - return "", fmt.Errorf("modal: readiness probe: %w", err) + return "", Credential{}, fmt.Errorf("modal: readiness probe: %w", err) } sb, err := c.mc.Sandboxes.Create(ctx, app, image, &modal.SandboxCreateParams{ @@ -147,9 +142,70 @@ func (c *sdkClient) CreateSandbox(ctx context.Context, spec SandboxSpec) (string ReadinessProbe: probe, }) if err != nil { - return "", err + return "", Credential{}, err } - return sb.SandboxID, nil + return sb.SandboxID, c.mintCredential(ctx, sb, spec), nil +} + +// mintCredential issues the sandbox's connect credential and RETURNS it. Nothing is +// stored here — not in a tag, not in memory. +// +// Not in a tag, because Modal's tags are plaintext and bulk-listable: one +// ListSandboxes would hand over every workload's token at once, with no revocation +// and no rotation to fall back on. Not in memory, because memory is not durable. The +// only place a credential belongs is an access-controlled Secret, which this layer +// has no cluster access to write — so it hands the pair to the one caller that does: +// the virtual kubelet, which writes both halves to a Secret in the Pod's namespace +// and the address to the Pod's endpoint annotation. +// +// Returning it is enough, and that is the part worth stating: the credential is +// minted on the one-shot CREATE path, but its consumers are durable Kubernetes +// objects, so nothing needs to re-derive it later. The annotation in etcd is the +// endpoint's record of truth (nothing on the read path ever clears it), which is +// strictly better than a tag — no provider round trip to recover a value the API +// server already stores. Modal therefore reports NO observed endpoint at all; see +// observe. +// +// Minting is one-shot: every CreateConnectToken call mints a FRESH token, so there is +// no read-back and no re-derivation later. A caller that drops the return value has +// lost the credential for the life of the sandbox. That is also why this cannot move +// to the read path — observe would hand the user a token that changed every tick. +// +// It can run this early because the RPC is keyed on the sandbox id alone +// (SandboxCreateConnectToken takes sandbox_id + port, and the SDK checks only client +// liveness): no task id, no running container, no booted GPU. Contrast Tunnels, which +// resolves only once the container is up. So the credential is in hand the moment +// Create returns, while the sandbox is still queued. +// +// Every workload gets one. An authenticated URL is the only general way to reach +// something running on a NeoCloud — there is no cluster network to fall back on — +// so the useful default is that the credential exists, and a workload with nothing +// to serve simply leaves it unused. The URL routes to the first of spec.Ports — one +// token routes to one port — and no declared port means Modal's own default (8080) +// applies. +// +// This is scoped to service-shaped workloads (a Deployment/StatefulSet/bare Pod +// that serves traffic). A Sandbox is reached by identity, not by address — +// `kubectl exec sbx-alice` — so it should NOT carry a credential; today it still +// gets one, because it reaches this adapter as an ordinary Pod and nothing here +// distinguishes the two classes yet. +// +// Best-effort by design: a sandbox that exists must be reported (and reclaimed) +// whether or not it got a credential, so a failure here returns the zero Credential +// and the instance is simply unreachable. The error is NOT returned, because that +// would fail a Provision whose sandbox is already running — leaking a paid instance +// to save an address. The error text is dropped rather than logged for the same +// reason a token is never logged: it can echo the request. +func (c *sdkClient) mintCredential(ctx context.Context, sb *modal.Sandbox, spec SandboxSpec) Credential { + creds, err := sb.CreateConnectToken(ctx, &modal.SandboxCreateConnectTokenParams{ + // Derived from the exposed set rather than carried separately, so the routed + // port cannot name one the sandbox was never told to accept traffic on. + Port: firstPort(spec.Ports), + }) + if err != nil || creds == nil || creds.Token == "" { + return Credential{} + } + return Credential{URL: creds.URL, Token: creds.Token} } // modalProbe maps a Pod readinessProbe onto Modal's Probe. Modal supports only @@ -302,15 +358,15 @@ func (c *sdkClient) ListSandboxes(ctx context.Context) ([]Sandbox, error) { return out, nil } -// observe normalizes a live SDK *Sandbox into the adapter-level Sandbox view: -// status (from Poll), tags (from GetTags), and a best-effort endpoint (from -// Tunnels). Poll and tunnel errors are tolerated so a single flaky sandbox doesn't -// fail the whole read — the poll loop will re-observe next tick. A TAG error is -// not: see below. +// observe normalizes a live SDK *Sandbox into the adapter-level Sandbox view: tags +// (from GetTags) and status (from Poll). A Poll error is tolerated so a single flaky +// sandbox doesn't fail the whole read — the poll loop will re-observe next tick. A TAG +// error is not: see below. // -// observe is a BOUNDED read: every call it makes carries a short deadline, so it -// returns promptly even for a sandbox that is still coming up. It must be, since -// it runs once per sandbox inside the List iteration. +// observe is a CHEAP read, and it must be: it runs once per sandbox inside the List +// iteration, on every poll tick. That is why it reports no endpoint (see the tail of +// this function) — an address lookup here would be a per-sandbox round trip on the +// hot path. func (c *sdkClient) observe(ctx context.Context, sb *modal.Sandbox) (Sandbox, error) { out := Sandbox{ID: sb.SandboxID} @@ -355,18 +411,17 @@ func (c *sdkClient) observe(ctx context.Context, sb *modal.Sandbox) (Sandbox, er } } - // Endpoint is only meaningful once running; look it up best-effort. - // TODO: why we need the tunnel. - if out.Status == statusRunning { - tctx, cancel := context.WithTimeout(ctx, c.endpointTimeout) - if tunnels, err := sb.Tunnels(tctx, c.endpointTimeout, &modal.SandboxTunnelsParams{}); err == nil { - for _, t := range tunnels { - out.Endpoint = t.URL() - break - } - } - cancel() - } + // No endpoint is read back. The sandbox's reachable address is its connect URL, + // minted at create and already persisted on the Pod's endpoint annotation, so + // re-deriving it per tick would be a round trip for a value the API server holds. + // + // The alternative would be a tunnel URL, and reporting one is worse than reporting + // nothing: a tunnel is PUBLIC to whoever learns it, so serving it as a stand-in for + // an authenticated URL silently downgrades access to the workload. A sandbox whose + // mint failed is therefore reported with no address at all — an honest "unreachable" + // rather than an open one — and that stays a FACT, not an error: observe's errors + // fail the entire List (see the tags block above), so one credential-less sandbox + // would otherwise stall status for every pod on the node. return out, nil } diff --git a/pkg/provider/modal/modal.go b/pkg/provider/modal/modal.go index 8126581..64ff168 100644 --- a/pkg/provider/modal/modal.go +++ b/pkg/provider/modal/modal.go @@ -64,8 +64,12 @@ var _ provider.Provider = (*Provider)(nil) // the operations the adapter needs, expressed in provider-agnostic terms, so a // real implementation (Modal SDK/HTTP) and a fake (tests) are interchangeable. type Client interface { - // CreateSandbox launches one sandbox from spec and returns its Modal id. - CreateSandbox(ctx context.Context, spec SandboxSpec) (id string, err error) + // CreateSandbox launches one sandbox from spec and returns its Modal id plus the + // connect credential minted for it. The credential is returned HERE and nowhere + // else — minting is one-shot and there is no read-back, so a caller that drops it + // has lost it for the sandbox's life. Zero when none could be minted; see + // sdkClient.mintCredential. + CreateSandbox(ctx context.Context, spec SandboxSpec) (id string, cred Credential, err error) // TerminateSandbox terminates a sandbox by id. Must be idempotent: // terminating an already-gone sandbox returns nil. TerminateSandbox(ctx context.Context, id string) error @@ -97,9 +101,10 @@ type SandboxSpec struct { // MemoryMiB is the requested memory in MiB, from the Pod's request. Zero lets // Modal apply its own default. MemoryMiB int - // Ports are the container ports to expose as encrypted tunnels, from the Pod's - // containerPorts. The reachable endpoint (reported as the Pod's address) is a - // tunnel to one of these, so a Pod that declares no port has no endpoint. + // Ports are the container ports to expose, from the Pod's containerPorts. They + // declare to Modal which ports may receive traffic at all, and the connect URL + // routes to the first of them (see firstPort) — one token routes to one port. + // Empty leaves both the exposed set and the routed port to Modal's own default. Ports []int // Timeout is the sandbox's maximum lifetime. It MUST be non-zero: Modal treats // a zero timeout as its 5-minute default, which would terminate a real @@ -118,11 +123,32 @@ type SandboxSpec struct { } // Sandbox is the adapter-level view of a Modal sandbox as observed. +// +// There is no endpoint here, deliberately. Modal's reachable address is the connect +// URL, which is minted at CREATE time (see Credential) and published straight to the +// Pod's endpoint annotation, where it persists in etcd for the sandbox's whole life. +// Re-deriving it on every read would be a Modal round trip for a value the API server +// already holds — and there is no other address to report: a tunnel URL would be one, +// but it is PUBLIC to anyone who learns it, so serving it as a substitute for an +// authenticated URL would silently downgrade access. type Sandbox struct { - ID string - Tags map[string]string - Status string // Modal's own status string, normalized by toState. - Endpoint string + ID string + Tags map[string]string + Status string // Modal's own status string, normalized by toState. +} + +// Credential is the connect pair CreateSandbox mints for a new sandbox: the URL a +// consumer calls and the bearer token that authenticates against it — +// +// curl -H "Authorization: Bearer $token" $url +// +// Token is a SECRET. It is never tagged (Modal's tags are plaintext and +// bulk-listable), never annotated onto the Pod, and never logged; the virtual kubelet +// writes the pair to a Secret in the Pod's namespace. It exists on the create path +// only — see provider.ProvisionResult, which this maps onto. +type Credential struct { + URL string + Token string } // ClaimTagKey is the sandbox tag under which the NodeClaim name is stored, so @@ -198,14 +224,18 @@ func (p *Provider) Capabilities() provider.Capabilities { // // The queued→running transition is then observed the same way readiness is, through // the poll loop's List: the sandbox reads statusInitializing until it is live. +// +// The connect credential comes back on this call and only this call, because Modal +// mints it once and cannot re-read it. The caller must persist it (the virtual kubelet +// writes it to a Secret) or it is lost; see mintCredential. func (p *Provider) Provision( ctx context.Context, pod *corev1.Pod, req provider.ProvisionRequest, -) (string, bool, error) { +) (provider.ProvisionResult, error) { if pod == nil { - return "", false, errors.New("modal: nil pod") + return provider.ProvisionResult{}, errors.New("modal: nil pod") } if req.ClaimName == "" { - return "", false, errors.New("modal: empty ClaimName in ProvisionRequest") + return provider.ProvisionResult{}, errors.New("modal: empty ClaimName in ProvisionRequest") } // Idempotency: if a sandbox already carries this claim tag, return it rather @@ -216,21 +246,36 @@ func (p *Provider) Provision( // ready) has necessarily been allocated capacity, whereas one still queued is // not yet reserved. That is strictly more information than a create can return, // so use it rather than flatly reporting false. + // + // It carries NO credential, per the interface contract: the original was minted at + // the first create and cannot be re-read, and minting a second one here would hand + // the consumer a token that changes on every retry. The consequence is a real gap — + // if the first create succeeded but its credential never reached a Secret, nothing + // recovers it. Closing that means re-minting for a sandbox with no Secret yet, which + // needs cluster access this layer does not have. if existing, err := p.findByClaim(ctx, req.ClaimName); err != nil { - return "", false, err + return provider.ProvisionResult{}, err } else if existing != nil { - return existing.ID, existing.State == provider.InstanceRunning, nil + return provider.ProvisionResult{ + InstanceID: existing.ID, + Reserved: existing.State == provider.InstanceRunning, + }, nil } spec, err := p.sandboxSpecFromPod(pod, req) if err != nil { - return "", false, err + return provider.ProvisionResult{}, err } - id, err := p.client.CreateSandbox(ctx, spec) + id, cred, err := p.client.CreateSandbox(ctx, spec) if err != nil { - return "", false, err + return provider.ProvisionResult{}, err } - return id, false, nil + return provider.ProvisionResult{ + InstanceID: id, + Reserved: false, + ConnectURL: cred.URL, + ConnectToken: cred.Token, + }, nil } // Terminate implements provider.Provider. Idempotent by the Client contract. @@ -285,6 +330,7 @@ func (p *Provider) ClassifyProvisionError(err error, accelerator, _ string) prov // findByClaim returns the sandbox tagged with claimName, or nil if none. func (p *Provider) findByClaim(ctx context.Context, claimName string) (*provider.Instance, error) { + // TODO: do we have performance issue here? sandboxes, err := p.client.ListSandboxes(ctx) if err != nil { return nil, err @@ -394,9 +440,9 @@ func resourceQty(c *corev1.Container, name corev1.ResourceName) *resource.Quanti return nil } -// containerPorts collects the container's declared ports so the Client can open a -// tunnel per port. The observed tunnel URL becomes the Pod's endpoint, so a Pod -// that declares no port is reachable-less by design. +// containerPorts collects the container's declared ports, which is what tells Modal +// which ports may receive traffic at all. The connect URL then routes to one of them +// (see firstPort). func containerPorts(c *corev1.Container) []int { if len(c.Ports) == 0 { return nil @@ -408,6 +454,17 @@ func containerPorts(c *corev1.Container) []int { return ports } +// firstPort returns the port the connect URL should route to, or 0 when the +// container declares none, which leaves the port to Modal's own default. Modal +// routes one port per token, so the first declared port wins; a workload serving on a +// second port has no address of its own today. +func firstPort(ports []int) int { + if len(ports) == 0 { + return 0 + } + return ports[0] +} + // sandboxTimeout maps the Pod's activeDeadlineSeconds (Kubernetes' own "maximum // lifetime of the pod") onto Modal's sandbox Timeout, defaulting to // defaultSandboxTimeout when the Pod does not pin one. It is never zero: a zero @@ -420,12 +477,15 @@ func sandboxTimeout(pod *corev1.Pod) time.Duration { } // toInstance normalizes a Modal sandbox into the provider-agnostic Instance. +// +// Endpoint is left empty: Modal's address is published from the create path (see +// Credential), and an observed empty endpoint never clears the annotation already on +// the Pod — the write paths all skip "". func (p *Provider) toInstance(sb Sandbox) provider.Instance { return provider.Instance{ ID: sb.ID, ClaimName: sb.Tags[ClaimTagKey], State: toState(sb.Status), - Endpoint: sb.Endpoint, // Modal is OnDemand-only; reflect that on observed instances. CapacityType: nebulav1alpha1.CapacityOnDemand, } diff --git a/pkg/provider/modal/modal_test.go b/pkg/provider/modal/modal_test.go index 3fbd899..0cce7d2 100644 --- a/pkg/provider/modal/modal_test.go +++ b/pkg/provider/modal/modal_test.go @@ -19,6 +19,7 @@ package modal import ( "context" "fmt" + "slices" "testing" "time" @@ -40,21 +41,22 @@ type fakeClient struct { createCnt int createErr error createID string + cred Credential // credential CreateSandbox returns; zero = none minted terminated []string } -func (f *fakeClient) CreateSandbox(_ context.Context, spec SandboxSpec) (string, error) { +func (f *fakeClient) CreateSandbox(_ context.Context, spec SandboxSpec) (string, Credential, error) { f.createCnt++ f.lastSpec = spec if f.createErr != nil { - return "", f.createErr + return "", Credential{}, f.createErr } id := f.createID if id == "" { id = "sb-new" } f.sandboxes = append(f.sandboxes, Sandbox{ID: id, Tags: spec.Tags, Status: "pending"}) - return id, nil + return id, f.cred, nil } func (f *fakeClient) TerminateSandbox(_ context.Context, id string) error { @@ -118,13 +120,14 @@ func TestProvision_GPUPod(t *testing.T) { f := &fakeClient{createID: "sb-1"} p := newTestProvider(f) - id, reserved, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 2), provider.ProvisionRequest{ + res, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 2), provider.ProvisionRequest{ ClaimName: "claim-a", CapacityType: nebulav1alpha1.CapacityOnDemand, }) if err != nil { t.Fatalf("Provision: %v", err) } + id, reserved := res.InstanceID, res.Reserved if id != "sb-1" { t.Fatalf("id = %q, want sb-1", id) } @@ -155,7 +158,7 @@ func TestProvision_LowercaseGPUAnnotation(t *testing.T) { // A user may write the accelerator-type label in any case (e.g. "h100"). It must // resolve to the canonical catalog accelerator ("H100") so the provisioned // sandbox — and any downstream key (blocklist/catalog) — uses one casing. - _, _, err := p.Provision(context.Background(), gpuPod("claim-lc", "h100", 1), provider.ProvisionRequest{ + _, err := p.Provision(context.Background(), gpuPod("claim-lc", "h100", 1), provider.ProvisionRequest{ ClaimName: "claim-lc", CapacityType: nebulav1alpha1.CapacityOnDemand, }) @@ -183,7 +186,7 @@ func TestProvision_MapsResourcesPortsAndTimeout(t *testing.T) { deadline := int64(3600) pod.Spec.ActiveDeadlineSeconds = &deadline - if _, _, err := p.Provision(context.Background(), pod, provider.ProvisionRequest{ClaimName: "claim-res"}); err != nil { + if _, err := p.Provision(context.Background(), pod, provider.ProvisionRequest{ClaimName: "claim-res"}); err != nil { t.Fatalf("Provision: %v", err) } if f.lastSpec.CPU != 2.5 { @@ -207,7 +210,7 @@ func TestProvision_DefaultsTimeoutWhenNoDeadline(t *testing.T) { // No activeDeadlineSeconds: the adapter must still set a non-zero timeout, else // Modal applies its 5-minute default and the workload dies almost immediately. req := provider.ProvisionRequest{ClaimName: "claim-dt"} - if _, _, err := p.Provision(context.Background(), gpuPod("claim-dt", "H100", 1), req); err != nil { + if _, err := p.Provision(context.Background(), gpuPod("claim-dt", "H100", 1), req); err != nil { t.Fatalf("Provision: %v", err) } if f.lastSpec.Timeout != defaultSandboxTimeout { @@ -219,7 +222,7 @@ func TestProvision_CPUOnly(t *testing.T) { f := &fakeClient{} p := newTestProvider(f) - _, _, err := p.Provision(context.Background(), gpuPod("claim-cpu", "", 0), provider.ProvisionRequest{ + _, err := p.Provision(context.Background(), gpuPod("claim-cpu", "", 0), provider.ProvisionRequest{ ClaimName: "claim-cpu", CapacityType: nebulav1alpha1.CapacityOnDemand, }) @@ -242,10 +245,11 @@ func TestProvision_Idempotent(t *testing.T) { p := newTestProvider(f) req := provider.ProvisionRequest{ClaimName: "claim-a"} - id, reserved, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), req) + res, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), req) if err != nil { t.Fatalf("Provision: %v", err) } + id, reserved := res.InstanceID, res.Reserved if id != "sb-existing" { t.Fatalf("id = %q, want sb-existing (idempotent reuse)", id) } @@ -272,11 +276,12 @@ func TestProvision_IdempotentInitializingIsNotReserved(t *testing.T) { } p := newTestProvider(f) - id, reserved, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), + res, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), provider.ProvisionRequest{ClaimName: "claim-a"}) if err != nil { t.Fatalf("Provision: %v", err) } + id, reserved := res.InstanceID, res.Reserved if id != "sb-existing" { t.Fatalf("id = %q, want sb-existing (idempotent reuse)", id) } @@ -289,7 +294,7 @@ func TestProvision_UnsupportedAccelerator(t *testing.T) { f := &fakeClient{} p := newTestProvider(f) req := provider.ProvisionRequest{ClaimName: "claim-x"} - _, _, err := p.Provision(context.Background(), gpuPod("claim-x", "TPU-v4", 1), req) + _, err := p.Provision(context.Background(), gpuPod("claim-x", "TPU-v4", 1), req) if err == nil { t.Fatal("expected error for unsupported accelerator") } @@ -450,7 +455,7 @@ func TestProvision_ProbeTagStampedOnlyWithProbe(t *testing.T) { pod := gpuPod("claim-a", "H100", 1) pod.Spec.Containers[0].ReadinessProbe = tc.probe - if _, _, err := p.Provision(context.Background(), pod, provider.ProvisionRequest{ClaimName: "claim-a"}); err != nil { + if _, err := p.Provision(context.Background(), pod, provider.ProvisionRequest{ClaimName: "claim-a"}); err != nil { t.Fatalf("Provision: %v", err) } _, present := f.lastSpec.Tags[ProbeTagKey] @@ -480,7 +485,7 @@ func TestProvision_ReadinessProbeCarriedThrough(t *testing.T) { } pod.Spec.Containers[0].ReadinessProbe = probe - if _, _, err := p.Provision(context.Background(), pod, provider.ProvisionRequest{ClaimName: "claim-a"}); err != nil { + if _, err := p.Provision(context.Background(), pod, provider.ProvisionRequest{ClaimName: "claim-a"}); err != nil { t.Fatalf("Provision: %v", err) } // The probe is carried onto the spec so the Client can configure Modal's own @@ -495,7 +500,7 @@ func TestProvision_NoProbeLeavesSpecUnset(t *testing.T) { p := newTestProvider(f) req := provider.ProvisionRequest{ClaimName: "claim-a"} - if _, _, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), req); err != nil { + if _, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), req); err != nil { t.Fatalf("Provision: %v", err) } if f.lastSpec.ReadinessProbe != nil { @@ -503,6 +508,157 @@ func TestProvision_NoProbeLeavesSpecUnset(t *testing.T) { } } +// The spec carries the container's declared ports verbatim: they are the set Modal is +// told to accept traffic on, and the connect URL routes to the first of them (the +// client derives it, so the routed port can never name one outside the set). No +// declared port is not "no endpoint" — every workload is credentialed — it means Modal +// picks, defaulting to 8080. +func TestProvision_CarriesDeclaredPorts(t *testing.T) { + for _, tc := range []struct { + name string + ports []corev1.ContainerPort + want []int + // wantRouted is the port the connect URL ends up on, which the client derives + // from want; 0 leaves it to Modal. + wantRouted int + }{ + {"no ports leaves the port to Modal", nil, nil, 0}, + {"single port", []corev1.ContainerPort{{ContainerPort: 8000}}, []int{8000}, 8000}, + // Modal routes one port per token, so the first declared port wins — but the + // whole set is still exposed. + { + "all exposed, first routed", + []corev1.ContainerPort{{ContainerPort: 8000}, {ContainerPort: 9090}}, + []int{8000, 9090}, + 8000, + }, + } { + t.Run(tc.name, func(t *testing.T) { + f := &fakeClient{createID: "sb-1"} + p := newTestProvider(f) + pod := gpuPod("claim-a", "H100", 1) + pod.Spec.Containers[0].Ports = tc.ports + + req := provider.ProvisionRequest{ClaimName: "claim-a"} + if _, err := p.Provision(context.Background(), pod, req); err != nil { + t.Fatalf("Provision: %v", err) + } + if !slices.Equal(f.lastSpec.Ports, tc.want) { + t.Fatalf("Ports = %v, want %v", f.lastSpec.Ports, tc.want) + } + if got := firstPort(f.lastSpec.Ports); got != tc.wantRouted { + t.Fatalf("routed port = %d, want %d", got, tc.wantRouted) + } + }) + } +} + +func TestFirstPort(t *testing.T) { + if got := firstPort(nil); got != 0 { + t.Fatalf("firstPort(nil) = %d, want 0 (leave the port to Modal)", got) + } + if got := firstPort([]int{9090, 8000}); got != 9090 { + t.Fatalf("firstPort = %d, want the first declared port 9090", got) + } +} + +// The credential reaches the caller through Provision and NOWHERE else: minting is +// one-shot, so this return value is the only copy that will ever exist. +func TestProvision_ReturnsMintedCredential(t *testing.T) { + f := &fakeClient{ + createID: "sb-1", + cred: Credential{URL: "https://x.modal.host", Token: "tok-abc"}, + } + p := newTestProvider(f) + + res, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), + provider.ProvisionRequest{ClaimName: "claim-a"}) + if err != nil { + t.Fatalf("Provision: %v", err) + } + if res.ConnectURL != "https://x.modal.host" { + t.Fatalf("ConnectURL = %q, want the minted URL", res.ConnectURL) + } + if res.ConnectToken != "tok-abc" { + t.Fatalf("ConnectToken = %q, want the minted token", res.ConnectToken) + } + // The token must never be written where a reader of the sandbox could find it: the + // tags are plaintext and one ListSandboxes dumps them all. + for k, v := range f.lastSpec.Tags { + if v == "tok-abc" { + t.Fatalf("token leaked into sandbox tag %q", k) + } + } +} + +// A sandbox that minted nothing yields no credential rather than an error: it still +// exists, still costs money, and must still be reported and reclaimed. +func TestProvision_NoCredentialWhenNoneMinted(t *testing.T) { + f := &fakeClient{createID: "sb-1"} // zero cred + p := newTestProvider(f) + + res, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), + provider.ProvisionRequest{ClaimName: "claim-a"}) + if err != nil { + t.Fatalf("Provision: %v", err) + } + if res.InstanceID != "sb-1" { + t.Fatalf("InstanceID = %q, want sb-1 even with no credential", res.InstanceID) + } + if res.ConnectURL != "" || res.ConnectToken != "" { + t.Fatalf("expected no credential, got url=%q token set=%t", res.ConnectURL, res.ConnectToken != "") + } +} + +// An idempotent re-Provision carries NO credential. The original was minted once and +// cannot be re-read, and minting a second one here would hand the consumer a token +// that changes on every retry. +func TestProvision_IdempotentReturnsNoCredential(t *testing.T) { + f := &fakeClient{ + sandboxes: []Sandbox{{ + ID: "sb-existing", + Tags: map[string]string{ClaimTagKey: "claim-a"}, + Status: statusRunning, + }}, + cred: Credential{URL: "https://x.modal.host", Token: "tok-abc"}, + } + p := newTestProvider(f) + + res, err := p.Provision(context.Background(), gpuPod("claim-a", "H100", 1), + provider.ProvisionRequest{ClaimName: "claim-a"}) + if err != nil { + t.Fatalf("Provision: %v", err) + } + if res.InstanceID != "sb-existing" { + t.Fatalf("InstanceID = %q, want sb-existing", res.InstanceID) + } + if res.ConnectURL != "" || res.ConnectToken != "" { + t.Fatalf("an adopted sandbox must carry no credential, got url=%q token set=%t", + res.ConnectURL, res.ConnectToken != "") + } +} + +// Modal reports NO observed endpoint. Its address is the connect URL, published from +// the create path onto the Pod's annotation, where it persists; re-deriving it per tick +// would be a round trip for a value the API server already holds. The alternative — +// falling back to a tunnel URL — is worse than nothing, since a tunnel is public to +// whoever learns it. +func TestToInstance_ReportsNoEndpoint(t *testing.T) { + p := newTestProvider(&fakeClient{}) + + got := p.toInstance(Sandbox{ + ID: "sb-1", + Status: statusRunning, + Tags: map[string]string{ClaimTagKey: "claim-a"}, + }) + if got.Endpoint != "" { + t.Fatalf("Endpoint = %q, want empty; the address comes from the create path", got.Endpoint) + } + if got.ClaimName != "claim-a" || got.State != provider.InstanceRunning { + t.Fatalf("claim/state = %q/%q, want claim-a/Running", got.ClaimName, got.State) + } +} + func TestModalProbe(t *testing.T) { // nil Pod probe => no Modal probe (probe-less workload). if got, err := modalProbe(nil); err != nil || got != nil { diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go index 0b5a207..340e204 100644 --- a/pkg/provider/provider.go +++ b/pkg/provider/provider.go @@ -27,6 +27,7 @@ package provider import ( "context" + "fmt" "time" corev1 "k8s.io/api/core/v1" @@ -56,10 +57,11 @@ type Provider interface { // only what the Pod cannot express: the optimizer-chosen capacity tier and // the claim identity. // - // It returns the provider instance id plus whether that instance is RESERVED: - // whether the provider has committed actual capacity to it, as opposed to - // merely accepting the request. The two are genuinely different guarantees and - // an id alone cannot express either one: + // It returns a ProvisionResult: the instance id, whether capacity is RESERVED, + // and any connect credential minted for the instance. Reserved is whether the + // provider has committed actual capacity, as opposed to merely accepting the + // request. The two are genuinely different guarantees and an id alone cannot + // express either one: // // - AWS reserves. CreateFleet with an *instant* request is synchronous, so an // id means EC2 found capacity and the instance is booting on real hardware; @@ -78,8 +80,10 @@ type Provider interface { // // Idempotency: if an instance already exists for req.ClaimName (encoded in // the provider's naming scheme, since most providers lack tags), return that - // id instead of creating a second. - Provision(ctx context.Context, pod *corev1.Pod, req ProvisionRequest) (instanceID string, reserved bool, err error) + // id instead of creating a second. Such a call returns NO credential — the + // original one was already published and cannot be re-read (see + // ProvisionResult.ConnectURL). + Provision(ctx context.Context, pod *corev1.Pod, req ProvisionRequest) (ProvisionResult, error) // Terminate destroys the instance by id. Must be idempotent: terminating an // already-gone instance returns nil (so the NodeClaim finalizer can retry @@ -167,6 +171,57 @@ type ProvisionRequest struct { Region string } +// ProvisionResult is what one Provision call produced. It is a struct rather than +// more positional returns because the credential below is delivered ONCE, and a +// value that can only ever be observed here deserves to be named. +type ProvisionResult struct { + // InstanceID is the provider's id for the instance. Non-empty on success, and + // carrying the full teardown obligation from that moment on. + InstanceID string + // Reserved is whether the provider committed actual capacity, as opposed to + // merely accepting the request. See Provision. + Reserved bool + // ConnectURL and ConnectToken are the credential for reaching the instance: an + // address plus the bearer token that authenticates against it. Together they are + // what a consumer needs and all they need: + // + // curl -H "Authorization: Bearer $token" $url + // + // They are returned HERE, and only here, because minting is one-shot: providers + // issue a credential at create and offer no way to read it back (Modal's + // CreateConnectToken returns a DIFFERENT token on a second call, so there is not + // even a stable value to re-read). This is therefore the single moment the pair + // exists in Nebula's hands, and the caller must persist it durably — the virtual + // kubelet writes both to a Secret in the Pod's namespace — or it is gone. Empty + // when the instance needs no credential, and empty on an idempotent re-Provision + // of an existing instance, whose credential was published already. + // + // The TOKEN is a SECRET: never log it, never put it on the Pod (an annotation is + // readable by anyone with `get pod` and lands unencrypted in etcd), never in an + // error string. Neither field is on Instance: List/Get are the level-triggered + // read path, and a credential that cannot be re-read has no business there. + // + // ConnectURL is not itself secret, and it is NOT the same value as + // Instance.Endpoint even when they agree. This is the one-shot create path; the + // endpoint is the observed read path, which is what reports an address that does + // not exist until the instance boots (AWS's public DNS name) and what still + // reports one after a restart, when no Provision call happens at all. + ConnectURL string + ConnectToken string +} + +// String redacts ConnectToken so a ProvisionResult can be logged safely. The +// compiler cannot stop a future log.Info("...", "result", res) from leaking it; +// %v/%s go through here instead. GoString covers %#v for the same reason. The URL +// is not secret and is printed as-is. +func (r ProvisionResult) String() string { + return fmt.Sprintf("ProvisionResult{InstanceID:%s Reserved:%t ConnectURL:%s ConnectToken:%s}", + r.InstanceID, r.Reserved, r.ConnectURL, redacted(r.ConnectToken)) +} + +// GoString implements fmt.GoStringer so %#v is redacted too. +func (r ProvisionResult) GoString() string { return r.String() } + // Capabilities declares provider quirks as data, so the control plane filters // and behaves generically rather than branching on provider name. type Capabilities struct { @@ -209,13 +264,23 @@ type Instance struct { ID string ClaimName string // recovered from the naming scheme (for tag-less providers) State InstanceState - // Endpoint is the reachable address once ready (e.g. SSH host:port). - Endpoint string // CapacityType reflects how the instance was provisioned, when known. CapacityType nebulav1alpha1.CapacityType // Region is where the instance actually lives, in the provider's own // vocabulary. Empty for region-simple providers that do not report one. Region string + // Endpoint is the reachable address once ready (e.g. SSH host:port). It is not + // secret, so unlike the connect credential it rides the level-triggered read path + // and is re-reported on every tick. + Endpoint string +} + +// redacted renders a secret's presence without its value. +func redacted(s string) string { + if s == "" { + return "" + } + return "[REDACTED]" } // InstanceState is the provider-agnostic lifecycle state, normalized from each diff --git a/pkg/vnode/handler.go b/pkg/vnode/handler.go index 132dfe8..b602c93 100644 --- a/pkg/vnode/handler.go +++ b/pkg/vnode/handler.go @@ -147,11 +147,15 @@ type trackedPod struct { pod *corev1.Pod claimName string instance string - // persistedEndpoint is the endpoint value last written to the Pod's metadata - // annotation via persistEndpoint. The notify callback fires every poll tick, so + // connectEndpoint is the endpoint value last written to the Pod's metadata + // annotation via patchEndpoint. The notify callback fires every poll tick, so // this dedups the metadata patch to the ticks where the reachable address // actually changed (first appearance, or a re-provision) rather than every tick. - persistedEndpoint string + // + // It is a cache, not a record: losing it (a restart) costs one redundant patch, + // never a lost value, because the annotation itself lives in etcd. No credential is + // held here — see persistCredential. + connectEndpoint string } // NewHandler builds a Handler for the given provider backend. The poll cadence @@ -203,11 +207,17 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // Bound the provision call so a wedged backend cannot pin this worker forever. // The provider may raise the deadline via Capabilities.ProvisionTimeout (AWS // does, to leave room for cross-zone failover); zero means "use the default". + // + // The deadline is scoped to that ONE call and must not be reused for the writes + // that follow it. A Provision returning just under the timeout would leave those + // writes with no time budget at all, so they would fail with DeadlineExceeded on + // success — and the credential write below cannot be retried, so a timeout there + // loses the token for good. The follow-up writes stay on the caller's ctx. timeout := h.prov.Capabilities().ProvisionTimeout if timeout <= 0 { timeout = defaultProvisionTimeout } - ctx, cancel := context.WithTimeout(ctx, timeout) + provisionCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() log.Info("provisioning external instance", @@ -219,7 +229,7 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { h.markStatus(pod, corev1.PodPending, reasonProvisioning, "allocating external instance") h.emit(pod) - id, reserved, err := h.prov.Provision(ctx, pod, req) + res, err := h.prov.Provision(provisionCtx, pod, req) if err != nil { log.Error(err, "provision failed; Pod marked Failed for failover") // Record the failure on the shared blocklist so placement fails over to the @@ -247,15 +257,63 @@ func (h *Handler) CreatePod(ctx context.Context, pod *corev1.Pod) error { // store runs either way: an id means the instance exists. markStatus precedes it // because store deep-copies — storing first would track a copy without the status // just written. - log.Info("external instance provisioned", "instanceID", id, "reserved", reserved) - if reserved { + log.Info("external instance provisioned", "instanceID", res.InstanceID, "reserved", res.Reserved) + if res.Reserved { h.markStatus(pod, corev1.PodPending, reasonInitializing, "external instance is initializing") } - h.store(pod, claim, id) + // Stamp a create-time address BEFORE store, because store deep-copies and the + // tracked copy is what the poll loop re-emits. That is the whole publication: the + // emit below carries it to the API server through the same notify wrapper the poll + // loop uses, and every later tick re-offers it until a write lands. No lock: this + // Pod is not shared until store, VK having handed us a copy of its own. + setEndpoint(pod, res.ConnectURL) + h.store(pod, claim, res.InstanceID) + + // The TOKEN, unlike the address, cannot ride the Pod (it would be readable by + // anyone with `get pod` and sit unencrypted in etcd), so it gets its own write + // here — the only place it exists, since the provider mints it once and cannot + // re-read it (see provider.ProvisionResult). It runs after store and before emit so + // a tracked pod is never observable without its credential, and it does not gate + // the status: a Pod that provisioned is reported provisioned even if the Secret + // write failed. + h.persistCredential(ctx, pod, res.ConnectURL, res.ConnectToken) + h.emit(pod) return nil } +// persistCredential writes the bearer token, and a copy of the address it +// authenticates against, into a Secret. +// +// It handles ONLY the secret half. The address is published by stamping it on the Pod +// (see CreatePod / setEndpoint), which routes it through the one endpoint write path +// every provider shares; this function exists because the token cannot travel that way. +// An annotation is readable by anyone with `get pod` and lands unencrypted in etcd, +// which is fine for an address and unacceptable for a credential, so the token goes to +// an access-controlled Secret — with the URL alongside it, so the pair is usable from +// one object. +// +// The token is one-shot and unrepeatable: minting is create-only with no read-back, so +// a failed write cannot be retried, here or later. That is the difference from the +// address, which the poll loop keeps re-offering until it lands. The Secret is instead +// written once and never rewritten, so it needs nothing held in memory. +// +// An empty url means the provider mints no credential (AWS, whose address is not known +// until boot and is reported through the observed endpoint instead) or that minting +// failed; an empty token means an address with nothing to authenticate, so a Secret +// would imply a credential that does not exist. Either way there is nothing to write +// and nothing to fail — the poll loop still reports the instance, and an unreachable +// workload is reported unreachable. +// +// Best-effort: a nil client (tests) is a no-op, and a failure is logged, never with the +// token. +func (h *Handler) persistCredential(ctx context.Context, pod *corev1.Pod, url, token string) { + if h.client == nil || url == "" || token == "" { + return + } + h.createConnectSecret(ctx, pod, url, token) +} + // UpdatePod is a no-op: the external instance's shape is immutable once // provisioned (recovery from any change is delete-and-recreate, matching the // NodeClaim ledger's immutability). We still refresh our tracked copy so @@ -264,10 +322,17 @@ func (h *Handler) UpdatePod(_ context.Context, pod *corev1.Pod) error { h.mu.Lock() defer h.mu.Unlock() if tp, ok := h.tracked[key(pod.Namespace, pod.Name)]; ok { - // Preserve the status we compute from the provider; only adopt spec/meta. + // Preserve what WE own and the API server does not yet know: the status we + // compute from the provider, and the endpoint — which may be an address minted + // at create whose patch has not landed yet, so the incoming Pod would not carry + // it. Dropping it would discard the only copy (a minted URL is never + // re-observed) and strand the retry with nothing to replay. Everything else is + // adopted from the incoming spec/meta. status := tp.pod.Status + endpoint := tp.pod.Annotations[nebulav1alpha1.EndpointAnnotation] tp.pod = pod.DeepCopy() tp.pod.Status = status + setEndpoint(tp.pod, endpoint) } return nil } @@ -390,10 +455,10 @@ func (h *Handler) GetPods(_ context.Context) ([]*corev1.Pod, error) { // (enqueuePodStatusUpdate -> UpdateStatus), which writes only the /status // subresource and silently drops any metadata change on the same object. The // reachable endpoint must live on the Pod's metadata (PodIP cannot hold a DNS -// name — see applyState), so the wrapper first persists the endpoint annotation -// with its own metadata patch, then hands the same Pod to VK for the status -// write. This folds the two writes onto the one notify signal: every status push -// also reconciles the endpoint annotation (deduped so only a real change patches). +// name — see applyState), so the wrapper publishes the instance's access details +// with their own writes first, then hands the same Pod to VK for the status write. +// This folds those writes onto the one notify signal: every status push also +// reconciles how the workload is reached. func (h *Handler) NotifyPods(ctx context.Context, cb func(*corev1.Pod)) { h.mu.Lock() h.notify = func(pod *corev1.Pod) { @@ -456,16 +521,9 @@ func (h *Handler) reconcileOnce(ctx context.Context) { } else { matched++ applyState(tp.pod, inst.State, inst.Endpoint, h.nowFn()) - // Surface the reachable address on the Pod metadata. PodIP can't hold a DNS - // name (see applyState), so the endpoint always rides an annotation. Set it - // on the tracked Pod here; the notify wrapper persists it to the API server - // (deduped against persistedEndpoint so only a real change patches). - if inst.Endpoint != "" && tp.pod.Annotations[nebulav1alpha1.EndpointAnnotation] != inst.Endpoint { - if tp.pod.Annotations == nil { - tp.pod.Annotations = map[string]string{} - } - tp.pod.Annotations[nebulav1alpha1.EndpointAnnotation] = inst.Endpoint - } + // The observed address, for a provider that cannot know it before boot. + // Empty for one that published at create, which must not clear it. + setEndpoint(tp.pod, inst.Endpoint) } // Log the before -> after status every tick. This is the "how does the system // look" signal an operator watches: the lifecycle progression (Provisioning -> @@ -505,6 +563,38 @@ func (h *Handler) reconcileOnce(ctx context.Context) { } } +// setEndpoint stamps a reachable address onto a Pod's annotation — the single stamp +// every path uses, so the annotation has one assignment site regardless of where the +// address came from. The notify wrapper then patches it to the API server (PodIP cannot +// hold a DNS name — see applyState), and its write half (persistEndpoint) reads the +// value back off the emitted Pod knowing nothing about its origin. +// +// Callers, and what each one knows: +// +// - CreatePod — an address MINTED at create, from ProvisionResult. Modal's connect +// URL, which exists before the sandbox does and is never reported again. +// - reconcileOnce — an address OBSERVED at boot, from a listed Instance. AWS assigns a +// public DNS name only once EC2 has one, so it can only arrive on the read path. +// - UpdatePod — nothing new; it re-applies what VK's replacement Pod may have +// dropped. +// +// An empty address is ignored rather than cleared, which is what lets those coexist: a +// provider that published at create and then reports no observed endpoint (Modal), or +// one that momentarily omits the value, never erases a working address. No credential +// is ever stamped here — a token cannot ride the Pod (see persistCredential), and +// cannot be observed at all, being minted once on the create path. +// +// Callers stamping a tracked pod's own Pod must hold h.mu. +func setEndpoint(pod *corev1.Pod, endpoint string) { + if endpoint == "" || pod.Annotations[nebulav1alpha1.EndpointAnnotation] == endpoint { + return + } + if pod.Annotations == nil { + pod.Annotations = map[string]string{} + } + pod.Annotations[nebulav1alpha1.EndpointAnnotation] = endpoint +} + // statusSignature is a compact rendering of the Pod status fields the poll loop // surfaces to the API server, logged before/after each tick so a pod's lifecycle // progression is visible. It intentionally goes beyond Phase: reason (Provisioning @@ -543,21 +633,24 @@ func (h *Handler) emit(pod *corev1.Pod) { } } -// persistEndpoint patches the endpoint annotation onto the Pod metadata. It runs -// inside the notify wrapper (see NotifyPods), just before VK's status callback: -// VK's callback writes only the /status subresource and drops any metadata change -// on the same object, so the reachable address — which is how anything reaches the -// workload, and which PodIP cannot hold when it is a DNS name (see applyState) — -// needs this dedicated metadata write. A merge patch scoped to the single -// annotation is a no-op for every other field and does not collide with the status -// write that follows. +// persistEndpoint writes the reachable address on the emitted Pod to the API server. It +// is the write half for whichever path stamped it (see setEndpoint): CreatePod for an +// address minted at create, the poll loop for one observed at boot. It runs inside the +// notify wrapper (see NotifyPods), just before VK's status callback, because VK's +// callback writes only the /status subresource and drops everything else on the same +// object. +// +// Because the poll loop re-emits every tracked pod each tick, this is also the retry for +// any patch that failed, including a create-time one: it keeps patching until a write +// succeeds. It carries no credential — a token is written once on the create path (see +// persistCredential), never per tick. +// +// This runs per pod per tick, so anything it does unconditionally is multiplied by the +// whole fleet — hence the dedup below, which narrows it to the ticks where the address +// changed or a previous patch has not yet landed. // -// The notify callback fires every poll tick, so this dedups against the tracked -// pod's persistedEndpoint and patches only when the value actually changed (first -// appearance or a re-provision); a steady Running pod is never re-patched. It is -// best-effort: a nil client (tests) or an endpoint-less Pod is a no-op, and a -// failed patch is logged and retried next tick (persistedEndpoint is only advanced -// on success), never fatal to the poll. NotFound is ignored — the Pod is gone. +// Best-effort: a nil client (tests) is a no-op, and a failure is logged and retried +// next tick rather than failing the poll. func (h *Handler) persistEndpoint(ctx context.Context, pod *corev1.Pod) { if h.client == nil { return @@ -567,15 +660,38 @@ func (h *Handler) persistEndpoint(ctx context.Context, pod *corev1.Pod) { return } - // Dedup: skip the patch when we have already persisted this exact endpoint. h.mu.Lock() tp, tracked := h.tracked[key(pod.Namespace, pod.Name)] - if tracked && tp.persistedEndpoint == endpoint { - h.mu.Unlock() - return + // An untracked pod has nothing to dedup against, so its endpoint is patched + // unconditionally: the annotation is the only place that address is published, and + // skipping it would strand an unreachable Pod. + patched := false + if tracked { + patched = tp.connectEndpoint == endpoint } h.mu.Unlock() + if !patched { + h.patchEndpoint(ctx, pod, endpoint) + } +} + +// patchEndpoint patches the endpoint annotation onto the Pod metadata. The +// reachable address — which is how anything reaches the workload, and which PodIP +// cannot hold when it is a DNS name (see applyState) — needs this dedicated +// metadata write, since VK's status callback drops metadata. A merge patch scoped to +// the single annotation is a no-op for every other field and so does not collide +// with the status write that follows. +// +// This is the ONLY write of the endpoint annotation, and persistEndpoint its only +// caller, so every address — minted at create or observed at boot — reaches etcd +// through this one merge patch. The annotation is then where the address LIVES, and +// nothing clears it: this is only ever called with a non-empty value, so it survives +// for the Pod's life without the provider being asked for it again. +// +// connectEndpoint advances only on success, so a failed patch is retried on the next +// tick. NotFound is ignored — the Pod is gone. +func (h *Handler) patchEndpoint(ctx context.Context, pod *corev1.Pod, endpoint string) { patch, err := json.Marshal(map[string]any{ "metadata": map[string]any{ "annotations": map[string]string{nebulav1alpha1.EndpointAnnotation: endpoint}, @@ -584,28 +700,105 @@ func (h *Handler) persistEndpoint(ctx context.Context, pod *corev1.Pod) { if err != nil { // A marshal of a fixed-shape map cannot realistically fail; guard anyway so a // future change surfaces rather than panics. - logf.FromContext(ctx).WithName("vnode-poll").Error(err, + logf.FromContext(ctx).WithName("vnode-handler").Error(err, "marshal endpoint annotation patch", "pod", key(pod.Namespace, pod.Name)) return } if _, err := h.client.CoreV1().Pods(pod.Namespace).Patch( ctx, pod.Name, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil { if !apierrors.IsNotFound(err) { - logf.FromContext(ctx).WithName("vnode-poll").Error(err, - "persist endpoint annotation; will retry next tick", + logf.FromContext(ctx).WithName("vnode-handler").Error(err, + "persist endpoint annotation; the poll loop retries next tick", "pod", key(pod.Namespace, pod.Name), "endpoint", endpoint) } - return // leave persistedEndpoint unchanged so the next tick retries + return // leave connectEndpoint unchanged so the next tick retries } // Record success so subsequent ticks skip the patch until the endpoint changes. h.mu.Lock() if tp, ok := h.tracked[key(pod.Namespace, pod.Name)]; ok { - tp.persistedEndpoint = endpoint + tp.connectEndpoint = endpoint } h.mu.Unlock() } +// ConnectSecretName is the Secret holding a Pod's connect credential. +func ConnectSecretName(podName string) string { return podName + "-connect" } + +// createConnectSecret writes the instance's connect URL and bearer token to a +// Secret in the Pod's namespace, so a consumer can reach the workload with +// `curl -H "Authorization: Bearer $token" $url`. +// +// A Secret, not an annotation: the token authenticates every request to the +// endpoint, and an annotation would expose it to anyone with `get pod` in the +// namespace and store it unencrypted in etcd. The URL is duplicated in here +// alongside it so the pair is usable from one object. +// +// It is ownerReferenced to the Pod, which is what reclaims it: teardown deletes the +// Pod, and the garbage collector removes the Secret with it. Deleting it on the +// DeletePod path instead would leak on every path that skips DeletePod (a force +// delete, a VK outage) — the same reason the NodeClaim finalizer exists. +// +// Write-once, and unrepeatable: this runs on the create path, from the one +// ProvisionResult that ever carries the token (see persistCredential). There is no +// retry loop behind it, because there is nothing to retry WITH — a later attempt has +// no credential to write, since minting is one-shot and cannot be re-read. So a +// failure here means the Secret is missing for this instance's life, and the recovery +// is to replace the instance (delete the Pod), not to poll. +// +// AlreadyExists is therefore treated as success rather than reconciled: it means a +// Secret under this name already exists, which happens when a Pod name is reused +// before the GC has collected the previous owner's Secret. Overwriting is not +// obviously better — the old Secret still belongs to a live-until-collected instance — +// and the stale copy resolves itself when the ownerReference GC removes it. +// +// A failure is logged (never with the token). +func (h *Handler) createConnectSecret(ctx context.Context, pod *corev1.Pod, url, token string) { + // No UID means the Pod is synthesized (GetPod's re-adoption stub), so an + // ownerReference would be invalid and the Secret would never be collected. Skip + // rather than create an unowned Secret that nothing would ever reclaim. In practice + // the create path always has the real Pod, so this is a guard, not a case. + if pod.UID == "" { + return + } + k := key(pod.Namespace, pod.Name) + log := logf.FromContext(ctx).WithName("vnode-handler") + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: ConnectSecretName(pod.Name), + Namespace: pod.Namespace, + Labels: map[string]string{ + nebulav1alpha1.ManagedByLabel: nebulav1alpha1.ManagedByValue, + }, + // UID is what makes this a real ownerReference; a name alone would be + // ignored by the GC. It is set on any Pod that exists at the API server. + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: "v1", + Kind: "Pod", + Name: pod.Name, + UID: pod.UID, + }}, + }, + Type: corev1.SecretTypeOpaque, + StringData: map[string]string{ + "token": token, + "url": url, + }, + } + _, err := h.client.CoreV1().Secrets(pod.Namespace).Create(ctx, secret, metav1.CreateOptions{}) + switch { + case apierrors.IsAlreadyExists(err): + return // a Secret under this name already exists; see above + case err != nil: + // Loud, because it is not retried: the token cannot be re-minted, so this + // instance stays credential-less until it is replaced. + log.Error(err, "write connect secret; the workload's credential is LOST (delete the Pod to re-provision)", + "pod", k, "secret", ConnectSecretName(pod.Name)) + return + } + log.Info("wrote connect secret", "pod", k, "secret", ConnectSecretName(pod.Name)) +} + // markStatus sets a coarse pod phase and a single readiness-style condition on // the passed Pod (which VK then reports to the API server). func (h *Handler) markStatus(pod *corev1.Pod, phase corev1.PodPhase, reason, msg string) { diff --git a/pkg/vnode/handler_test.go b/pkg/vnode/handler_test.go index 15615a1..1c6d227 100644 --- a/pkg/vnode/handler_test.go +++ b/pkg/vnode/handler_test.go @@ -27,9 +27,12 @@ import ( "github.com/virtual-kubelet/virtual-kubelet/errdefs" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/fake" + corev1client "k8s.io/client-go/kubernetes/typed/core/v1" k8stesting "k8s.io/client-go/testing" nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" @@ -46,15 +49,19 @@ type fakeProvider struct { // case), so a test that cares about the reserved path must opt in — the zero value // should not silently assert the stronger guarantee. provisionReserved bool - provisionErr error - provisionCnt int - lastReq provider.ProvisionRequest - terminateCnt int - terminateID string - terminateErr error - list []provider.Instance - listErr error - capabilities provider.Capabilities + // provisionURL/provisionToken are the connect credential Provision returns — the + // one-shot value the handler must persist, since it can never be re-read. + provisionURL string + provisionToken string + provisionErr error + provisionCnt int + lastReq provider.ProvisionRequest + terminateCnt int + terminateID string + terminateErr error + list []provider.Instance + listErr error + capabilities provider.Capabilities // classifyScope is what ClassifyProvisionError returns for a failure; the zero // value (empty scope) means "not blocklistable". classifyAccel/classifyRegion // record what the handler passed in, so a test can assert it resolved them off the @@ -65,24 +72,40 @@ type fakeProvider struct { // provisionHook runs inside Provision, before it returns, so a test can observe // what the handler published for the window in which the call is still in flight. provisionHook func() + // provisionBlocksUntilDeadline makes Provision consume its whole ctx budget before + // returning successfully, the way a real backend that answers just under the + // timeout does. It exists to prove the provision deadline does not leak into the + // writes that follow. + provisionBlocksUntilDeadline bool } func (f *fakeProvider) Name() string { return "fake" } func (f *fakeProvider) Capabilities() provider.Capabilities { return f.capabilities } func (f *fakeProvider) Provision( - _ context.Context, _ *corev1.Pod, req provider.ProvisionRequest, -) (string, bool, error) { + ctx context.Context, _ *corev1.Pod, req provider.ProvisionRequest, +) (provider.ProvisionResult, error) { // Outside the lock: the hook reads Handler state, and holding f.mu here would // deadlock a hook that touches the provider. if f.provisionHook != nil { f.provisionHook() } + if f.provisionBlocksUntilDeadline { + <-ctx.Done() // burn the provision budget, then succeed anyway + } f.mu.Lock() defer f.mu.Unlock() f.provisionCnt++ f.lastReq = req - return f.provisionID, f.provisionReserved, f.provisionErr + if f.provisionErr != nil { + return provider.ProvisionResult{}, f.provisionErr + } + return provider.ProvisionResult{ + InstanceID: f.provisionID, + Reserved: f.provisionReserved, + ConnectURL: f.provisionURL, + ConnectToken: f.provisionToken, + }, nil } func (f *fakeProvider) Terminate(_ context.Context, id string) error { @@ -572,6 +595,409 @@ func TestNotify_PersistsEndpointAnnotationOnce(t *testing.T) { } } +// connectSecret fetches the connect Secret for a pod, or nil when absent. +func connectSecret(t *testing.T, client *fake.Clientset, ns, podName string) *corev1.Secret { + t.Helper() + s, err := client.CoreV1().Secrets(ns).Get(context.Background(), ConnectSecretName(podName), metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + t.Fatalf("get connect secret: %v", err) + } + return s +} + +// secretValue reads a key from either half of a Secret. The API server folds +// StringData into Data on write, but the fake client stores it verbatim, so a test +// asserting on Data alone would only be exercising the fake. +func secretValue(s *corev1.Secret, key string) string { + if v, ok := s.StringData[key]; ok { + return v + } + return string(s.Data[key]) +} + +// The credential must land in a Secret and NOT on the Pod: an annotation is +// readable by anyone with `get pod` in the namespace and sits unencrypted in etcd. +// The URL rides along so the pair is usable from one object, and separately goes on +// the Pod's endpoint annotation, which is not secret. +func TestCreatePod_WritesConnectSecret(t *testing.T) { + const url, token = "https://sb-1.modal.host", "tok-abc" + pod := testPod("default", "p1") + pod.UID = "uid-1" + client := fake.NewSimpleClientset(pod) + + fp := &fakeProvider{provisionID: "inst-1", provisionURL: url, provisionToken: token} + h := NewHandler(fp, client, nil) + // Register the notifier BEFORE CreatePod, as VK does (it wires NotifyPods before + // any pod work starts). The endpoint reaches the API server through it. + h.NotifyPods(context.Background(), func(*corev1.Pod) {}) + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + got := connectSecret(t, client, "default", "p1") + if got == nil { + t.Fatal("expected a connect Secret from the credential Provision returned") + } + if v := secretValue(got, "token"); v != token { + t.Fatalf("secret token = %q, want %q", v, token) + } + if v := secretValue(got, "url"); v != url { + t.Fatalf("secret url = %q, want %q", v, url) + } + // ownerReferenced to the Pod BY UID — a name alone is ignored by the GC — so + // teardown of the Pod reclaims the Secret on every path, including a force delete. + if len(got.OwnerReferences) != 1 { + t.Fatalf("ownerReferences = %v, want exactly one (the Pod)", got.OwnerReferences) + } + if ref := got.OwnerReferences[0]; ref.Kind != "Pod" || ref.Name != "p1" || ref.UID != "uid-1" { + t.Fatalf("ownerReference = %+v, want Pod/p1 with the Pod's UID", ref) + } + + live, err := client.CoreV1().Pods("default").Get(context.Background(), "p1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get pod: %v", err) + } + // The address IS published on the Pod — that is how a consumer finds it, and how + // the Sandbox controller reports Status.Endpoint. + if got := live.Annotations[nebulav1alpha1.EndpointAnnotation]; got != url { + t.Fatalf("endpoint annotation = %q, want the connect URL %q", got, url) + } + // The token must never reach the Pod itself. + for k, v := range live.Annotations { + if strings.Contains(v, token) { + t.Fatalf("token leaked onto Pod annotation %q", k) + } + } +} + +// The Secret is written ONCE, on the create path — not per poll tick. This is the +// difference between one write per workload and one per workload per tick forever +// (at 10k pods on a 15s cadence, ~666 writes/sec that can only ever say +// AlreadyExists). It is also the only place it CAN be written: the token is minted +// once and cannot be re-read, so no later tick has anything to write. +func TestConnectSecret_WrittenOnceNotPerTick(t *testing.T) { + pod := testPod("default", "p1") + pod.UID = "uid-1" + client := fake.NewSimpleClientset(pod) + + var creates int + client.PrependReactor("create", "secrets", func(k8stesting.Action) (bool, runtime.Object, error) { + creates++ + return false, nil, nil // fall through to the tracker + }) + + fp := &fakeProvider{ + provisionID: "inst-1", + provisionURL: "https://sb-1.modal.host", + provisionToken: "tok-abc", + } + h := NewHandler(fp, client, nil) + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + if creates != 1 { + t.Fatalf("CreatePod issued %d secret creates, want exactly 1", creates) + } + h.NotifyPods(context.Background(), func(*corev1.Pod) {}) + + fp.list = []provider.Instance{{ + ID: "inst-1", ClaimName: "default-p1", State: provider.InstanceRunning, + }} + for i := 0; i < 3; i++ { + h.reconcileOnce(context.Background()) + } + if creates != 1 { + t.Fatalf("the poll loop touched the Secret: %d creates after 3 ticks, want 1", creates) + } + if got := connectSecret(t, client, "default", "p1"); got == nil || + secretValue(got, "token") != "tok-abc" { + t.Fatalf("secret = %v, want the original credential intact", got) + } +} + +// An endpoint-less workload (a training job, a batch script) gets no credential from +// the provider, and so must get no Secret: creating an empty one would imply a +// reachable surface that does not exist. This is also every AWS instance, whose +// address is observed later rather than minted here. +func TestCreatePod_NoSecretWithoutCredential(t *testing.T) { + pod := testPod("default", "p1") + pod.UID = "uid-1" + client := fake.NewSimpleClientset(pod) + + fp := &fakeProvider{provisionID: "inst-1"} // no URL, no token + h := NewHandler(fp, client, nil) + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + if got := connectSecret(t, client, "default", "p1"); got != nil { + t.Fatalf("expected no connect Secret for a credential-less instance, got %+v", got) + } +} + +// A URL with no token still publishes the address: the endpoint is how the workload +// is found, and a Secret holding only a URL would imply a credential that does not +// exist. +func TestCreatePod_URLWithoutTokenPatchesEndpointOnly(t *testing.T) { + const url = "https://sb-1.modal.host" + pod := testPod("default", "p1") + pod.UID = "uid-1" + client := fake.NewSimpleClientset(pod) + + fp := &fakeProvider{provisionID: "inst-1", provisionURL: url} + h := NewHandler(fp, client, nil) + h.NotifyPods(context.Background(), func(*corev1.Pod) {}) + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + live, err := client.CoreV1().Pods("default").Get(context.Background(), "p1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get pod: %v", err) + } + if got := live.Annotations[nebulav1alpha1.EndpointAnnotation]; got != url { + t.Fatalf("endpoint annotation = %q, want %q", got, url) + } + if got := connectSecret(t, client, "default", "p1"); got != nil { + t.Fatalf("expected no Secret without a token, got %+v", got) + } +} + +// A UID-less Pod would get an ownerReference the GC cannot resolve, so the Secret +// would never be collected — leaking a live credential. Skip instead. +func TestCreateConnectSecret_SkipsUIDLessPod(t *testing.T) { + client := fake.NewSimpleClientset() + h := NewHandler(&fakeProvider{}, client, nil) + + h.createConnectSecret(context.Background(), testPod("default", "p1"), // no UID + "https://sb-9.modal.host", "tok-abc") + + if got := connectSecret(t, client, "default", "p1"); got != nil { + t.Fatalf("expected no Secret for a UID-less Pod (it would never be GC'd), got %+v", got) + } +} + +// A failed Provision publishes nothing. There is no credential to publish, and an +// endpoint annotation would advertise an instance that does not exist. +func TestCreatePod_NoCredentialPersistedOnProvisionFailure(t *testing.T) { + pod := testPod("default", "p1") + pod.UID = "uid-1" + client := fake.NewSimpleClientset(pod) + + fp := &fakeProvider{provisionErr: errors.New("no capacity")} + h := NewHandler(fp, client, nil) + if err := h.CreatePod(context.Background(), pod); err == nil { + t.Fatal("expected CreatePod to fail") + } + + if got := connectSecret(t, client, "default", "p1"); got != nil { + t.Fatalf("expected no Secret after a failed Provision, got %+v", got) + } + live, err := client.CoreV1().Pods("default").Get(context.Background(), "p1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get pod: %v", err) + } + if got := live.Annotations[nebulav1alpha1.EndpointAnnotation]; got != "" { + t.Fatalf("endpoint annotation = %q, want empty after a failed Provision", got) + } +} + +// The endpoint annotation is where the address LIVES, so a restart does not lose it: +// nothing on the read path clears it, and a provider that reports no observed +// endpoint (Modal, which published at create) leaves the stored value alone. +func TestReconcileOnce_EmptyObservedEndpointDoesNotClearAnnotation(t *testing.T) { + const url = "https://sb-1.modal.host" + pod := testPod("default", "p1") + pod.UID = "uid-1" + client := fake.NewSimpleClientset(pod) + + fp := &fakeProvider{provisionID: "inst-1", provisionURL: url, provisionToken: "tok-abc"} + h := NewHandler(fp, client, nil) + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + h.NotifyPods(context.Background(), func(*corev1.Pod) {}) + + // The instance is observed with NO endpoint, the way Modal reports it. + fp.list = []provider.Instance{{ + ID: "inst-1", ClaimName: "default-p1", State: provider.InstanceRunning, + }} + h.reconcileOnce(context.Background()) + + live, err := client.CoreV1().Pods("default").Get(context.Background(), "p1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get pod: %v", err) + } + if got := live.Annotations[nebulav1alpha1.EndpointAnnotation]; got != url { + t.Fatalf("endpoint annotation = %q, want %q retained across a tick that observed none", got, url) + } +} + +// A create-time URL comes from the provider ONCE and is never re-observed (Modal +// reports no endpoint on the read path), so a failed patch cannot be recovered from the +// provider. It is recovered from the tracked Pod instead: CreatePod stamps the address +// there, and every poll tick re-emits it, so persistEndpoint keeps patching until one +// write lands — then dedups. Without that, one transient 500 at create leaves the +// workload permanently unreachable. +func TestCreatePod_FailedEndpointPatchIsRetriedByPollLoop(t *testing.T) { + const url = "https://sb-1.modal.host" + pod := testPod("default", "p1") + pod.UID = "uid-1" + client := fake.NewSimpleClientset(pod) + + var patches int + failing := true + client.PrependReactor("patch", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { + patches++ + if failing { + return true, nil, errors.New("boom") + } + return false, nil, nil // fall through to the tracker + }) + + fp := &fakeProvider{provisionID: "inst-1", provisionURL: url, provisionToken: "tok-abc"} + h := NewHandler(fp, client, nil) + h.NotifyPods(context.Background(), func(*corev1.Pod) {}) + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod must not fail on a failed endpoint patch: %v", err) + } + if patches != 1 { + t.Fatalf("expected CreatePod's emit to attempt one patch, got %d", patches) + } + live, err := client.CoreV1().Pods("default").Get(context.Background(), "p1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get pod: %v", err) + } + if got := live.Annotations[nebulav1alpha1.EndpointAnnotation]; got != "" { + t.Fatalf("precondition: the patch was supposed to fail, but the annotation is %q", got) + } + + // Modal's read path reports no endpoint, so the retry can only come from what was + // stamped at create. + fp.list = []provider.Instance{{ + ID: "inst-1", ClaimName: "default-p1", State: provider.InstanceRunning, + }} + + // Still failing: the tick retries rather than giving up. + h.reconcileOnce(context.Background()) + if patches != 2 { + t.Fatalf("expected the poll tick to retry the patch, got %d patches total", patches) + } + + failing = false + h.reconcileOnce(context.Background()) + if patches != 3 { + t.Fatalf("expected a third patch attempt, got %d", patches) + } + live, err = client.CoreV1().Pods("default").Get(context.Background(), "p1", metav1.GetOptions{}) + if err != nil { + t.Fatalf("get pod: %v", err) + } + if got := live.Annotations[nebulav1alpha1.EndpointAnnotation]; got != url { + t.Fatalf("endpoint annotation = %q, want the create-time URL %q recovered", got, url) + } + + // And once it lands, it dedups — the retry must not become a per-tick write. + h.reconcileOnce(context.Background()) + if patches != 3 { + t.Fatalf("a landed endpoint must not re-patch; got %d patches", patches) + } +} + +// ctxRecordingClient wraps a clientset to capture the ctx a Secret create receives. +// The fake clientset ignores ctx entirely, so asserting the Secret merely EXISTS would +// pass whether or not the provision deadline leaked; the ctx itself has to be inspected. +// Embedding keeps this to the one method under test. +type ctxRecordingClient struct { + kubernetes.Interface + seen *context.Context +} + +func (c ctxRecordingClient) CoreV1() corev1client.CoreV1Interface { + return ctxRecordingCoreV1{c.Interface.CoreV1(), c.seen} +} + +type ctxRecordingCoreV1 struct { + corev1client.CoreV1Interface + seen *context.Context +} + +func (c ctxRecordingCoreV1) Secrets(ns string) corev1client.SecretInterface { + return ctxRecordingSecrets{c.CoreV1Interface.Secrets(ns), c.seen} +} + +type ctxRecordingSecrets struct { + corev1client.SecretInterface + seen *context.Context +} + +func (s ctxRecordingSecrets) Create( + ctx context.Context, secret *corev1.Secret, opts metav1.CreateOptions, +) (*corev1.Secret, error) { + *s.seen = ctx + return s.SecretInterface.Create(ctx, secret, opts) +} + +// The provision timeout bounds the Provision CALL, not the writes that follow it. A +// backend that answers just under the deadline would otherwise hand those writes a ctx +// with no budget left, so they would fail with DeadlineExceeded on the SUCCESS path — +// and the Secret is never retried, so the one-shot token would be lost for good. +// +// Only the Secret is at risk: the endpoint patch runs on the notify wrapper's own +// long-lived ctx (see NotifyPods), not on CreatePod's. +func TestCreatePod_ProvisionDeadlineDoesNotLeakIntoCredentialWrite(t *testing.T) { + const url, token = "https://sb-1.modal.host", "tok-abc" + pod := testPod("default", "p1") + pod.UID = "uid-1" + client := fake.NewSimpleClientset(pod) + + var secretCtx context.Context + fp := &fakeProvider{ + provisionID: "inst-1", + provisionURL: url, + provisionToken: token, + provisionBlocksUntilDeadline: true, + // A tiny budget the call is guaranteed to exhaust. + capabilities: provider.Capabilities{ProvisionTimeout: time.Millisecond}, + } + h := NewHandler(fp, ctxRecordingClient{client, &secretCtx}, nil) + h.NotifyPods(context.Background(), func(*corev1.Pod) {}) + if err := h.CreatePod(context.Background(), pod); err != nil { + t.Fatalf("CreatePod: %v", err) + } + + if secretCtx == nil { + t.Fatal("the Secret write never ran") + } + if err := secretCtx.Err(); err != nil { + t.Fatalf("the Secret write got an already-expired ctx (%v): the provision deadline "+ + "leaked into it, and this token can never be re-minted", err) + } + if got := connectSecret(t, client, "default", "p1"); got == nil || + secretValue(got, "token") != token { + t.Fatalf("connect Secret = %v, want the token %q", got, token) + } +} + +// setEndpoint ignores an empty endpoint rather than clearing it. That is what lets the + +// setEndpoint ignores an empty endpoint rather than clearing it. That is what lets the +// create-time and observed paths coexist: Modal publishes its URL at create and then +// reports no endpoint at all, which must not erase the address. +func TestSetEndpoint_EmptyValueDoesNotClear(t *testing.T) { + pod := testPod("default", "p1") + pod.Annotations = map[string]string{nebulav1alpha1.EndpointAnnotation: "https://sb-1.modal.host"} + + setEndpoint(pod, "") + + if got := pod.Annotations[nebulav1alpha1.EndpointAnnotation]; got != "https://sb-1.modal.host" { + t.Fatalf("endpoint = %q, want the previous value retained", got) + } +} + func TestReconcileOnce_NotifiesOnProvisioningToInitializing(t *testing.T) { // The Pod starts at "Provisioning" (phase Pending), and the instance comes up // but has not yet passed its readiness checks => InstancePending, which maps to diff --git a/pkg/vnode/status.go b/pkg/vnode/status.go index 4adf782..9e61636 100644 --- a/pkg/vnode/status.go +++ b/pkg/vnode/status.go @@ -72,7 +72,7 @@ func applyState(pod *corev1.Pod, state provider.InstanceState, endpoint string, // when the endpoint actually is one — an AWS public DNS name (the common // case) would make the whole status UpdateStatus fail with a 422 and strand // the Pod on its prior (Initializing) status forever. The endpoint is always - // surfaced on EndpointAnnotation regardless of form (see Handler.applyEndpoint), + // surfaced on EndpointAnnotation regardless of form (see Handler.patchEndpoint), // so a DNS-only instance still exposes its reachable address; PodIP just stays // empty in that case. if endpoint != "" && net.ParseIP(endpoint) != nil {