From e429bbeb4e1dae8044943abeb9566bc1a567342a Mon Sep 17 00:00:00 2001 From: kerthcet Date: Sat, 29 Aug 2026 10:43:46 +0100 Subject: [PATCH] support runpod Signed-off-by: kerthcet --- .env.example | 15 +- README.md | 6 +- cmd/main.go | 14 + config/catalog/kustomization.yaml | 1 + config/manager/manager.yaml | 12 +- config/samples/nodepool.yaml | 14 +- docs/deploy.md | 1 + docs/status.md | 48 +- hack/deploy.sh | 7 +- pkg/provider/catalog/data/runpod.csv | 69 +++ pkg/provider/provider.go | 19 +- pkg/provider/runpod/client.go | 544 +++++++++++++++++++ pkg/provider/runpod/client_test.go | 487 +++++++++++++++++ pkg/provider/runpod/runpod.go | 738 ++++++++++++++++++++++++++ pkg/provider/runpod/runpod_test.go | 750 +++++++++++++++++++++++++++ 15 files changed, 2708 insertions(+), 17 deletions(-) create mode 100644 pkg/provider/catalog/data/runpod.csv create mode 100644 pkg/provider/runpod/client.go create mode 100644 pkg/provider/runpod/client_test.go create mode 100644 pkg/provider/runpod/runpod.go create mode 100644 pkg/provider/runpod/runpod_test.go diff --git a/.env.example b/.env.example index 7a8bcb0..6ef70d0 100644 --- a/.env.example +++ b/.env.example @@ -45,6 +45,17 @@ MODAL_ENVIRONMENT= AWS_ACCESS_KEY_ID= AWS_SECRET_ACCESS_KEY= +# --- RunPod provider ------------------------------------------------------- +# One API key, from https://console.runpod.io/user/settings. It needs READ+WRITE on +# Pods: Nebula creates and deletes them, and also creates container-registry-auth +# objects when a workload uses an imagePullSecret. A read-only key registers fine +# and then fails every provision with an auth error, which blocklists the whole +# provider until it is replaced. +# +# Unlike AWS there is no ambient identity to fall back on, so leaving this blank +# skips both the Secret and the provider registration — not fatal. +RUNPOD_API_KEY= + # --- Additional providers (add as adapters land) --------------------------- -# Each provider gets its OWN secret (see hack/deploy.sh PROVIDER_SECRETS), e.g.: -# RUNPOD_API_KEY= +# Each provider gets its OWN secret; see hack/deploy.sh PROVIDER_SECRETS and the +# RunPod block above for the shape. diff --git a/README.md b/README.md index a1a0556..642ac22 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,9 @@ metadata: spec: providers: - name: modal # NeoCloud; regions omitted = place anywhere (cheapest) + - name: runpod # NeoCloud with real Spot; a region is a country code + regions: # ("us") or one data center ("us-ks-2") + - us - name: aws # hyperscaler; "us" expands to every US region regions: - us @@ -95,7 +98,8 @@ placement controller owns those. > `kubectl logs` and `kubectl exec` both work on Modal, `-f`/`--tail` and `-it` > included: the manager serves the two kubelet routes the API server proxies. > `--timestamps`/`--previous`/`--since` and `-c` are ignored, and a terminal resize is -> not forwarded. On providers that do not support them yet, both answer NotFound. +> not forwarded. On providers that do not support them yet — AWS and RunPod — both +> answer NotFound. ## Getting started diff --git a/cmd/main.go b/cmd/main.go index c964697..3c1226f 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -53,6 +53,7 @@ import ( awsprovider "github.com/InftyAI/Nebula/pkg/provider/aws" "github.com/InftyAI/Nebula/pkg/provider/fake" "github.com/InftyAI/Nebula/pkg/provider/modal" + "github.com/InftyAI/Nebula/pkg/provider/runpod" "github.com/InftyAI/Nebula/pkg/vnode" // +kubebuilder:scaffold:imports ) @@ -505,6 +506,19 @@ func registerProviders(ctx context.Context, c client.Client) { setupLog.Info("registered provider", "provider", p.Name()) } + // RunPod. Like Modal and unlike AWS, its credential is a single API key read from the + // environment (RUNPOD_API_KEY, delivered by the per-provider Secret) — there is no + // role/instance-identity path to fall back on, so an absent key is exactly the + // logged-and-skipped case. Also like Modal, there is no region config here: a pool's + // regions become RunPod data centers or country codes at provision time, so editing a + // NodePool changes placement without a restart. + if p, err := runpod.NewSDKClient(ctx); err != nil { + setupLog.Info("skipping RunPod provider registration", "reason", err.Error()) + } else { + provider.Register(p) + setupLog.Info("registered provider", "provider", p.Name()) + } + // The fake provider is an in-memory backend used only by the e2e suite to // exercise the full control-plane loop without cloud credentials. It ships in // the binary but registers ONLY when explicitly enabled, so it can never place diff --git a/config/catalog/kustomization.yaml b/config/catalog/kustomization.yaml index 39d2f64..2862055 100644 --- a/config/catalog/kustomization.yaml +++ b/config/catalog/kustomization.yaml @@ -17,6 +17,7 @@ configMapGenerator: files: - modal.csv=../../pkg/provider/catalog/data/modal.csv - aws.csv=../../pkg/provider/catalog/data/aws.csv + - runpod.csv=../../pkg/provider/catalog/data/runpod.csv generatorOptions: # Stable name (no content-hash suffix) so `kubectl edit` and the volume diff --git a/config/manager/manager.yaml b/config/manager/manager.yaml index 67a1928..3c1201b 100644 --- a/config/manager/manager.yaml +++ b/config/manager/manager.yaml @@ -116,10 +116,14 @@ spec: - secretRef: name: nebula-aws-credentials optional: true - # Add one secretRef per provider as adapters land, e.g.: - # - secretRef: - # name: nebula-runpod-credentials - # optional: true + # RunPod: a single API key (RUNPOD_API_KEY). Unlike AWS there is no ambient + # identity to fall back on, so an absent Secret means the provider is simply + # skipped at registration. Regions come from the NodePool, so this is the only + # RunPod config here. + - secretRef: + name: nebula-runpod-credentials + optional: true + # Add one secretRef per provider as adapters land, following the pattern above. ports: # The kubelet API the API server dials for `kubectl logs` (10250, like a real # kubelet). Declaring it is documentation and NetworkPolicy surface; the diff --git a/config/samples/nodepool.yaml b/config/samples/nodepool.yaml index f68529f..2f21acf 100644 --- a/config/samples/nodepool.yaml +++ b/config/samples/nodepool.yaml @@ -19,7 +19,14 @@ spec: - us - eu - ap-melbourne - # - name: runpod + - name: runpod + # RunPod takes a region as either an ISO country code (US, SE) or one exact data + # center (US-KS-2, EU-RO-1) — no group tokens: "eu" is not a country code, so it + # would be forwarded and rejected at provision time. Each entry is one failover + # candidate, so a shortage in one blocklists only that one. + regions: + - US + - EU-RO-1 # Outer axis: try OnDemand on every provider first, fall back to Spot. capacityTypes: - OnDemand @@ -32,6 +39,11 @@ spec: # sandbox stays reachable through its connect URL and token under every mode — it just # cannot call out. # + # Setting ANY mode other than Open narrows this pool to the providers that can enforce + # it: placement skips a provider whose Capabilities report SupportsEgressPolicy=false + # (RunPod, which exposes no outbound knob at all), rather than provisioning something + # with open internet access under a policy that says otherwise. + # # Blocked permits nothing: # egress: # mode: Blocked diff --git a/docs/deploy.md b/docs/deploy.md index 8d2d5ee..33cb866 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -115,6 +115,7 @@ itself at startup (see [Webhook TLS](#webhook-tls-no-cert-manager)). | `MODAL_ENVIRONMENT` | Modal | no | Modal Environment to create sandboxes in. Blank omits the key and the SDK uses the token profile's default. See [Modal Environments](#modal-environments). | | `AWS_ACCESS_KEY_ID` | AWS | dev only | Prefer IRSA / instance role in production and leave blank — the SDK's default credential chain finds the role. Set only for local/dev. | | `AWS_SECRET_ACCESS_KEY` | AWS | dev only | Pairs with `AWS_ACCESS_KEY_ID`; both required together or both blank. | +| `RUNPOD_API_KEY` | RunPod | yes | From [console.runpod.io/user/settings](https://console.runpod.io/user/settings). Needs **read+write on Pods**: a read-only key registers fine and then fails every provision with an auth error, which blocklists the whole provider. Unlike AWS there is no ambient identity, so blank skips RunPod entirely. | Non-secret config, passed as `make` variables: diff --git a/docs/status.md b/docs/status.md index 94d4c6d..5efe3b3 100644 --- a/docs/status.md +++ b/docs/status.md @@ -25,6 +25,7 @@ enters the system. - [Provider mappings](#provider-mappings) - [AWS](#aws) - [Modal](#modal) + - [RunPod](#runpod) - [fake](#fake) - [Logs and exec](#logs-and-exec) - [What is not observable](#what-is-not-observable) @@ -79,8 +80,9 @@ therefore emits without storing, and stores only once `Provision` returns. `Provision` returns `(id, reserved, error)`. `reserved` means the provider committed capacity, not merely accepted the request: AWS always does (`CreateFleet` with -`FleetTypeInstant` is synchronous), a fresh Modal sandbox never does (the GPU may -still be queued). Only a reserved instance advances to `Initializing`; an unreserved +`FleetTypeInstant` is synchronous), RunPod always does for the same reason (`POST +/pods` allocates a host before it answers, and a shortage comes back as an error), a +fresh Modal sandbox never does (the GPU may still be queued). Only a reserved instance advances to `Initializing`; an unreserved id holds at `Provisioning`, which is still exactly true — the id is real and must be reclaimed, but nothing is allocated. Either way the Pod is now tracked: `reserved` constrains what the status may claim, not what is owed, and says nothing about @@ -232,6 +234,48 @@ only two signals and has to record a third fact itself. first poll tick. An *adopted* sandbox has been observed, so a `running` one is known to be reserved. See below for why queued is not reported distinctly. +### RunPod + +One RunPod Pod per NodeClaim. The single signal is `desiredStatus`, and the name says +what the trap is: it is the state RunPod *intends*, not the one it has reached. + +| `desiredStatus` | `lastStartedAt` | `InstanceState` | Pod | +|---|---|---|---| +| `RUNNING` | set | `Running` | `Running` / `Ready=True` | +| `RUNNING` | empty | `Pending` | `Pending` / `Initializing` | +| `EXITED`, `TERMINATED` | — | `Terminated` | `Failed` / `Terminated` | +| anything else | — | `Pending` | `Pending` / `Initializing` | +| absent from `List` | — | `Terminated` | `Failed` / `Terminated` | + +- **`RUNNING` alone is not running.** RunPod reports it from the moment it accepts the + Pod, while the image may still be pulling. `lastStartedAt` is the one field that + appears only once the container has actually started, so it is the gate — the same + role AWS's 2/2 reachability checks play. Without it a Deployment's replica would read + ready before anything was listening. +- **There is no readiness concept beyond that.** RunPod has no probe, so "started" is + the strongest signal available; a container that is up but not yet serving reads + `Running`. Contrast Modal, which has a real probe and latches it. +- **There is no queueing**, as with AWS: `POST /pods` allocates a host machine before + it answers, and a capacity shortfall is a synchronous error + (`ErrNoCapacity`, plus `ErrSpotCapacity` on the interruptible tier) that drives + region/tier failover. So `Provision` always returns `reserved` and the Pod goes + straight to `Initializing`. +- **No `Failed` case.** `EXITED` covers a clean exit and a crash alike — RunPod does + not distinguish them here and exposes no exit code — so a workload that died reads + as `Terminated`, indistinguishable from teardown. A spot reclaim arrives the same + way (`TERMINATED`, no notice), which is why the poll interval is 10s. +- **Identity rides the Pod name**, not tags: RunPod Pods have none, so `List` filters + on the `nebula-` prefix and the claim name is recovered by stripping it. A Pod whose + name would exceed RunPod's 191-character cap is refused at `Provision` rather than + truncated — two truncated claims would collide onto one Pod. +- The endpoint is **derived, not read back**: `https://-.proxy.runpod.net` + is known at create time, so it is published from `CreatePod` like Modal's, but with + no token — that proxy is unauthenticated. A Pod with a public IP and an assigned + `/tcp` port mapping reports that direct address instead, once the poll loop sees it. +- **Neither `kubectl logs` nor `kubectl exec` works.** RunPod's REST v1 surface has no + pod-log endpoint and its only way into a container is SSH, so the adapter implements + neither optional half and both routes answer NotFound. + ### fake The in-memory e2e provider reports `InstanceRunning` as soon as an instance is diff --git a/hack/deploy.sh b/hack/deploy.sh index e79d45b..6c4014b 100755 --- a/hack/deploy.sh +++ b/hack/deploy.sh @@ -102,7 +102,12 @@ PROVIDER_SECRETS=( # instance role (the preferred path). Region is NON-SECRET (on the manager # Deployment); the adapter self-configures the rest (GPU AMI + subnets). "nebula-aws-credentials|AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY|" - # "nebula-runpod-credentials|RUNPOD_API_KEY|" + # RunPod: one API key, and the only credential it has — there is no ambient identity to + # fall back on as AWS has, so a blank key skips the Secret AND the provider. Mint it at + # https://console.runpod.io/user/settings with read+write on Pods; a read-only key + # registers fine and then fails every create with an auth error, which blocklists the + # whole provider. + "nebula-runpod-credentials|RUNPOD_API_KEY|" ) # create_provider_secret diff --git a/pkg/provider/catalog/data/runpod.csv b/pkg/provider/catalog/data/runpod.csv new file mode 100644 index 0000000..f8071aa --- /dev/null +++ b/pkg/provider/catalog/data/runpod.csv @@ -0,0 +1,69 @@ +# RunPod price/availability catalog — community-maintained. +# +# RunPod exposes GPU prices through its API, but only per GPU TYPE and without the +# canonical-name mapping Nebula needs, so these are SECURE-cloud list prices +# transcribed from https://runpod.io/pricing by hand. Treat them as a starting +# point, not a billing source, and refresh via `make update-catalog` +# (see hack/refresh.go). +# +# Prices are per GPU-HOUR, as Modal's are — not per instance-hour like aws.csv. +# RunPod bills per GPU, and the adapter passes the count as a runtime parameter. +# +# Only SECURE cloud is priced here. RunPod's COMMUNITY cloud rents the same GPUs +# from peer hosts for roughly 30-50% less, but this file has no cloud-type column +# to say which tier a row prices, so listing both would make the number the +# optimizer reads a coin flip. The adapter pins cloudType=SECURE to match; adding +# COMMUNITY means adding that column first (see pkg/provider/runpod's package doc). +# +# Columns (shared header across all provider CSVs; unused cells left blank): +# accelerator_type canonical Nebula accelerator type, matched case-insensitively +# against the nebula.inftyai.com/accelerator-type label +# accelerator_id ALWAYS SET here, unlike modal.csv: RunPod's ids are marketing +# strings ("NVIDIA H100 80GB HBM3") that share nothing with the +# canonical names, so every row carries its own translation. +# +# SEVERAL rows may share one accelerator_type, and the order +# matters: MapAccelerator returns them in file order, so the +# FIRST is the primary and the rest are interchangeable +# alternates. RunPod's create takes the whole list and picks by +# availability, so alternates widen a single launch — they never +# widen what a failure blocklists, which keys on the canonical +# pool. Put the variant with the best interconnect first (SXM +# before NVL before PCIe); RunPod falls back only if it must. +# gpu_count BLANK. RunPod takes the GPU count as a request parameter, so it +# is not a lookup dimension (contrast aws.csv, where the count is +# baked into the instance type). A blank row matches any count. +# capacity_type OnDemand | Spot. Spot is RunPod's `interruptible` tier: real, +# abruptly reclaimed, and with no bid to name in the REST v1 API. +# price_per_hour approximate USD per GPU-hour on SECURE cloud +# available whether Nebula may schedule onto it. Flipping a row to false +# removes it from placement everywhere without touching Go — the +# escape hatch for a GPU type RunPod has stopped offering. +# region BLANK. RunPod's prices are not partitioned by data center, so +# one row prices every region. Region is still a real placement +# axis for this provider (a pool's regions become dataCenterIds); +# it is just not a pricing one. +# updated documentation only, ignored by the parser +accelerator_type,accelerator_id,gpu_count,capacity_type,price_per_hour,available,region,updated +L4,NVIDIA L4,,OnDemand,0.43,true,,2026-08-29 +L4,NVIDIA L4,,Spot,0.25,true,,2026-08-29 +A40,NVIDIA A40,,OnDemand,0.40,true,,2026-08-29 +A40,NVIDIA A40,,Spot,0.23,true,,2026-08-29 +RTX4090,NVIDIA GeForce RTX 4090,,OnDemand,0.69,true,,2026-08-29 +RTX4090,NVIDIA GeForce RTX 4090,,Spot,0.35,true,,2026-08-29 +L40S,NVIDIA L40S,,OnDemand,0.86,true,,2026-08-29 +L40S,NVIDIA L40S,,Spot,0.49,true,,2026-08-29 +A100-40GB,NVIDIA A100-PCIE-40GB,,OnDemand,1.19,true,,2026-08-29 +A100-40GB,NVIDIA A100-PCIE-40GB,,Spot,0.69,true,,2026-08-29 +A100-80GB,NVIDIA A100-SXM4-80GB,,OnDemand,1.74,true,,2026-08-29 +A100-80GB,NVIDIA A100-SXM4-80GB,,Spot,0.99,true,,2026-08-29 +A100-80GB,NVIDIA A100 80GB PCIe,,OnDemand,1.64,true,,2026-08-29 +A100-80GB,NVIDIA A100 80GB PCIe,,Spot,0.94,true,,2026-08-29 +H100,NVIDIA H100 80GB HBM3,,OnDemand,2.99,true,,2026-08-29 +H100,NVIDIA H100 80GB HBM3,,Spot,1.65,true,,2026-08-29 +H100,NVIDIA H100 NVL,,OnDemand,2.79,true,,2026-08-29 +H100,NVIDIA H100 NVL,,Spot,1.55,true,,2026-08-29 +H100,NVIDIA H100 PCIe,,OnDemand,2.39,true,,2026-08-29 +H100,NVIDIA H100 PCIe,,Spot,1.35,true,,2026-08-29 +H200,NVIDIA H200,,OnDemand,3.99,true,,2026-08-29 +H200,NVIDIA H200,,Spot,2.19,true,,2026-08-29 diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go index ae6e52c..2041c03 100644 --- a/pkg/provider/provider.go +++ b/pkg/provider/provider.go @@ -414,9 +414,16 @@ type Offering struct { PricePerHour float64 Available bool // Region is the provider region this row prices, in the provider's own - // vocabulary (e.g. AWS "us-east-1"). Empty for region-simple providers whose - // catalog is not region-partitioned (Modal, RunPod); a region-aware provider - // emits one row per {accelerator, capacityType, region}. + // vocabulary (e.g. AWS "us-east-1"). Empty when a provider's catalog is not + // region-partitioned; a region-aware provider emits one row per {accelerator, + // capacityType, region}. + // + // Empty here is about PRICING, and says nothing about whether the provider has a + // region axis at all — the two are independent. Modal has neither. RunPod prices + // every data center alike, so its rows carry no region, yet region IS a real + // placement axis for it (a pool's regions become RunPod dataCenterIds). AWS's rows + // are blank for a third reason: its per-region truth is probed live rather than + // hand-maintained. Region string // AcceleratorID is this provider's own name for what serves the canonical // AcceleratorType (AWS "p5.48xlarge" for H100) — the lookup data MapAccelerator @@ -458,9 +465,9 @@ type BlockScope struct { Accelerator *string // CapacityType empty => blocks all capacity types. CapacityType nebulav1alpha1.CapacityType - // Region: nil => the provider has no region axis (Modal/RunPod, whose candidates - // carry an empty region too); &"us-east-1" => that region only, so a shortage there - // does not disqualify us-west-2. + // Region: nil => the provider has no region axis (Modal, whose candidates carry an + // empty region too); &"us-east-1" => that region only, so a shortage there does not + // disqualify us-west-2. Region *string // DenyAll true => block everything on this provider (auth/quota errors), ignoring the // fields above. Still scoped to this one provider; it never spans providers. diff --git a/pkg/provider/runpod/client.go b/pkg/provider/runpod/client.go new file mode 100644 index 0000000..0ec77d9 --- /dev/null +++ b/pkg/provider/runpod/client.go @@ -0,0 +1,544 @@ +/* +Copyright 2026 The InftyAI Team. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package runpod + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strings" + "time" + + "github.com/InftyAI/Nebula/pkg/provider" + "github.com/InftyAI/Nebula/pkg/provider/catalog" +) + +// The RunPod REST API. There is no official Go SDK, so this is plain net/http — which is +// also why the whole surface is one small file: the adapter needs five operations. +const ( + // defaultBaseURL is RunPod's REST v1 root. Overridable only in tests (see newClient). + defaultBaseURL = "https://rest.runpod.io/v1" + // apiKeyEnv is where the credential comes from. It is delivered by the per-provider + // Secret the manager mounts via envFrom; absent means the provider is skipped at + // registration rather than failing the process. + apiKeyEnv = "RUNPOD_API_KEY" + // requestTimeout bounds one HTTP call. Generous because a create allocates a machine + // server-side, but finite: a hung call would otherwise pin a Provision until the + // caller's own context expired. + requestTimeout = 60 * time.Second + // maxResponseBytes caps how much of a response is read, so a malformed or hostile + // response cannot exhaust memory. A full Pod list is a few KB per Pod. + maxResponseBytes = 8 << 20 + // maxErrorBodyChars caps how much of an error response reaches the error string, which + // is logged and may land on a Pod condition. + maxErrorBodyChars = 512 + // cloudTypeSecure is the only cloud type Nebula requests; see the package doc for why + // COMMUNITY is out until the catalog can price it. + cloudTypeSecure = "SECURE" +) + +// restClient is the real Client, backed by RunPod's REST API. Every RunPod-specific HTTP +// call lives here so the adapter and its tests stay transport-free. +type restClient struct { + http *http.Client + baseURL string + apiKey string +} + +// compile-time assertion that restClient satisfies the adapter's Client seam. +var _ Client = (*restClient)(nil) + +// NewSDKClient builds a RunPod-backed Provider, reading the API key from RUNPOD_API_KEY. +// An absent key is an ERROR rather than a client that fails on first use, so +// registerProviders can log and skip RunPod the same way it skips Modal and AWS. +// +// The context is accepted for symmetry with the other adapters' constructors (and so a +// future availability probe can use it); nothing here makes a call. +func NewSDKClient(_ context.Context) (*Provider, error) { + apiKey := strings.TrimSpace(os.Getenv(apiKeyEnv)) + if apiKey == "" { + return nil, fmt.Errorf("runpod: %s is not set", apiKeyEnv) + } + cat, err := catalog.Load() + if err != nil { + return nil, fmt.Errorf("runpod: load price catalog: %w", err) + } + return New(newClient(defaultBaseURL, apiKey), cat), nil +} + +// newClient builds a restClient against baseURL. Separate from NewSDKClient so a test can +// point it at an httptest.Server. +func newClient(baseURL, apiKey string) *restClient { + return &restClient{ + http: &http.Client{Timeout: requestTimeout}, + baseURL: strings.TrimSuffix(baseURL, "/"), + apiKey: apiKey, + } +} + +// apiError is one non-2xx RunPod response. It carries the status and RunPod's own message +// so the classify helpers can key on both, and so an operator reading a log sees what +// RunPod actually said. +// +// It deliberately holds NOTHING from the REQUEST body: that body carries the workload's +// resolved environment and, on a registry-auth create, a registry password. Only the method +// and path are echoed back. +type apiError struct { + status int + method string + path string + message string +} + +func (e *apiError) Error() string { + return fmt.Sprintf("runpod: %s %s: HTTP %d: %s", e.method, e.path, e.status, e.message) +} + +// notFound reports whether err is a 404. Both Get and Terminate treat that as "already +// gone" rather than a failure, which is what makes Terminate idempotent for the finalizer. +func notFound(err error) bool { + var ae *apiError + return errors.As(err, &ae) && ae.status == http.StatusNotFound +} + +// do performs one API call: body is JSON-encoded when non-nil, out is JSON-decoded when +// non-nil, and any non-2xx becomes an *apiError. +func (c *restClient) do(ctx context.Context, method, path string, body, out any) error { + var payload io.Reader + if body != nil { + encoded, err := json.Marshal(body) + if err != nil { + return fmt.Errorf("runpod: encode %s %s request: %w", method, path, err) + } + payload = bytes.NewReader(encoded) + } + req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, payload) + if err != nil { + return fmt.Errorf("runpod: build %s %s request: %w", method, path, err) + } + req.Header.Set("Authorization", "Bearer "+c.apiKey) + req.Header.Set("Accept", "application/json") + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + + resp, err := c.http.Do(req) + if err != nil { + // A transport failure, wrapped WITHOUT a sentinel on purpose: nobody knows whether + // RunPod acted on the request, and provider.IsRejection reads an unwrapped + // transport error as unattributable, which keeps the Pod retrying instead of being + // failed under an instance whose id we never saw. + return fmt.Errorf("runpod: %s %s: %w", method, path, err) + } + defer func() { _ = resp.Body.Close() }() + + raw, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes)) + if err != nil { + return fmt.Errorf("runpod: %s %s: read response: %w", method, path, err) + } + if resp.StatusCode >= http.StatusMultipleChoices { + return &apiError{ + status: resp.StatusCode, + method: method, + path: path, + message: errorMessage(raw), + } + } + if out == nil { + return nil + } + if err := json.Unmarshal(raw, out); err != nil { + return fmt.Errorf("runpod: %s %s: decode response: %w", method, path, err) + } + return nil +} + +// errorMessage pulls the human-readable part out of an error response, trying RunPod's +// JSON envelope first and falling back to the raw text — some gateway errors are HTML, and +// an empty message would leave the classifier nothing to read. +func errorMessage(raw []byte) string { + var envelope struct { + Error string `json:"error"` + Message string `json:"message"` + } + if err := json.Unmarshal(raw, &envelope); err == nil { + if envelope.Error != "" { + return truncate(envelope.Error) + } + if envelope.Message != "" { + return truncate(envelope.Message) + } + } + return truncate(strings.TrimSpace(string(raw))) +} + +// truncate bounds a message so an error string stays loggable. +func truncate(s string) string { + if len(s) <= maxErrorBodyChars { + return s + } + return s[:maxErrorBodyChars] + "…" +} + +// classifyCreate wraps a create failure with the shared sentinel that matches it, which is +// what lets the control plane act on the failure without knowing anything about RunPod (see +// docs/add-a-provider.md, "Wrap the errors your Provision returns"). Unwrapped, every one of +// these would land on nebula_provision_failures_total{reason="other"} and be retried +// forever against a provider that has already given its answer. +// +// interruptible is the request's tier, which the error itself never carries; a capacity +// failure on the spot tier also gets ErrSpotCapacity so ClassifyProvisionError can block +// Spot alone and leave OnDemand purchasable. +func classifyCreate(err error, interruptible bool) error { + var ae *apiError + if !errors.As(err, &ae) { + return err // a transport/encode failure: unattributable, and already wrapped + } + msg := strings.ToLower(ae.message) + + switch { + case ae.status == http.StatusUnauthorized, ae.status == http.StatusForbidden: + // Whole-provider: nothing succeeds until the key is fixed. + return fmt.Errorf("%w: %w", err, provider.ErrAuth) + + case ae.status >= http.StatusInternalServerError: + // Left UNWRAPPED, deliberately. A 5xx says RunPod failed to answer, not that it + // said no, and it may well have created the Pod before falling over. Wrapping a + // sentinel here would fail the Pod and blocklist a candidate on the strength of a + // server-side blip — and could reap a Pod out from under a paid instance. + return err + + case ae.status == http.StatusTooManyRequests: + return fmt.Errorf("%w: %w", err, provider.ErrQuota) + + // Money, not capacity, but scoped the same way: it is transient, it is not an + // authentication problem, and ErrQuota is the sentinel for "a limit stopped this". It + // does block more than the one candidate in practice, which the TTL bounds. + case ae.status == http.StatusPaymentRequired, + containsAny(msg, "insufficient funds", "insufficient balance", "not enough credit"): + return fmt.Errorf("%w: %w", err, provider.ErrQuota) + + case containsAny(msg, "no longer any instances available", "no instances available", + "no instance available", "out of capacity", "no capacity", "not available", + "unavailable", "sold out"): + if interruptible { + return fmt.Errorf("%w: %w: %w", err, provider.ErrNoCapacity, ErrSpotCapacity) + } + return fmt.Errorf("%w: %w", err, provider.ErrNoCapacity) + + case containsAny(msg, "invalid gpu", "unknown gpu", "gpu type", "unsupported"): + // A GPU id RunPod does not recognize: durable until runpod.csv is corrected, and + // accelerator-scoped so the rest of the provider stays usable. + return fmt.Errorf("%w: %w", err, provider.ErrUnsupportedAccelerator) + + case containsAny(msg, "registry", "image", "pull", "manifest"): + // Belongs to the REQUEST, not the candidate, so ErrImagePull — which blocklists + // NOTHING. Blocking here would exclude an accelerator that is serving every other + // Pod fine, because one Pod named an image RunPod could not fetch. + return fmt.Errorf("%w: %w", err, provider.ErrImagePull) + + default: + // An unrecognized 4xx. Left unwrapped rather than guessed at: none of the shared + // sentinels describes "RunPod rejected this and we do not know why", and every + // available guess is worse than retrying — ErrAuth would fence off the whole + // provider, a capacity wrap would evict a healthy candidate. A sustained + // reason="other" rate on this provider's metrics is the signal that a condition + // belongs in the table above. + return err + } +} + +// containsAny reports whether s contains any of subs. A local copy because the shared one +// in package provider is unexported, and the alternative — exporting it — would widen that +// package's surface for one adapter's convenience. +func containsAny(s string, subs ...string) bool { + for _, sub := range subs { + if strings.Contains(s, sub) { + return true + } + } + return false +} + +// createPodRequest is RunPod's POST /pods body. Only the fields Nebula sets are present; +// everything omitted takes RunPod's own default, which is the point of the omitempty tags — +// a zero we did not mean would override a sane default with 0. +type createPodRequest struct { + Name string `json:"name"` + ImageName string `json:"imageName"` + CloudType string `json:"cloudType"` + ComputeType string `json:"computeType"` + + GPUTypeIDs []string `json:"gpuTypeIds,omitempty"` + GPUCount int32 `json:"gpuCount,omitempty"` + GPUTypePriority string `json:"gpuTypePriority,omitempty"` + MinVCPUPerGPU int `json:"minVCPUPerGPU,omitempty"` + MinRAMPerGPU int `json:"minRAMPerGPU,omitempty"` + VCPUCount int `json:"vcpuCount,omitempty"` + + ContainerDiskInGb int `json:"containerDiskInGb,omitempty"` + // VolumeInGb has NO omitempty: 0 is exactly the value we mean, and it must be sent to + // override RunPod's default of a 20 GiB persistent volume. A Nebula instance is cattle + // with nothing to persist, so that volume would be pure cost. + VolumeInGb int `json:"volumeInGb"` + + Env map[string]string `json:"env,omitempty"` + DockerEntrypoint []string `json:"dockerEntrypoint,omitempty"` + DockerStartCmd []string `json:"dockerStartCmd,omitempty"` + Ports []string `json:"ports,omitempty"` + + DataCenterIDs []string `json:"dataCenterIds,omitempty"` + DataCenterPriority string `json:"dataCenterPriority,omitempty"` + CountryCodes []string `json:"countryCodes,omitempty"` + + // Interruptible has no omitempty either: false is the OnDemand tier, and being explicit + // about the tier that costs money is worth four bytes. + Interruptible bool `json:"interruptible"` + SupportPublicIP bool `json:"supportPublicIp,omitempty"` + RegistryAuthID string `json:"containerRegistryAuthId,omitempty"` +} + +// podResponse is the subset of RunPod's Pod object this adapter reads. Fields it ignores +// (savings plans, template, network volume, cost) are omitted rather than carried, so the +// struct states exactly what the adapter's behaviour depends on. +type podResponse struct { + ID string `json:"id"` + Name string `json:"name"` + DesiredStatus string `json:"desiredStatus"` + LastStartedAt string `json:"lastStartedAt"` + Interruptible bool `json:"interruptible"` + Ports []string `json:"ports"` + PortMappings map[string]int `json:"portMappings"` + PublicIP string `json:"publicIp"` + // Machine carries the data center, and is only populated when the request asks for it + // (includeMachine=true) — hence the query on every read path. A pointer because RunPod + // sends null when it is not included, and Region must then stay empty rather than + // reporting a placement we did not observe. + Machine *struct { + DataCenterID string `json:"dataCenterId"` + } `json:"machine"` +} + +// toPod converts the wire shape into the adapter's view. +func (r podResponse) toPod() Pod { + pd := Pod{ + ID: r.ID, + Name: r.Name, + DesiredStatus: r.DesiredStatus, + LastStartedAt: r.LastStartedAt, + Interruptible: r.Interruptible, + Ports: r.Ports, + PublicIP: r.PublicIP, + PortMappings: r.PortMappings, + } + if r.Machine != nil { + pd.DataCenterID = r.Machine.DataCenterID + } + return pd +} + +// CreatePod implements Client. +func (c *restClient) CreatePod(ctx context.Context, spec PodSpec) (string, error) { + body := createPodRequest{ + Name: spec.Name, + ImageName: spec.Image, + CloudType: cloudTypeSecure, + ComputeType: "GPU", + + GPUTypeIDs: spec.GPUTypeIDs, + GPUCount: spec.GPUCount, + MinVCPUPerGPU: spec.VCPUPerGPU, + MinRAMPerGPU: spec.RAMPerGPUGiB, + + ContainerDiskInGb: spec.ContainerDiskGiB, + VolumeInGb: 0, + + Env: spec.Env, + DockerEntrypoint: spec.Entrypoint, + DockerStartCmd: spec.StartCmd, + Ports: spec.Ports, + + DataCenterIDs: spec.DataCenterIDs, + CountryCodes: spec.CountryCodes, + + Interruptible: spec.Interruptible, + // Ask for a public IP so a /tcp port can be reached directly. Harmless for the + // /http-only Pods this adapter creates today, and it is what would make a raw-TCP + // workload addressable without a second change here. + SupportPublicIP: true, + RegistryAuthID: spec.RegistryAuthID, + } + if spec.GPUCount == 0 { + // A CPU-only Pod is a different product on RunPod: the GPU sizing fields are + // meaningless and vcpuCount replaces them. + body.ComputeType = "CPU" + body.VCPUCount = spec.VCPUCount + } else if len(spec.GPUTypeIDs) > 1 { + // Only meaningful with alternates: it tells RunPod to satisfy the list by whichever + // id has capacity rather than insisting on the first. With one id there is nothing + // to prioritize, and sending it would just be noise in the request. + body.GPUTypePriority = "availability" + } + if len(spec.DataCenterIDs) > 0 { + // custom, not availability: the pool NAMED these data centers, so honouring them is + // the point. availability would let RunPod place elsewhere, silently breaking a + // constraint an operator set for data residency. + body.DataCenterPriority = "custom" + } + + var out podResponse + if err := c.do(ctx, http.MethodPost, "/pods", body, &out); err != nil { + return "", classifyCreate(err, spec.Interruptible) + } + if out.ID == "" { + // A 2xx with no id is unusable and, worse, ambiguous: a Pod may exist that we can + // never name to terminate. Reported as an error with no sentinel, so the Pod retries + // (Provision is idempotent on the claim name, and the name lookup will find any Pod + // this call did create). + return "", fmt.Errorf("runpod: create pod %q: response carried no id", spec.Name) + } + return out.ID, nil +} + +// TerminatePod implements Client. Idempotent: a 404 means the Pod is already gone, which is +// success for the caller (the NodeClaim finalizer retries against this). +func (c *restClient) TerminatePod(ctx context.Context, id string) error { + err := c.do(ctx, http.MethodDelete, "/pods/"+url.PathEscape(id), nil, nil) + if err != nil && !notFound(err) { + return err + } + return nil +} + +// GetPod implements Client, returning (nil, nil) for a Pod that no longer exists. +func (c *restClient) GetPod(ctx context.Context, id string) (*Pod, error) { + var out podResponse + err := c.do(ctx, http.MethodGet, "/pods/"+url.PathEscape(id)+"?includeMachine=true", nil, &out) + if notFound(err) { + return nil, nil + } + if err != nil { + return nil, err + } + pd := out.toPod() + return &pd, nil +} + +// ListPods implements Client: one call for the whole account. Filtering to Nebula's own Pods +// is the adapter's job, since it owns the naming scheme (see Provider.List). +func (c *restClient) ListPods(ctx context.Context) ([]Pod, error) { + var out []podResponse + if err := c.do(ctx, http.MethodGet, "/pods?includeMachine=true", nil, &out); err != nil { + return nil, err + } + pods := make([]Pod, 0, len(out)) + for _, r := range out { + pods = append(pods, r.toPod()) + } + return pods, nil +} + +// registryAuthPath is the collection RunPod stores image-pull credentials in. +const registryAuthPath = "/containerregistryauth" + +// registryAuthResponse is the subset of a containerRegistryAuth object this adapter reads. +// Notably NOT the password: RunPod does not return it, and nothing here needs it back. +type registryAuthResponse struct { + ID string `json:"id"` + Name string `json:"name"` +} + +// EnsureRegistryAuth implements Client. RunPod's create takes a containerRegistryAuth ID, +// never an inline username/password, so a credential has to become an OBJECT in RunPod's +// account before a Pod can use it. +// +// The object is CONTENT-ADDRESSED — its name is a hash of the credential (see +// registryAuthName) — which is what makes this safe to call on every Provision: +// +// - Idempotent. The same credential resolves to the same name, so the list-then-create +// finds the existing object instead of accumulating one object per Pod. +// - Correct across rotation. A changed password hashes differently, so it becomes a new +// object rather than silently reusing a stale one that would 401 at pull time. +// +// Objects are never DELETED, and that is deliberate: one object is shared by every Pod using +// that credential, so deleting it on any single teardown would break the others' next pull. +// The population is bounded by the number of distinct credentials, not by the number of Pods. +func (c *restClient) EnsureRegistryAuth(ctx context.Context, auth *provider.RegistryAuth) (string, error) { + if auth == nil || auth.Basic == nil { + // The adapter vets the kind before calling (checkRegistryAuth), so this is a + // programming error rather than a user-facing one — but it must not become a silent + // anonymous pull. + return "", fmt.Errorf("runpod: registry auth %s cannot be expressed as a RunPod credential: %w", + auth, provider.ErrImagePull) + } + name := registryAuthName(auth.Basic.Username, auth.Basic.Password) + + var existing []registryAuthResponse + if err := c.do(ctx, http.MethodGet, registryAuthPath, nil, &existing); err != nil { + return "", err + } + for _, e := range existing { + if e.Name == name && e.ID != "" { + return e.ID, nil + } + } + + body := struct { + Name string `json:"name"` + Username string `json:"username"` + Password string `json:"password"` + }{Name: name, Username: auth.Basic.Username, Password: auth.Basic.Password} + + var created registryAuthResponse + if err := c.do(ctx, http.MethodPost, registryAuthPath, body, &created); err != nil { + // Wrapped as an image-pull failure, which blocklists nothing: a credential RunPod + // would not store is a fact about this Pod's imagePullSecret, not about the + // accelerator or region the Pod was headed for. + return "", fmt.Errorf("%w: %w", err, provider.ErrImagePull) + } + if created.ID == "" { + return "", fmt.Errorf("runpod: create registry auth %q: response carried no id: %w", + name, provider.ErrImagePull) + } + return created.ID, nil +} + +// registryAuthName is the content-addressed name of a stored credential: a fixed prefix +// (so Nebula's objects are recognizable in the RunPod console) plus a hash of the +// credential itself. +// +// The hash is what makes EnsureRegistryAuth idempotent, and it is over BOTH fields so a +// rotated password yields a new object. Hashed rather than named after the registry or the +// claim for two reasons: a RunPod object name is not a secret and appears in its UI, so the +// username must not be in it; and naming it after the claim would create one object per +// NodeClaim for a credential every claim shares. +// +// Truncated to 16 hex characters — 64 bits, which for a per-account population of at most a +// handful of credentials is far past any collision concern, and keeps the name readable. +func registryAuthName(username, password string) string { + // The NUL separator keeps ("ab", "c") from hashing the same as ("a", "bc"). + sum := sha256.Sum256([]byte(username + "\x00" + password)) + return namePrefix + hex.EncodeToString(sum[:])[:16] +} diff --git a/pkg/provider/runpod/client_test.go b/pkg/provider/runpod/client_test.go new file mode 100644 index 0000000..63c8e1a --- /dev/null +++ b/pkg/provider/runpod/client_test.go @@ -0,0 +1,487 @@ +/* +Copyright 2026 The InftyAI Team. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package runpod + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/InftyAI/Nebula/pkg/provider" +) + +// recordedRequest is what the fake RunPod saw, so a test can assert on the wire form the +// adapter produced rather than only on what it got back. +type recordedRequest struct { + method string + path string + query string + auth string + body map[string]any +} + +// testServer stands in for RunPod's REST API. handler answers each call; every request is +// recorded first. Returns the client under test and a pointer to the log. +func testServer(t *testing.T, handler http.HandlerFunc) (*restClient, *[]recordedRequest) { + t.Helper() + var seen []recordedRequest + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + rec := recordedRequest{ + method: r.Method, + path: r.URL.Path, + query: r.URL.RawQuery, + auth: r.Header.Get("Authorization"), + } + if raw, err := io.ReadAll(r.Body); err == nil && len(raw) > 0 { + _ = json.Unmarshal(raw, &rec.body) + } + seen = append(seen, rec) + handler(w, r) + })) + t.Cleanup(srv.Close) + return newClient(srv.URL, "test-key"), &seen +} + +// jsonReply writes one canned response. +func jsonReply(status int, body string) http.HandlerFunc { + return func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = io.WriteString(w, body) + } +} + +func TestClassifyCreate(t *testing.T) { + // Every one of these must carry a sentinel or deliberately carry none: unwrapped, a + // rejection lands on nebula_provision_failures_total{reason="other"} and is retried + // forever against a provider that has already given its answer. + cases := []struct { + name string + // status and message are what RunPod answered; spot is the request's tier, which the + // error itself never carries. + status int + message string + spot bool + + want error // the sentinel the error must wrap, nil for "none" + wantSpot bool // must also carry the spot-tier marker + }{{ + name: "401 is auth", status: 401, message: "Unauthorized", want: provider.ErrAuth, + }, { + name: "403 is auth", status: 403, message: "Forbidden", want: provider.ErrAuth, + }, { + // A 5xx says RunPod failed to ANSWER, not that it said no — and it may have created + // the Pod before falling over. A sentinel here would fail the Pod and blocklist a + // candidate on a server-side blip, possibly reaping a paid instance. + name: "500 stays unwrapped", status: 500, message: "internal error", want: nil, + }, { + name: "502 stays unwrapped", status: 502, message: "bad gateway", want: nil, + }, { + name: "429 is quota", status: 429, message: "Too Many Requests", want: provider.ErrQuota, + }, { + // Money, not capacity, but scoped the same way: transient, and not an auth problem. + name: "402 is quota", status: 402, message: "Payment Required", want: provider.ErrQuota, + }, { + name: "insufficient funds is quota", + status: 400, + message: "Insufficient funds to start this pod", + want: provider.ErrQuota, + }, { + name: "no instances available is capacity", + status: 400, + message: "There are no longer any instances available with the requested specifications", + want: provider.ErrNoCapacity, + }, { + // The tier marker is what keeps a Spot shortage from disabling OnDemand capacity + // that is still purchasable at the higher price. + name: "a spot shortage also marks the tier", + status: 400, + message: "no instances available", + spot: true, + want: provider.ErrNoCapacity, + wantSpot: true, + }, { + name: "an unknown gpu type is an accelerator problem", + status: 400, + message: "invalid gpu type id", + want: provider.ErrUnsupportedAccelerator, + }, { + // Belongs to the REQUEST, not the candidate: ErrImagePull blocklists nothing, so one + // Pod's bad credential cannot exclude an accelerator serving every other Pod. + name: "a registry failure is an image-pull problem", + status: 400, + message: "could not authenticate with registry", + want: provider.ErrImagePull, + }, { + // None of the shared sentinels describes "rejected, reason unknown", and every guess + // is worse than retrying: ErrAuth would fence off the provider, a capacity wrap would + // evict a healthy candidate. A sustained reason="other" rate is the signal to add a + // row to the table. + name: "an unrecognized 4xx stays unwrapped", status: 400, message: "something new", want: nil, + }} + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c, _ := testServer(t, jsonReply(tc.status, fmt.Sprintf(`{"error":%q}`, tc.message))) + + _, err := c.CreatePod(context.Background(), PodSpec{ + Name: "nebula-claim-a", Image: "img", GPUCount: 1, + GPUTypeIDs: []string{"NVIDIA H100 80GB HBM3"}, Interruptible: tc.spot, + }) + if err == nil { + t.Fatal("CreatePod succeeded against an error response") + } + if tc.want != nil && !errors.Is(err, tc.want) { + t.Errorf("error %v does not wrap %v", err, tc.want) + } + if tc.want == nil { + // Assert it wraps NONE of them: the point of leaving it unwrapped is that + // provider.IsRejection reads it as unattributable. + for _, s := range []error{ + provider.ErrAuth, provider.ErrQuota, provider.ErrNoCapacity, + provider.ErrUnsupportedAccelerator, provider.ErrImagePull, + } { + if errors.Is(err, s) { + t.Errorf("error %v wraps %v; it must stay unattributable", err, s) + } + } + } + if got := errors.Is(err, ErrSpotCapacity); got != tc.wantSpot { + t.Errorf("errors.Is(err, ErrSpotCapacity) = %t, want %t", got, tc.wantSpot) + } + // The status and RunPod's own words survive into the message, which is what an + // operator reads off a Pod condition. + if !strings.Contains(err.Error(), tc.message) { + t.Errorf("error %q dropped RunPod's message %q", err, tc.message) + } + }) + } +} + +func TestCreatePod_WireForm(t *testing.T) { + c, seen := testServer(t, jsonReply(200, `{"id":"pod-1"}`)) + + id, err := c.CreatePod(context.Background(), PodSpec{ + Name: "nebula-claim-a", + Image: "myimg:latest", + GPUTypeIDs: []string{"NVIDIA H100 80GB HBM3", "NVIDIA H100 NVL"}, + GPUCount: 2, + VCPUPerGPU: 5, + RAMPerGPUGiB: 50, + DataCenterIDs: []string{"US-KS-2"}, + Interruptible: true, + }) + if err != nil { + t.Fatalf("CreatePod: %v", err) + } + if id != "pod-1" { + t.Fatalf("id = %q, want pod-1", id) + } + req := (*seen)[0] + if req.method != http.MethodPost || req.path != "/pods" { + t.Errorf("%s %s, want POST /pods", req.method, req.path) + } + if req.auth != "Bearer test-key" { + t.Errorf("Authorization = %q", req.auth) + } + // volumeInGb must be PRESENT and 0: RunPod's default is a billable 20 GiB persistent + // volume, and a Nebula instance is cattle with nothing to persist. omitempty here would + // silently restore that default — which is why the field carries no omitempty tag. + v, ok := req.body["volumeInGb"] + if !ok { + t.Error("volumeInGb absent; RunPod would then attach its default 20 GiB billable volume") + } else if v != float64(0) { + t.Errorf("volumeInGb = %v, want 0", v) + } + // Likewise interruptible: false is the tier that costs money, so it is stated rather + // than left to a default. + if got, ok := req.body["interruptible"]; !ok || got != true { + t.Errorf("interruptible = %v (present=%t), want true", got, ok) + } + // SECURE only: COMMUNITY prices the same GPU differently and the catalog has no + // cloud-type axis to express that, so offering it would make the price a guess. + if req.body["cloudType"] != cloudTypeSecure { + t.Errorf("cloudType = %v, want %q", req.body["cloudType"], cloudTypeSecure) + } + if req.body["computeType"] != "GPU" { + t.Errorf("computeType = %v, want GPU", req.body["computeType"]) + } + // Only meaningful WITH alternates: it tells RunPod to satisfy the list by whichever id + // has capacity rather than insisting on the first. + if req.body["gpuTypePriority"] != "availability" { + t.Errorf("gpuTypePriority = %v, want availability for a multi-id request", req.body["gpuTypePriority"]) + } + // custom, not availability: the pool NAMED this data center, so letting RunPod place + // elsewhere would silently break a constraint an operator set for data residency. + if req.body["dataCenterPriority"] != "custom" { + t.Errorf("dataCenterPriority = %v, want custom", req.body["dataCenterPriority"]) + } + if req.body["minVCPUPerGPU"] != float64(5) || req.body["minRAMPerGPU"] != float64(50) { + t.Errorf("per-GPU sizing = %v/%v, want 5/50", + req.body["minVCPUPerGPU"], req.body["minRAMPerGPU"]) + } +} + +func TestCreatePod_SingleGPUIDAndCPUOnly(t *testing.T) { + t.Run("one id sends no priority", func(t *testing.T) { + // With a single id there is nothing to prioritize, so the field would be noise. + c, seen := testServer(t, jsonReply(200, `{"id":"pod-1"}`)) + if _, err := c.CreatePod(context.Background(), PodSpec{ + Name: "nebula-a", Image: "img", GPUCount: 1, GPUTypeIDs: []string{"NVIDIA L4"}, + }); err != nil { + t.Fatalf("CreatePod: %v", err) + } + if _, ok := (*seen)[0].body["gpuTypePriority"]; ok { + t.Error("gpuTypePriority sent for a single-id request") + } + }) + + t.Run("no GPU is a different product", func(t *testing.T) { + // A CPU-only Pod's per-GPU sizing fields are meaningless; vcpuCount replaces them. + c, seen := testServer(t, jsonReply(200, `{"id":"pod-cpu"}`)) + if _, err := c.CreatePod(context.Background(), PodSpec{ + Name: "nebula-c", Image: "img", VCPUCount: 3, + }); err != nil { + t.Fatalf("CreatePod: %v", err) + } + body := (*seen)[0].body + if body["computeType"] != "CPU" { + t.Errorf("computeType = %v, want CPU", body["computeType"]) + } + if body["vcpuCount"] != float64(3) { + t.Errorf("vcpuCount = %v, want 3", body["vcpuCount"]) + } + if _, ok := body["gpuCount"]; ok { + t.Error("gpuCount sent on a CPU-only Pod") + } + }) +} + +func TestCreatePod_SuccessWithNoID(t *testing.T) { + // A 2xx with no id is worse than an error: a Pod may exist that we can never name to + // terminate. It must fail WITHOUT a sentinel so the Pod retries — Provision is + // idempotent on the claim name, and the name lookup will find whatever this call created. + c, _ := testServer(t, jsonReply(201, `{}`)) + + _, err := c.CreatePod(context.Background(), PodSpec{Name: "nebula-a", Image: "img", GPUCount: 1}) + if err == nil { + t.Fatal("CreatePod accepted a response with no id") + } + if provider.IsRejection(err) { + t.Errorf("error %v reads as a rejection; it must stay retryable", err) + } +} + +func TestTerminatePod_404IsSuccess(t *testing.T) { + // The NodeClaim finalizer retries Terminate, so an already-gone Pod has to be success — + // otherwise the finalizer never clears and the Pod is stuck deleting forever. + c, seen := testServer(t, jsonReply(404, `{"error":"pod not found"}`)) + if err := c.TerminatePod(context.Background(), "pod-gone"); err != nil { + t.Fatalf("TerminatePod on a missing Pod = %v, want nil", err) + } + if req := (*seen)[0]; req.method != http.MethodDelete || req.path != "/pods/pod-gone" { + t.Errorf("%s %s, want DELETE /pods/pod-gone", req.method, req.path) + } + + // Any OTHER failure must still surface: swallowing a 500 would drop the teardown + // obligation and leak a billing instance. + c2, _ := testServer(t, jsonReply(500, `{"error":"boom"}`)) + if err := c2.TerminatePod(context.Background(), "pod-1"); err == nil { + t.Error("TerminatePod swallowed a 500; the instance would leak") + } +} + +func TestGetPod_404IsGone(t *testing.T) { + // Absent means terminated, per the interface contract. + c, seen := testServer(t, jsonReply(404, `{"error":"not found"}`)) + pd, err := c.GetPod(context.Background(), "pod-gone") + if err != nil || pd != nil { + t.Fatalf("GetPod(missing) = %v, %v; want nil, nil", pd, err) + } + // includeMachine is what populates the data center, so every read path must ask for it — + // without it Region would silently stay empty on every instance. + if req := (*seen)[0]; !strings.Contains(req.query, "includeMachine=true") { + t.Errorf("GetPod query = %q, want includeMachine=true", req.query) + } +} + +func TestListPods_DecodesMachineAndPorts(t *testing.T) { + c, seen := testServer(t, jsonReply(200, `[ + {"id":"pod-1","name":"nebula-claim-a","desiredStatus":"RUNNING", + "lastStartedAt":"2026-08-29T00:00:00Z","interruptible":true, + "ports":["8000/http"],"machine":{"dataCenterId":"EU-RO-1"}}, + {"id":"pod-2","name":"nebula-claim-b","desiredStatus":"EXITED","machine":null} + ]`)) + + pods, err := c.ListPods(context.Background()) + if err != nil { + t.Fatalf("ListPods: %v", err) + } + // The list path needs includeMachine for the same reason Get does; it is the poll loop's + // only source of an instance's region. + if req := (*seen)[0]; !strings.Contains(req.query, "includeMachine=true") { + t.Errorf("ListPods query = %q, want includeMachine=true", req.query) + } + if len(pods) != 2 { + t.Fatalf("got %d pods, want 2", len(pods)) + } + if pods[0].DataCenterID != "EU-RO-1" || !pods[0].Interruptible { + t.Errorf("pod-1 = %+v", pods[0]) + } + if len(pods[0].Ports) != 1 || pods[0].Ports[0] != "8000/http" { + t.Errorf("pod-1 ports = %v; they are what makes an /http-only Pod addressable", pods[0].Ports) + } + // A null machine must leave the region EMPTY rather than reporting a placement that was + // never observed. + if pods[1].DataCenterID != "" { + t.Errorf("pod-2 region = %q, want empty for a null machine", pods[1].DataCenterID) + } +} + +func TestEnsureRegistryAuth(t *testing.T) { + auth := &provider.RegistryAuth{ + Registry: "ghcr.io", + Basic: &provider.BasicAuth{Username: "u", Password: "p4ssw0rd"}, + } + // Content-addressed: the SAME credential always resolves to the same object name, which + // is what makes calling this on every Provision safe. + name := registryAuthName("u", "p4ssw0rd") + + t.Run("reuses an existing object", func(t *testing.T) { + // One object is shared by every Pod using that credential. Creating a second per Pod + // would accumulate objects without bound, and RunPod never garbage-collects them. + c, seen := testServer(t, func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + t.Error("POST issued although a matching object already exists") + } + jsonReply(200, fmt.Sprintf(`[{"id":"cra-existing","name":%q}, + {"id":"cra-other","name":"nebula-deadbeefdeadbeef"}]`, name))(w, r) + }) + + id, err := c.EnsureRegistryAuth(context.Background(), auth) + if err != nil { + t.Fatalf("EnsureRegistryAuth: %v", err) + } + if id != "cra-existing" { + t.Errorf("id = %q, want cra-existing", id) + } + if len(*seen) != 1 { + t.Errorf("made %d calls, want 1 (the list)", len(*seen)) + } + }) + + t.Run("creates when absent, and never sends the password back on the list", func(t *testing.T) { + c, seen := testServer(t, func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + jsonReply(200, `[]`)(w, r) + return + } + jsonReply(201, `{"id":"cra-new","name":"whatever"}`)(w, r) + }) + + id, err := c.EnsureRegistryAuth(context.Background(), auth) + if err != nil { + t.Fatalf("EnsureRegistryAuth: %v", err) + } + if id != "cra-new" { + t.Errorf("id = %q, want cra-new", id) + } + if len(*seen) != 2 { + t.Fatalf("made %d calls, want 2 (list then create)", len(*seen)) + } + post := (*seen)[1] + if post.method != http.MethodPost || post.path != registryAuthPath { + t.Errorf("%s %s, want POST %s", post.method, post.path, registryAuthPath) + } + // The object NAME must be the hash, not the username or the claim: a RunPod object + // name is not secret and shows in its UI, and naming it after the claim would create + // one object per NodeClaim for a credential every claim shares. + if post.body["name"] != name { + t.Errorf("name = %v, want the content-addressed %q", post.body["name"], name) + } + if post.body["username"] != "u" || post.body["password"] != "p4ssw0rd" { + t.Errorf("credential did not reach the create body: %v", post.body["username"]) + } + }) + + t.Run("a rotated password becomes a new object", func(t *testing.T) { + // The hash covers BOTH fields, so a rotation cannot silently reuse a stale object + // that would 401 at pull time. + if registryAuthName("u", "old") == registryAuthName("u", "new") { + t.Error("a rotated password hashes to the same object name") + } + // And the NUL separator keeps ("ab","c") from colliding with ("a","bc"). + if registryAuthName("ab", "c") == registryAuthName("a", "bc") { + t.Error("username/password boundary is not separated in the hash") + } + if !strings.HasPrefix(name, namePrefix) { + t.Errorf("name %q lacks the %q prefix that makes Nebula's objects recognizable", + name, namePrefix) + } + }) + + t.Run("a create failure blocklists nothing", func(t *testing.T) { + // A credential RunPod will not store is a fact about this Pod's imagePullSecret, not + // about the accelerator or region it was headed for — so ErrImagePull, which the + // classifier maps to the zero BlockScope. + c, _ := testServer(t, func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + jsonReply(200, `[]`)(w, r) + return + } + jsonReply(400, `{"error":"invalid credential"}`)(w, r) + }) + + _, err := c.EnsureRegistryAuth(context.Background(), auth) + if !errors.Is(err, provider.ErrImagePull) { + t.Errorf("error = %v, want it to wrap ErrImagePull", err) + } + }) + + t.Run("a kind RunPod cannot express is refused, not silently dropped", func(t *testing.T) { + // The adapter vets the kind first, so this is a programming error — but it must never + // become a silent ANONYMOUS pull, which either 401s opaquely or succeeds against a + // PUBLIC image of the same name. + c, seen := testServer(t, jsonReply(200, `[]`)) + _, err := c.EnsureRegistryAuth(context.Background(), &provider.RegistryAuth{ + Registry: "1234.dkr.ecr.us-east-1.amazonaws.com", + AWSRole: &provider.AWSRoleAuth{RoleARN: "arn:aws:iam::1234:role/pull", Region: "us-east-1"}, + }) + if !errors.Is(err, provider.ErrImagePull) { + t.Errorf("error = %v, want it to wrap ErrImagePull", err) + } + if len(*seen) != 0 { + t.Errorf("made %d API calls for a credential it cannot express", len(*seen)) + } + }) +} + +func TestNewSDKClient_MissingKeyIsSkippable(t *testing.T) { + // An absent key must be an ERROR rather than a client that fails on first use, so + // registerProviders can log and skip RunPod the way it skips Modal and AWS — an operator + // who configured only Modal must not get a fatal boot. + t.Setenv(apiKeyEnv, "") + if _, err := NewSDKClient(context.Background()); err == nil { + t.Fatal("NewSDKClient succeeded with no API key; registration would silently register a dead provider") + } +} diff --git a/pkg/provider/runpod/runpod.go b/pkg/provider/runpod/runpod.go new file mode 100644 index 0000000..efca8a4 --- /dev/null +++ b/pkg/provider/runpod/runpod.go @@ -0,0 +1,738 @@ +/* +Copyright 2026 The InftyAI Team. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package runpod implements the provider.Provider interface for RunPod +// (https://runpod.io), a NeoCloud renting GPU containers by the second. One NodeClaim +// maps to one RunPod Pod. +// +// RunPod sits between the two adapters that came before it, and the differences are +// what drive the decisions here: +// +// - Lifecycle is create/terminate only for our purposes, so SupportsStop=false. RunPod +// does expose stop/start, but a stopped Pod still bills for its disk and releases the +// GPU, so it is neither free nor resumable in the sense the capability promises. +// - Spot is REAL and is one boolean: `interruptible: true`, with no bid to name (the +// REST v1 API dropped bidPerGpu). So SupportsSpot=true and, unlike Modal, the +// capacity tier on the request is actually honoured. +// - There is NO preemption push and no notice window, so reclaims are noticed only by +// the poll loop — hence PreemptionNotice=0 and a faster-than-default PollInterval. +// - Create FAILS SYNCHRONOUSLY when capacity is short ("no longer any instances +// available..."), the AWS behaviour rather than Modal's queue-and-accept. That is +// what makes region failover meaningful, so ExpandRegions is left as catalog.Base's +// pass-through: one candidate per declared region, each blocklistable on its own. +// - Pods have NO tags. NativeTags=false, and Nebula's identity rides the Pod NAME +// (see podName/claimFromName), which is what List filters on. +// - There is no outbound-allowlist knob at all, so SupportsEgressPolicy=false and +// placement skips RunPod for any pool that restricts egress rather than provisioning +// something with open internet access under a policy that says otherwise. +// - Only SECURE cloud is used. RunPod's cheaper COMMUNITY cloud prices the same GPU +// differently, and the catalog CSV has no cloud-type axis to express that, so +// offering both would make the price the optimizer reads a guess. +// +// kubectl logs and exec are NOT served: RunPod's REST v1 surface has no pod-log endpoint, +// and its only way into a container is SSH, which needs key material this adapter has +// nowhere to put. Both are optional halves of provider.Provider resolved by type +// assertion, so leaving them out costs nothing but a NotFound. +// +// The concrete HTTP API lives behind the Client seam, so this file holds only +// provider-agnostic translation and is unit-testable without network access. +package runpod + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + + nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" + "github.com/InftyAI/Nebula/pkg/provider" + "github.com/InftyAI/Nebula/pkg/provider/catalog" + "github.com/InftyAI/Nebula/pkg/util" +) + +// namePrefix marks a RunPod Pod as Nebula's. RunPod has no tags, so the name is the +// ONLY carrier of ownership and identity: List filters on this prefix so an unrelated +// Pod in the same account is never adopted, terminated, or reported as an instance. +const namePrefix = "nebula-" + +// maxNameLen is RunPod's own cap on the Pod name (191 chars). Nebula refuses a claim +// whose name would exceed it rather than truncating — see podName. +const maxNameLen = 191 + +// spotPollInterval is how often to re-list. RunPod's interruptible tier reclaims +// abruptly with no notice pushed to us, so a faster cadence than the vnode default is +// what turns "the Pod vanished" into a NodeClaim update promptly. Matches AWS. +const spotPollInterval = 10 * time.Second + +// ErrSpotCapacity is a marker the Client wraps onto an interruptible-tier capacity +// failure (alongside provider.ErrNoCapacity) so ClassifyProvisionError — handed only the +// error, never the request — can recover that the failing tier was Spot and block only +// Spot, leaving OnDemand serviceable. The same device as aws.ErrSpotCapacity. +var ErrSpotCapacity = errors.New("runpod: spot capacity") + +// compile-time assertion that Provider satisfies the interface. LogStreamer and Executor +// are deliberately absent; see the package doc. +var _ provider.Provider = (*Provider)(nil) + +// Client is the narrow seam over RunPod's REST API: only the operations the adapter +// needs, in provider-agnostic terms, so the real HTTP implementation and a test fake are +// interchangeable. +type Client interface { + // CreatePod launches one Pod from spec and returns its RunPod id. A capacity + // shortage is an ERROR here, not a queued Pod, and must be wrapped with + // provider.ErrNoCapacity (plus ErrSpotCapacity on the interruptible tier). + CreatePod(ctx context.Context, spec PodSpec) (id string, err error) + // TerminatePod deletes a Pod by id. Must be idempotent: deleting an already-gone + // Pod returns nil, since the NodeClaim finalizer retries against it. + TerminatePod(ctx context.Context, id string) error + // GetPod returns one Pod, or (nil, nil) if it no longer exists. + GetPod(ctx context.Context, id string) (*Pod, error) + // ListPods returns every Pod in the account, in as few calls as possible. Filtering + // down to Nebula's own is the ADAPTER's job (it owns the naming scheme), so this + // must not filter. + ListPods(ctx context.Context) ([]Pod, error) + // EnsureRegistryAuth resolves auth to a RunPod containerRegistryAuth id, creating + // the object if this credential has no id yet. RunPod's create takes an id, never an + // inline username/password, so this indirection is unavoidable. + EnsureRegistryAuth(ctx context.Context, auth *provider.RegistryAuth) (id string, err error) +} + +// PodSpec is the resolved, RunPod-shaped request the Client turns into a Pod. The +// adapter builds it from the Pod (source of truth) plus the resolved accelerator ids. +type PodSpec struct { + // Name is the RunPod Pod name, which carries Nebula's identity because RunPod has no + // tags: namePrefix + the NodeClaim name. See podName. + Name string + // Image is the container image, from the Pod's first container. + Image string + // Entrypoint and StartCmd are the container's command and args, mapped onto RunPod's + // dockerEntrypoint/dockerStartCmd. They stay SEPARATE, unlike Modal where both + // concatenate into one command: RunPod's two fields are ENTRYPOINT and CMD, so + // Kubernetes' command/args land on their exact Docker counterparts. + Entrypoint []string + StartCmd []string + // Env is the environment, taken whole from provider.ProvisionRequest.Env: literals + // plus everything envFrom/valueFrom referenced, already resolved by the caller. + // + // SECRET-BEARING, hence the redacting String below. + Env map[string]string + // GPUTypeIDs are RunPod's own accelerator ids that can serve the request, PRIMARY + // first. All of them go out in one create: RunPod takes an array and picks by + // availability, so interchangeable ids broaden a SINGLE launch (as AWS's fleet spans + // instance types) without widening what a failure blocklists — the block keys on the + // canonical pool, not on whichever id RunPod landed. Empty for a CPU-only Pod. + GPUTypeIDs []string + // GPUCount is how many accelerators to attach; 0 selects a CPU-only Pod. + GPUCount int32 + // VCPUPerGPU and RAMPerGPUGiB are the Pod's cpu/memory requests expressed RunPod's + // way — PER GPU, not in total, so the adapter divides by GPUCount and rounds UP + // (rounding down would hand the workload less than it asked for). Zero leaves + // RunPod's own defaults (2 vCPU, 8 GiB per GPU). + VCPUPerGPU int + RAMPerGPUGiB int + // VCPUCount is the CPU-only equivalent, an absolute count rather than a per-GPU one. + // Only read when GPUCount is 0. + VCPUCount int + // ContainerDiskGiB is the writable container disk, from the Pod's ephemeral-storage + // request. Zero leaves RunPod's default (50 GiB). + // + // No persistent volume is ever requested: RunPod defaults to a billable 20 GiB one, + // and a Nebula instance is cattle with nothing to persist, so the Client pins it to 0. + ContainerDiskGiB int + // Ports are the container ports to expose, in RunPod's "/" form (see + // containerPorts). Empty leaves RunPod's default exposure. + Ports []string + // DataCenterIDs and CountryCodes are the placement constraint, split out of the ONE + // region candidate this request carries (see splitRegion). At most one is ever set, + // and both empty means unconstrained — the widest capacity pool, and the normal case + // for a pool that declares no regions. + DataCenterIDs []string + CountryCodes []string + // Interruptible asks for the spot tier, from ProvisionRequest.CapacityType. RunPod's + // REST v1 takes no bid alongside it, so nothing else is needed to price it. + Interruptible bool + // RegistryAuthID is the RunPod containerRegistryAuth object authenticating the image + // pull, already resolved from the canonical credential by Client.EnsureRegistryAuth. + // Empty is an anonymous pull. + RegistryAuthID string +} + +// String redacts Env so a spec can be logged or wrapped in an error safely: key names +// print (they are in the Pod spec already), values never do. RegistryAuthID is an opaque +// object id, not a credential, so it prints as-is. +func (s PodSpec) String() string { + return fmt.Sprintf("PodSpec{Name:%s Image:%s Entrypoint:%v StartCmd:%v Env:%s "+ + "GPUTypeIDs:%v GPUCount:%d VCPUPerGPU:%d RAMPerGPUGiB:%d VCPUCount:%d "+ + "ContainerDiskGiB:%d Ports:%v DataCenterIDs:%v CountryCodes:%v Interruptible:%t "+ + "RegistryAuthID:%s}", + s.Name, s.Image, s.Entrypoint, s.StartCmd, provider.RedactedEnv(s.Env), + s.GPUTypeIDs, s.GPUCount, s.VCPUPerGPU, s.RAMPerGPUGiB, s.VCPUCount, + s.ContainerDiskGiB, s.Ports, s.DataCenterIDs, s.CountryCodes, s.Interruptible, + s.RegistryAuthID) +} + +// GoString implements fmt.GoStringer so %#v is redacted too. +func (s PodSpec) GoString() string { return s.String() } + +// Pod is the adapter-level view of a RunPod Pod as observed. +type Pod struct { + ID string + Name string + // DesiredStatus is RunPod's own status string (RUNNING/EXITED/TERMINATED). It is the + // DESIRED state, not the observed one, which is why toState also needs LastStartedAt. + DesiredStatus string + // LastStartedAt is when the container last started, empty until it has. Paired with + // DesiredStatus it is the closest thing RunPod offers to a readiness signal. + LastStartedAt string + // Interruptible is whether this Pod is on the spot tier, so an observed instance + // reports the tier it actually got. + Interruptible bool + // DataCenterID is where RunPod placed it, in RunPod's own vocabulary. + DataCenterID string + // Ports are the exposed ports RunPod echoes back, in the same "/" form + // PodSpec.Ports sends. They are what the proxy URL routes to, and unlike PortMappings + // they are present from creation, which is what makes an /http-only Pod addressable. + Ports []string + // PublicIP and PortMappings are the DIRECT address, keyed by container port, and only + // ever populated for a /tcp port RunPod has finished assigning. Both empty otherwise — + // including for every /http port, which is why they are a preference and not the + // endpoint's only source. + PublicIP string + PortMappings map[string]int +} + +// Provider is the RunPod implementation of provider.Provider. It embeds catalog.Base for +// the generic catalog methods — Name, Offerings, ExpandRegions and the catalog-driven +// MapAccelerator, which does real work here: RunPod's accelerator ids are marketing +// strings ("NVIDIA H100 80GB HBM3") that share nothing with Nebula's canonical names, so +// every row in runpod.csv carries an accelerator_id. +type Provider struct { + catalog.Base + client Client +} + +// New returns a RunPod Provider backed by client and price catalog. Both must be +// non-nil; use catalog.Load() to build the catalog from the CSV/ConfigMap data. cat is +// the catalog.Lookup seam, so tests can inject a fake. +func New(client Client, cat catalog.Lookup) *Provider { + return &Provider{ + Base: catalog.Base{ProviderName: provider.ProviderRunPod, Catalog: cat}, + client: client, + } +} + +// Capabilities implements provider.Provider. See the package doc for why each trait is +// set the way it is. +func (p *Provider) Capabilities() provider.Capabilities { + return provider.Capabilities{ + SupportsStop: false, // a stopped Pod still bills and loses its GPU + SupportsSpot: true, // `interruptible`, no bid to name + SupportsEgressPolicy: false, // no outbound allowlist in the API at all + NativeTags: false, // identity rides the Pod name + PreemptionNotice: 0, // abrupt; detected only by polling + PollInterval: spotPollInterval, // Spot reclaims are abrupt; poll faster than default + // No ProvisionTimeout: one create call, no internal sweep across capacity pools + // to bound (contrast AWS, which walks a region's AZs itself). + } +} + +// Provision implements provider.Provider. The Pod is the source of truth for the +// workload; req carries only the claim identity, the capacity tier and the region. +// +// A successful create is RESERVED, unlike Modal: RunPod allocates a host machine before +// it answers, and a shortage comes back as an error rather than a queued Pod. So an id +// here means real capacity, and the Pod may honestly move on from "provisioning". +// +// The connect URL is RunPod's HTTP proxy for the first declared port — deterministic from +// the Pod id, so no read-back is needed. There is no token: the proxy is unauthenticated, +// which is why ConnectToken stays empty rather than carrying a placeholder. +func (p *Provider) Provision( + ctx context.Context, pod *corev1.Pod, req provider.ProvisionRequest, +) (provider.ProvisionResult, error) { + if pod == nil { + return provider.ProvisionResult{}, errors.New("runpod: nil pod") + } + if req.ClaimName == "" { + return provider.ProvisionResult{}, errors.New("runpod: empty ClaimName in ProvisionRequest") + } + // Refuse a restrictive egress policy rather than dropping it. Placement checks + // SupportsEgressPolicy and should never route such a pool here, but the request can be + // built by anyone, and silently ignoring it would put the workload on the open + // internet under a policy that says otherwise. + if mode := req.Egress.ModeOrOpen(); mode != nebulav1alpha1.EgressOpen { + return provider.ProvisionResult{}, fmt.Errorf( + "runpod: cannot enforce egress mode %q; RunPod exposes no outbound policy", mode) + } + + // Idempotency: RunPod has no tags, so the claim is looked up by the Pod NAME this + // adapter mints. A repeat after a partial create returns the existing Pod rather than + // paying for a second. + // + // Reserved is true for the same reason as a fresh create: the Pod exists, so a machine + // was allocated. That is the capacity question; readiness is separate and observed + // through List. No credential comes back — the interface forbids it on a re-Provision, + // and here there is nothing to re-mint anyway, since the proxy URL is derivable. + existing, err := p.findByClaim(ctx, req.ClaimName) + if err != nil { + return provider.ProvisionResult{}, err + } + if existing != nil { + return provider.ProvisionResult{InstanceID: existing.ID, Reserved: true}, nil + } + + spec, err := p.podSpecFromPod(pod, req) + if err != nil { + return provider.ProvisionResult{}, err + } + // Resolve the pull credential LAST among the request-shaping steps, because unlike + // everything else in the spec it costs API calls; a spec that was going to be rejected + // for its accelerator or its name has already failed by now. + if req.RegistryAuth != nil { + if err := checkRegistryAuth(req.RegistryAuth); err != nil { + return provider.ProvisionResult{}, err + } + authID, err := p.client.EnsureRegistryAuth(ctx, req.RegistryAuth) + if err != nil { + return provider.ProvisionResult{}, err + } + spec.RegistryAuthID = authID + } + + id, err := p.client.CreatePod(ctx, spec) + if err != nil { + return provider.ProvisionResult{}, err + } + return provider.ProvisionResult{ + InstanceID: id, + Reserved: true, + ConnectURL: proxyURL(id, spec.Ports), + }, nil +} + +// Terminate implements provider.Provider. Idempotent by the Client contract. The region +// is ignored: RunPod's API is global and a Pod id addresses it from anywhere, so region is +// only ever a placement input. +func (p *Provider) Terminate(ctx context.Context, instanceID, _ string) error { + if instanceID == "" { + return nil // nothing provisioned yet; treat as already gone + } + return p.client.TerminatePod(ctx, instanceID) +} + +// Get implements provider.Provider. The region is ignored, as in Terminate. +func (p *Provider) Get(ctx context.Context, instanceID, _ string) (*provider.Instance, error) { + pd, err := p.client.GetPod(ctx, instanceID) + if err != nil { + return nil, err + } + if pd == nil { + return nil, nil // absent => terminated, per interface contract + } + inst := toInstance(*pd) + return &inst, nil +} + +// List implements provider.Provider. One API call, then the name filter that stands in for +// the tags RunPod does not have: a Pod without Nebula's prefix belongs to someone else in +// the same account and must never be reported (the poll loop would adopt it, and the +// NodeClaim controller would eventually terminate it). +func (p *Provider) List(ctx context.Context) ([]provider.Instance, error) { + pods, err := p.client.ListPods(ctx) + if err != nil { + return nil, err + } + out := make([]provider.Instance, 0, len(pods)) + for _, pd := range pods { + if !strings.HasPrefix(pd.Name, namePrefix) { + continue + } + out = append(out, toInstance(pd)) + } + return out, nil +} + +// ClassifyProvisionError implements provider.Provider. The categories and the +// scope-derivation rule are shared (provider.ClassifyError and the sentinels the Client +// wraps), so this supplies only the two RunPod-specific facts: which tier failed, and the +// region axis. +func (p *Provider) ClassifyProvisionError(err error, accelerator, region string) provider.BlockScope { + // No failure, no block. ClassifyError already returns the zero scope, but the region + // decoration below would repopulate it into a scope recordBlock would install. + if err == nil { + return provider.BlockScope{} + } + // The failing tier is not on the error's face, so the Client marks an interruptible + // shortage. Getting this wrong in the safe direction matters: a Spot failure blocked + // as OnDemand would disable capacity that is still purchasable. + tier := nebulav1alpha1.CapacityOnDemand + if errors.Is(err, ErrSpotCapacity) { + tier = nebulav1alpha1.CapacitySpot + } + scope := provider.ClassifyError(err, tier, accelerator) + // The zero scope means BLOCK NOTHING — a rejection of this request that says nothing + // about the candidate, such as an image credential RunPod cannot use. Stamping a region + // onto it would make it non-empty, and recordBlock would install a region-wide block + // across every accelerator: the same trap as the err == nil guard above. + if scope == (provider.BlockScope{}) { + return scope + } + // DenyAll already covers every region (auth fails everywhere), so narrowing it would + // contradict the category. An empty region leaves Region nil, which per BlockScope + // matches only candidates that carry no region either — the unconstrained pool — so + // the block never leaks onto region-pinned candidates. + if region != "" && !scope.DenyAll { + scope.Region = ®ion + } + return scope +} + +// findByClaim returns the Nebula-owned Pod for claimName, or nil if none. RunPod has no +// server-side tag filter, so this is List plus a name comparison. +func (p *Provider) findByClaim(ctx context.Context, claimName string) (*provider.Instance, error) { + name, err := podName(claimName) + if err != nil { + return nil, err + } + pods, err := p.client.ListPods(ctx) + if err != nil { + return nil, err + } + for _, pd := range pods { + if pd.Name == name { + inst := toInstance(pd) + return &inst, nil + } + } + return nil, nil +} + +// podName is the RunPod Pod name for a NodeClaim: the prefix that marks ownership plus +// the claim name, which is what makes List/findByClaim work on a backend with no tags. +// +// A name that would exceed RunPod's cap is an ERROR, never a truncation. Truncating would +// map two long claim names onto one Pod name, and every consequence of that collision is +// severe: findByClaim adopts the other claim's instance, so one Pod is billed twice and +// the other claim's teardown reaps the survivor. Refusing is loud and fixable (claim names +// derive from the Pod's, so the workload can be renamed); a collision is silent. +func podName(claimName string) (string, error) { + name := namePrefix + claimName + if len(name) > maxNameLen { + return "", fmt.Errorf( + "runpod: claim name %q is too long: RunPod caps a pod name at %d characters and identity "+ + "rides that name, so it cannot be shortened", claimName, maxNameLen) + } + return name, nil +} + +// claimFromName recovers the NodeClaim name podName encoded, or "" for a Pod that is not +// Nebula's. It is the tag read that RunPod's lack of tags forces into the naming scheme. +func claimFromName(name string) string { + return strings.TrimPrefix(name, namePrefix) +} + +// podSpecFromPod reads the workload off the Pod (source of truth) and the placement +// decisions off req, then maps the accelerator to RunPod's own ids. +func (p *Provider) podSpecFromPod(pod *corev1.Pod, req provider.ProvisionRequest) (PodSpec, error) { + if len(pod.Spec.Containers) == 0 { + return PodSpec{}, errors.New("runpod: pod has no containers") + } + c := pod.Spec.Containers[0] + if c.Image == "" { + return PodSpec{}, errors.New("runpod: pod's first container has no image") + } + name, err := podName(req.ClaimName) + if err != nil { + return PodSpec{}, err + } + + dcs, countries := splitRegion(req.Region) + spec := PodSpec{ + Name: name, + Image: c.Image, + // command → ENTRYPOINT, args → CMD: Kubernetes' two fields land on the Docker + // fields they are defined in terms of, so an image whose own CMD supplies the args + // keeps working when only command is overridden. + Entrypoint: c.Command, + StartCmd: c.Args, + // The caller's resolved map is the whole environment — the Pod's literals plus + // everything envFrom/valueFrom referenced. pod.Spec.Containers[0].Env is NOT read + // here: it holds references this adapter has no cluster access to follow. + Env: req.Env, + ContainerDiskGiB: ephemeralGiB(&c), + Ports: containerPorts(&c), + DataCenterIDs: dcs, + CountryCodes: countries, + Interruptible: req.CapacityType == nebulav1alpha1.CapacitySpot, + } + + // Accelerator type comes from the AcceleratorTypeLabel; count from the container's + // nvidia.com/gpu resource (see util.AcceleratorRequest). + canonical, count, err := util.AcceleratorRequest(pod) + if err != nil { + return PodSpec{}, fmt.Errorf("runpod: %w", err) + } + if canonical != "" { + // Every id, not just the primary: RunPod's gpuTypeIds is an array it selects from + // by availability, so alternates widen this one launch. ids[0] stays the pool + // identity failover blocks on, which is the caller's business, not RunPod's. + ids, ok := p.MapAccelerator(canonical, count) + if !ok { + return PodSpec{}, fmt.Errorf("runpod: unsupported accelerator %q: %w", + canonical, provider.ErrUnsupportedAccelerator) + } + spec.GPUTypeIDs = ids + spec.GPUCount = count + // RunPod sizes a GPU Pod's cpu/memory PER GPU, so the Pod's totals are divided by + // the count. Zero (nothing requested) leaves RunPod's own per-GPU defaults. + spec.VCPUPerGPU = perGPU(cores(resourceQty(&c, corev1.ResourceCPU)), count) + spec.RAMPerGPUGiB = perGPU(gib(resourceQty(&c, corev1.ResourceMemory)), count) + return spec, nil + } + // No accelerator label => a CPU-only Pod, which RunPod sizes with an absolute vCPU + // count instead of a per-GPU one. + spec.VCPUCount = cores(resourceQty(&c, corev1.ResourceCPU)) + return spec, nil +} + +// checkRegistryAuth reports whether RunPod can honour a pull credential, so Provision only +// pays for an EnsureRegistryAuth call on a kind that can work. +// +// A refusal, never a fallback: an anonymous pull of a private image either 401s opaquely or +// succeeds against a PUBLIC image of the same name. +func checkRegistryAuth(a *provider.RegistryAuth) error { + switch { + case a.Basic != nil: + if err := a.Validate(); err != nil { + return fmt.Errorf("runpod: %w", err) // every error out of this adapter is prefixed + } + return nil + default: + // AWSRole is the kind the canonical form carries that RunPod has no equivalent for: + // its registry credentials are a static username/password object, with nothing that + // assumes an IAM role on the workload's behalf. An ECR "password" is a 12-hour + // token, so smuggling one in as Basic would provision a Pod that stops being able + // to pull halfway through the day. + return a.Unsupported("runpod") + } +} + +// splitRegion turns the ONE region candidate placement chose into RunPod's two placement +// fields. Empty means unconstrained (no pool regions declared) and yields neither, which is +// the widest capacity pool. +// +// The split is by SHAPE, not by a lookup table: RunPod's data-center ids are compound +// ("EU-RO-1", "US-KS-2"), while a bare two-letter token is an ISO country code its +// countryCodes field takes ("us", "se"). So a pool declaring a geography gets one, a pool +// naming a data center gets the other, and no static list of data centers has to be +// maintained here — which matters because such a list rots silently: a stale entry only +// surfaces as a rejected create for a region that exists. +// +// This is also why ExpandRegions is left as catalog.Base's pass-through. AWS has to expand +// its group tokens because "us" is not a callable region name; for RunPod it IS callable, +// as a country code, so there is nothing to expand and one declared token stays one +// blocklistable candidate. +func splitRegion(region string) (dataCenterIDs, countryCodes []string) { + region = strings.TrimSpace(region) + if region == "" { + return nil, nil + } + if len(region) == 2 { + return nil, []string{strings.ToUpper(region)} + } + return []string{region}, nil +} + +// containerPorts renders the container's declared ports in RunPod's "/" form. +// +// Every port goes out as /http, not /tcp, and that is a real choice: RunPod's http scheme +// publishes a proxy URL derivable from the Pod id (see proxyURL), so the workload is +// reachable the moment it comes up, whereas /tcp reaches it only through a randomly +// assigned public port that must be read back after boot. A Pod serving raw TCP is +// therefore not addressable today; a containerPort carries no hint of its protocol above +// TCP/UDP, so the common case is what gets served. +func containerPorts(c *corev1.Container) []string { + if len(c.Ports) == 0 { + return nil + } + ports := make([]string, 0, len(c.Ports)) + for _, p := range c.Ports { + ports = append(ports, fmt.Sprintf("%d/http", p.ContainerPort)) + } + return ports +} + +// proxyURL is RunPod's HTTP proxy address for a Pod's first declared port. It is derived, +// not read back: the form is fixed, so the URL is known at create time and survives a +// manager restart without an API call. +// +// Empty when the container declares no port — there is then no port to route to, and a +// guess would publish an endpoint that answers nothing. +func proxyURL(id string, ports []string) string { + if id == "" || len(ports) == 0 { + return "" + } + port, _, _ := strings.Cut(ports[0], "/") + return fmt.Sprintf("https://%s-%s.proxy.runpod.net", id, port) +} + +// resourceQty returns the container's request for name, falling back to its limit, or nil +// when neither is present. Requests first because that is the floor the workload declared; +// RunPod has no separate ceiling to set, so limits are only a fallback source of a number. +func resourceQty(c *corev1.Container, name corev1.ResourceName) *resource.Quantity { + if q, ok := c.Resources.Requests[name]; ok { + return &q + } + if q, ok := c.Resources.Limits[name]; ok { + return &q + } + return nil +} + +// cores converts a CPU quantity to whole vCPUs, RunPod's unit, rounding UP: a request of +// 500m is one vCPU, and 1500m is two. Rounding down would hand the workload less CPU than +// it asked for, and a fractional vCPU is not something RunPod can express. Nil (unset) is 0, +// which leaves RunPod's own default. +func cores(q *resource.Quantity) int { + if q == nil { + return 0 + } + return ceilDiv(int(q.MilliValue()), 1000) +} + +// gib converts a memory quantity to whole GiB, RunPod's unit, rounding up as cores does and +// for the same reason. Nil is 0. +func gib(q *resource.Quantity) int { + if q == nil { + return 0 + } + const giB = 1024 * 1024 * 1024 + return ceilDiv(int(q.Value()), giB) +} + +// ephemeralGiB reads the container's ephemeral-storage request as the container disk size. +// Zero (unset) leaves RunPod's 50 GiB default, which is generous enough that most workloads +// never need to state one. +func ephemeralGiB(c *corev1.Container) int { + return gib(resourceQty(c, corev1.ResourceEphemeralStorage)) +} + +// perGPU divides a Pod-wide total by the accelerator count, rounding up, because RunPod +// sizes cpu and memory PER GPU. Rounding up keeps the total at or above what the Pod asked +// for; rounding down would under-provision every request that does not divide evenly. +// +// A zero total stays zero (unset → RunPod's default), and a zero count is treated as one so +// a malformed request cannot divide by zero. +func perGPU(total int, count int32) int { + if total <= 0 { + return 0 + } + if count <= 0 { + return total + } + return ceilDiv(total, int(count)) +} + +// ceilDiv divides rounding away from zero for positive inputs. Its own function because +// every conversion above rounds the same way, and an inlined `(a+b-1)/b` is easy to get +// subtly wrong once. +func ceilDiv(a, b int) int { + if a <= 0 || b <= 0 { + return 0 + } + return (a + b - 1) / b +} + +// RunPod's desiredStatus values, the only three the API documents. +const ( + // statusRunning: RunPod WANTS this Pod running. It says nothing about whether the + // container is up yet, which is why toState pairs it with LastStartedAt. + statusRunning = "RUNNING" + // statusExited: the container exited. RunPod does not distinguish a clean exit from a + // crash here, so it maps to Terminated — "gone", with no claim about why. + statusExited = "EXITED" + // statusTerminated: the Pod was destroyed (our own Terminate, or a spot reclaim). + statusTerminated = "TERMINATED" +) + +// toState maps an observed Pod to the provider-agnostic lifecycle state. +// +// The subtlety is that desiredStatus is DESIRED, not observed: RunPod reports RUNNING from +// the moment it accepts the Pod, while the image may still be pulling. Reporting Running +// then would advance the Pod — and its Deployment's ready replicas — before anything is +// listening. LastStartedAt is the one field that only appears once the container has +// actually started, so it is the gate. Same shape as AWS holding an instance at Pending +// until its status checks clear. +// +// Everything unrecognized falls to Pending, so a status this adapter has not seen keeps the +// poll loop watching rather than going terminal on a live, billing Pod. +func toState(pd Pod) provider.InstanceState { + switch strings.ToUpper(pd.DesiredStatus) { + case statusRunning: + if pd.LastStartedAt == "" { + return provider.InstancePending + } + return provider.InstanceRunning + case statusExited, statusTerminated: + return provider.InstanceTerminated + default: + return provider.InstancePending + } +} + +// toInstance normalizes an observed RunPod Pod into the provider-agnostic Instance. +// +// Endpoint prefers the public IP and mapped port when RunPod has assigned them, because +// that is the address that reaches a Pod directly; the derived proxy URL is the fallback, +// and the only address a /http-only Pod ever has. Either way it is re-reported on every +// tick, and an empty value never clears what is already on the Pod (the write paths skip ""). +func toInstance(pd Pod) provider.Instance { + tier := nebulav1alpha1.CapacityOnDemand + if pd.Interruptible { + tier = nebulav1alpha1.CapacitySpot + } + return provider.Instance{ + ID: pd.ID, + ClaimName: claimFromName(pd.Name), + State: toState(pd), + CapacityType: tier, + Region: pd.DataCenterID, + Endpoint: endpointOf(pd), + } +} + +// endpointOf renders the Pod's reachable address: host:port from the public IP and its +// first port mapping, else the derived HTTP proxy URL, else empty while the Pod is still +// coming up. +// +// The mapping is walked in sorted key order because Go randomizes map iteration and this +// value is written to the Pod: an unsorted pick would rewrite the endpoint on alternating +// poll ticks for a Pod exposing two ports, which reads as flapping. +func endpointOf(pd Pod) string { + if pd.PublicIP != "" && len(pd.PortMappings) > 0 { + keys := make([]string, 0, len(pd.PortMappings)) + for k := range pd.PortMappings { + keys = append(keys, k) + } + sort.Strings(keys) + return fmt.Sprintf("%s:%d", pd.PublicIP, pd.PortMappings[keys[0]]) + } + return proxyURL(pd.ID, pd.Ports) +} diff --git a/pkg/provider/runpod/runpod_test.go b/pkg/provider/runpod/runpod_test.go new file mode 100644 index 0000000..1787419 --- /dev/null +++ b/pkg/provider/runpod/runpod_test.go @@ -0,0 +1,750 @@ +/* +Copyright 2026 The InftyAI Team. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package runpod + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + nebulav1alpha1 "github.com/InftyAI/Nebula/api/v1alpha1" + "github.com/InftyAI/Nebula/pkg/provider" + "github.com/InftyAI/Nebula/pkg/util" +) + +// fakeClient is an in-memory Client. It records the last CreatePod spec — the thing the +// adapter actually decides — and lets a test seed existing Pods or inject an error. +type fakeClient struct { + pods []Pod + lastSpec PodSpec + createCnt int + createErr error + createID string + + terminated []string + + // authID is what EnsureRegistryAuth resolves to, authFor the credential it was handed, + // and authCnt how many times it was called — Provision must not pay for it when the + // spec was going to be refused anyway. + authID string + authErr error + authFor *provider.RegistryAuth + authCnt int +} + +func (f *fakeClient) CreatePod(_ context.Context, spec PodSpec) (string, error) { + f.createCnt++ + f.lastSpec = spec + if f.createErr != nil { + return "", f.createErr + } + id := f.createID + if id == "" { + id = "pod-new" + } + f.pods = append(f.pods, Pod{ID: id, Name: spec.Name, DesiredStatus: statusRunning}) + return id, nil +} + +func (f *fakeClient) TerminatePod(_ context.Context, id string) error { + f.terminated = append(f.terminated, id) + return nil +} + +func (f *fakeClient) GetPod(_ context.Context, id string) (*Pod, error) { + for i := range f.pods { + if f.pods[i].ID == id { + pd := f.pods[i] + return &pd, nil + } + } + return nil, nil +} + +func (f *fakeClient) ListPods(_ context.Context) ([]Pod, error) { return f.pods, nil } + +func (f *fakeClient) EnsureRegistryAuth(_ context.Context, a *provider.RegistryAuth) (string, error) { + f.authCnt++ + f.authFor = a + if f.authErr != nil { + return "", f.authErr + } + return f.authID, nil +} + +// fakeCatalog is a trivial catalog.Lookup. +type fakeCatalog struct{ rows []provider.Offering } + +func (c fakeCatalog) Offerings(_ string) []provider.Offering { return c.rows } + +// newTestProvider builds a Provider over a fake client and a catalog shaped like +// runpod.csv: H100 carries THREE interchangeable RunPod ids (so MapAccelerator's +// primary-then-alternates order is observable), A100-80GB one, and L4 a Spot row. +func newTestProvider(f *fakeClient) *Provider { + od, spot := nebulav1alpha1.CapacityOnDemand, nebulav1alpha1.CapacitySpot + row := func(typ, id string, tier nebulav1alpha1.CapacityType, price float64) provider.Offering { + return provider.Offering{ + AcceleratorType: typ, AcceleratorID: id, CapacityType: tier, + PricePerHour: price, Available: true, + } + } + return New(f, fakeCatalog{rows: []provider.Offering{ + row("H100", "NVIDIA H100 80GB HBM3", od, 2.99), + row("H100", "NVIDIA H100 NVL", od, 2.79), + row("H100", "NVIDIA H100 PCIe", od, 2.39), + row("A100-80GB", "NVIDIA A100-SXM4-80GB", od, 1.74), + row("L4", "NVIDIA L4", spot, 0.22), + }}) +} + +// gpuPod builds a Pod whose accelerator type rides on the label and whose count rides on +// the container's nvidia.com/gpu limit. count<=0 means CPU-only (neither is set). accel is +// passed through verbatim so a test can also exercise non-canonical casing. +func gpuPod(accel string, count int64) *corev1.Pod { + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "default"}, + Spec: corev1.PodSpec{Containers: []corev1.Container{{ + Name: "main", + Image: "myimg:latest", + Command: []string{"/entry.sh"}, + Args: []string{"--flag", "v"}, + }}}, + } + if accel != "" && count > 0 { + pod.Labels = map[string]string{nebulav1alpha1.AcceleratorTypeLabel: accel} + pod.Spec.Containers[0].Resources.Limits = corev1.ResourceList{ + util.NvidiaGPUResource: *resource.NewQuantity(count, resource.DecimalSI), + } + } + return pod +} + +// requests sets the container's resource requests from a k8s-notation map. +func requests(pod *corev1.Pod, m map[corev1.ResourceName]string) *corev1.Pod { + c := &pod.Spec.Containers[0] + if c.Resources.Requests == nil { + c.Resources.Requests = corev1.ResourceList{} + } + for k, v := range m { + c.Resources.Requests[k] = resource.MustParse(v) + } + return pod +} + +// compile-time check that the fake really satisfies the seam the adapter is written to. +var _ Client = (*fakeClient)(nil) + +func TestProvision_GPUPod(t *testing.T) { + f := &fakeClient{createID: "pod-1"} + p := newTestProvider(f) + + pod := requests(gpuPod("H100", 2), map[corev1.ResourceName]string{ + corev1.ResourceCPU: "9", + corev1.ResourceMemory: "100Gi", + corev1.ResourceEphemeralStorage: "80Gi", + }) + pod.Spec.Containers[0].Ports = []corev1.ContainerPort{{ContainerPort: 8000}, {ContainerPort: 9090}} + + res, err := p.Provision(context.Background(), pod, provider.ProvisionRequest{ + ClaimName: "claim-a", + CapacityType: nebulav1alpha1.CapacityOnDemand, + Region: "US-KS-2", + Env: map[string]string{"HF_TOKEN": "hf_secret"}, + }) + if err != nil { + t.Fatalf("Provision: %v", err) + } + // A RunPod create allocates a machine before it answers — a shortage comes back as an + // error, not a queued Pod — so an id here means real capacity was reserved. This is the + // one place RunPod differs from Modal, and getting it wrong would report capacity that + // was never granted. + if !res.Reserved { + t.Error("Reserved = false; a successful RunPod create means a host was allocated") + } + if res.InstanceID != "pod-1" { + t.Errorf("InstanceID = %q, want pod-1", res.InstanceID) + } + // Derived from the id and the FIRST declared port, so it needs no read-back. + if want := "https://pod-1-8000.proxy.runpod.net"; res.ConnectURL != want { + t.Errorf("ConnectURL = %q, want %q", res.ConnectURL, want) + } + // RunPod's HTTP proxy is unauthenticated: there is no credential to hand back, and a + // placeholder would look like one the Pod could authenticate with. + if res.ConnectToken != "" { + t.Errorf("ConnectToken = %q, want empty (the proxy is unauthenticated)", res.ConnectToken) + } + + s := f.lastSpec + if s.Name != "nebula-claim-a" { + t.Errorf("Name = %q, want nebula-claim-a", s.Name) + } + if s.Image != "myimg:latest" { + t.Errorf("Image = %q", s.Image) + } + // command → ENTRYPOINT and args → CMD stay SEPARATE, unlike Modal where both + // concatenate into one command. + if strings.Join(s.Entrypoint, " ") != "/entry.sh" || strings.Join(s.StartCmd, " ") != "--flag v" { + t.Errorf("Entrypoint = %v, StartCmd = %v", s.Entrypoint, s.StartCmd) + } + if s.Env["HF_TOKEN"] != "hf_secret" { + t.Errorf("Env = %v; the caller's resolved env must go out whole", provider.RedactedEnv(s.Env)) + } + // All three H100 ids ride one create, primary first: RunPod picks from the array by + // availability, so alternates broaden a SINGLE launch. + if len(s.GPUTypeIDs) != 3 || s.GPUTypeIDs[0] != "NVIDIA H100 80GB HBM3" { + t.Errorf("GPUTypeIDs = %v, want all three H100 ids with the primary first", s.GPUTypeIDs) + } + if s.GPUCount != 2 { + t.Errorf("GPUCount = %d, want 2", s.GPUCount) + } + // RunPod sizes cpu/memory PER GPU, so the Pod's totals divide by 2 and round UP: + // 9 vCPU → 5, 100 GiB → 50. Rounding down would under-provision the request. + if s.VCPUPerGPU != 5 || s.RAMPerGPUGiB != 50 { + t.Errorf("VCPUPerGPU = %d, RAMPerGPUGiB = %d, want 5/50 (per-GPU, rounded up)", + s.VCPUPerGPU, s.RAMPerGPUGiB) + } + if s.VCPUCount != 0 { + t.Errorf("VCPUCount = %d, want 0; it is only read for a CPU-only Pod", s.VCPUCount) + } + if s.ContainerDiskGiB != 80 { + t.Errorf("ContainerDiskGiB = %d, want 80", s.ContainerDiskGiB) + } + if strings.Join(s.Ports, ",") != "8000/http,9090/http" { + t.Errorf("Ports = %v, want both as /http", s.Ports) + } + // A compound token is a data center; the country-code field stays empty. + if len(s.DataCenterIDs) != 1 || s.DataCenterIDs[0] != "US-KS-2" || len(s.CountryCodes) != 0 { + t.Errorf("DataCenterIDs = %v, CountryCodes = %v", s.DataCenterIDs, s.CountryCodes) + } + if s.Interruptible { + t.Error("Interruptible = true on the OnDemand tier") + } +} + +func TestProvision_SpotAndCountryRegion(t *testing.T) { + f := &fakeClient{createID: "pod-spot"} + p := newTestProvider(f) + + if _, err := p.Provision(context.Background(), gpuPod("l4", 1), provider.ProvisionRequest{ + ClaimName: "claim-s", + CapacityType: nebulav1alpha1.CapacitySpot, + Region: "us", + }); err != nil { + t.Fatalf("Provision: %v", err) + } + s := f.lastSpec + if !s.Interruptible { + t.Error("Interruptible = false on the Spot tier; RunPod's spot tier is this one boolean") + } + // A bare two-letter token is an ISO country code RunPod takes natively, upper-cased — + // so nothing has to expand it and one declared token stays one blocklistable candidate. + if len(s.CountryCodes) != 1 || s.CountryCodes[0] != "US" || len(s.DataCenterIDs) != 0 { + t.Errorf("CountryCodes = %v, DataCenterIDs = %v, want [US]/[]", s.CountryCodes, s.DataCenterIDs) + } + // The label's casing is the user's; the catalog id that goes out is not. + if len(s.GPUTypeIDs) != 1 || s.GPUTypeIDs[0] != "NVIDIA L4" { + t.Errorf("GPUTypeIDs = %v, want [NVIDIA L4] from a lowercase label", s.GPUTypeIDs) + } +} + +func TestProvision_CPUOnlyPod(t *testing.T) { + f := &fakeClient{createID: "pod-cpu"} + p := newTestProvider(f) + + pod := requests(gpuPod("", 0), map[corev1.ResourceName]string{corev1.ResourceCPU: "2500m"}) + if _, err := p.Provision(context.Background(), pod, provider.ProvisionRequest{ + ClaimName: "claim-c", + CapacityType: nebulav1alpha1.CapacityOnDemand, + }); err != nil { + t.Fatalf("Provision: %v", err) + } + s := f.lastSpec + // No accelerator: RunPod sizes the Pod with an ABSOLUTE vCPU count (rounded up from + // 2500m), not the per-GPU pair, and no GPU fields are set at all. + if s.VCPUCount != 3 || s.VCPUPerGPU != 0 || s.GPUCount != 0 || len(s.GPUTypeIDs) != 0 { + t.Errorf("VCPUCount = %d, VCPUPerGPU = %d, GPUCount = %d, GPUTypeIDs = %v", + s.VCPUCount, s.VCPUPerGPU, s.GPUCount, s.GPUTypeIDs) + } + // No region declared leaves both placement fields empty — the widest capacity pool. + if len(s.DataCenterIDs) != 0 || len(s.CountryCodes) != 0 { + t.Errorf("region fields = %v/%v, want both empty when unconstrained", + s.DataCenterIDs, s.CountryCodes) + } +} + +func TestProvision_Idempotent(t *testing.T) { + // A Pod already carrying this claim's name is the ONLY record of ownership RunPod + // offers, so a repeat after a partial create must find it rather than pay twice. + f := &fakeClient{pods: []Pod{{ + ID: "pod-existing", Name: "nebula-claim-a", DesiredStatus: statusRunning, + LastStartedAt: "2026-08-29T00:00:00Z", + }}} + p := newTestProvider(f) + + res, err := p.Provision(context.Background(), gpuPod("H100", 1), provider.ProvisionRequest{ + ClaimName: "claim-a", + CapacityType: nebulav1alpha1.CapacityOnDemand, + }) + if err != nil { + t.Fatalf("Provision: %v", err) + } + if res.InstanceID != "pod-existing" { + t.Errorf("InstanceID = %q, want the existing pod-existing", res.InstanceID) + } + if !res.Reserved { + t.Error("Reserved = false; the Pod exists, so a machine was allocated") + } + if f.createCnt != 0 { + t.Errorf("CreatePod called %d times; a second Pod would be billed twice", f.createCnt) + } + // The interface forbids re-minting a credential on a repeat, and there is nothing to + // mint here anyway — the proxy URL is derivable from the id. + if res.ConnectToken != "" { + t.Errorf("ConnectToken = %q, want empty on a re-Provision", res.ConnectToken) + } +} + +func TestProvision_RefusesOverlongClaimName(t *testing.T) { + // RunPod caps a pod name at 191 chars and the name is Nebula's ONLY carrier of + // identity, so a name that does not fit is refused rather than truncated: two + // truncated claims would collide onto one Pod, which bills one twice and lets the + // other's teardown reap the survivor. + f := &fakeClient{} + p := newTestProvider(f) + + long := strings.Repeat("a", maxNameLen-len(namePrefix)+1) + _, err := p.Provision(context.Background(), gpuPod("H100", 1), provider.ProvisionRequest{ + ClaimName: long, + CapacityType: nebulav1alpha1.CapacityOnDemand, + }) + if err == nil { + t.Fatal("Provision succeeded with an over-long claim name; the name would have collided") + } + if f.createCnt != 0 { + t.Errorf("CreatePod called %d times despite the refusal", f.createCnt) + } + + // One char shorter fits exactly, so the boundary is not off by one. + if _, err := podName(strings.Repeat("a", maxNameLen-len(namePrefix))); err != nil { + t.Errorf("podName rejected a name that fits exactly: %v", err) + } +} + +func TestProvision_RefusesRestrictedEgress(t *testing.T) { + // RunPod has no outbound-allowlist knob at all. Placement should never route such a + // pool here, but a request can be built by anyone, and provisioning it anyway would + // put the workload on the open internet under a policy that says otherwise. + f := &fakeClient{} + p := newTestProvider(f) + + _, err := p.Provision(context.Background(), gpuPod("H100", 1), provider.ProvisionRequest{ + ClaimName: "claim-e", + CapacityType: nebulav1alpha1.CapacityOnDemand, + Egress: &nebulav1alpha1.EgressPolicy{Mode: nebulav1alpha1.EgressBlocked}, + }) + if err == nil { + t.Fatal("Provision accepted a restricted egress policy RunPod cannot enforce") + } + if f.createCnt != 0 { + t.Errorf("CreatePod called %d times despite the refusal", f.createCnt) + } +} + +func TestProvision_RegistryAuth(t *testing.T) { + basic := &provider.RegistryAuth{ + Registry: "ghcr.io", + Basic: &provider.BasicAuth{Username: "u", Password: "p4ssw0rd"}, + } + + t.Run("basic resolves to an auth id", func(t *testing.T) { + f := &fakeClient{createID: "pod-auth", authID: "cra-123"} + p := newTestProvider(f) + + if _, err := p.Provision(context.Background(), gpuPod("H100", 1), provider.ProvisionRequest{ + ClaimName: "claim-r", + CapacityType: nebulav1alpha1.CapacityOnDemand, + RegistryAuth: basic, + }); err != nil { + t.Fatalf("Provision: %v", err) + } + // RunPod's create takes an OBJECT ID, never an inline username/password, so the + // indirection has to happen before the create. + if f.lastSpec.RegistryAuthID != "cra-123" { + t.Errorf("RegistryAuthID = %q, want cra-123", f.lastSpec.RegistryAuthID) + } + if f.authCnt != 1 { + t.Errorf("EnsureRegistryAuth called %d times, want 1", f.authCnt) + } + }) + + t.Run("aws role is refused without an API call", func(t *testing.T) { + // An ECR "password" is a 12-hour token, so there is no honest way to flatten a role + // into RunPod's static credential object: a Pod pulling from it would stop being able + // to pull halfway through the day. Refuse, and never fall back to an anonymous pull. + f := &fakeClient{} + p := newTestProvider(f) + + _, err := p.Provision(context.Background(), gpuPod("H100", 1), provider.ProvisionRequest{ + ClaimName: "claim-ecr", + CapacityType: nebulav1alpha1.CapacityOnDemand, + RegistryAuth: &provider.RegistryAuth{ + Registry: "1234.dkr.ecr.us-east-1.amazonaws.com", + AWSRole: &provider.AWSRoleAuth{RoleARN: "arn:aws:iam::1234:role/pull", Region: "us-east-1"}, + }, + }) + if err == nil { + t.Fatal("Provision accepted an AWSRole credential RunPod has no equivalent for") + } + // A rejection of the REQUEST, not of the candidate: the Pod fails with the reason + // instead of retrying, and nothing gets blocklisted. + if !errors.Is(err, provider.ErrImagePull) { + t.Errorf("error = %v, want it to wrap ErrImagePull", err) + } + if f.authCnt != 0 || f.createCnt != 0 { + t.Errorf("authCnt = %d, createCnt = %d; a refused credential must cost no API calls", + f.authCnt, f.createCnt) + } + }) +} + +func TestProvision_UnsupportedAccelerator(t *testing.T) { + // A type the catalog has no RunPod id for cannot be launched, and the sentinel is what + // turns this into a capacity-class block rather than nebula_provision_failures_total + // {reason="other"}. + f := &fakeClient{} + p := newTestProvider(f) + + _, err := p.Provision(context.Background(), gpuPod("TPUv5", 1), provider.ProvisionRequest{ + ClaimName: "claim-x", + CapacityType: nebulav1alpha1.CapacityOnDemand, + }) + if !errors.Is(err, provider.ErrUnsupportedAccelerator) { + t.Fatalf("error = %v, want it to wrap ErrUnsupportedAccelerator", err) + } + if f.createCnt != 0 { + t.Errorf("CreatePod called %d times for an accelerator with no RunPod id", f.createCnt) + } +} + +func TestPodSpecString_Redacts(t *testing.T) { + // A spec reaches logs and error strings, and Env holds everything envFrom/valueFrom + // resolved — Secret values included. Key NAMES are already in the Pod spec, so they + // may print; values never may. + s := PodSpec{ + Name: "nebula-claim-a", + Image: "myimg:latest", + Env: map[string]string{"HF_TOKEN": "hf_supersecret", "PLAIN": "visible"}, + } + for _, form := range []string{s.String(), fmt.Sprintf("%v", s), fmt.Sprintf("%#v", s)} { + if strings.Contains(form, "hf_supersecret") || strings.Contains(form, "visible") { + t.Errorf("rendered spec leaks an env VALUE: %s", form) + } + if !strings.Contains(form, "HF_TOKEN") { + t.Errorf("rendered spec dropped the env key names, which are safe: %s", form) + } + } +} + +func TestList_FiltersToNebulaPods(t *testing.T) { + // The name prefix stands in for the tags RunPod does not have. A Pod without it belongs + // to someone else in the same account: reporting it would have the poll loop adopt it + // and the NodeClaim controller eventually TERMINATE it. + f := &fakeClient{pods: []Pod{ + {ID: "pod-1", Name: "nebula-claim-a", DesiredStatus: statusRunning, LastStartedAt: "t"}, + {ID: "pod-2", Name: "my-own-dev-box", DesiredStatus: statusRunning, LastStartedAt: "t"}, + {ID: "pod-3", Name: "nebula-claim-b", DesiredStatus: statusExited}, + }} + p := newTestProvider(f) + + got, err := p.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(got) != 2 { + t.Fatalf("List returned %d instances, want the 2 Nebula-owned ones: %+v", len(got), got) + } + // ClaimName is recovered by stripping the prefix — the tag read the naming scheme + // stands in for. + if got[0].ClaimName != "claim-a" || got[1].ClaimName != "claim-b" { + t.Errorf("claim names = %q/%q, want claim-a/claim-b", got[0].ClaimName, got[1].ClaimName) + } + if got[0].State != provider.InstanceRunning || got[1].State != provider.InstanceTerminated { + t.Errorf("states = %v/%v", got[0].State, got[1].State) + } +} + +func TestGetAndTerminate(t *testing.T) { + f := &fakeClient{pods: []Pod{{ + ID: "pod-1", Name: "nebula-claim-a", DesiredStatus: statusRunning, + LastStartedAt: "t", Interruptible: true, DataCenterID: "EU-RO-1", + Ports: []string{"8000/http"}, + }}} + p := newTestProvider(f) + + inst, err := p.Get(context.Background(), "pod-1", "") + if err != nil { + t.Fatalf("Get: %v", err) + } + if inst == nil { + t.Fatal("Get returned nil for a live Pod") + } + // The observed instance reports the tier it actually GOT and where RunPod actually put + // it, both of which can differ from what was asked for. + if inst.CapacityType != nebulav1alpha1.CapacitySpot || inst.Region != "EU-RO-1" { + t.Errorf("CapacityType = %q, Region = %q", inst.CapacityType, inst.Region) + } + // An /http-only Pod has no public IP or port mapping ever, so the derived proxy URL is + // its only address. + if want := "https://pod-1-8000.proxy.runpod.net"; inst.Endpoint != want { + t.Errorf("Endpoint = %q, want %q", inst.Endpoint, want) + } + + // A Pod that is gone reports (nil, nil) — absent means terminated, per the interface. + gone, err := p.Get(context.Background(), "pod-missing", "") + if err != nil || gone != nil { + t.Errorf("Get(missing) = %v, %v; want nil, nil", gone, err) + } + + if err := p.Terminate(context.Background(), "pod-1", "EU-RO-1"); err != nil { + t.Fatalf("Terminate: %v", err) + } + if len(f.terminated) != 1 || f.terminated[0] != "pod-1" { + t.Errorf("terminated = %v, want [pod-1]", f.terminated) + } + // Nothing was ever provisioned: there is no id to delete and no call to make. + if err := p.Terminate(context.Background(), "", ""); err != nil { + t.Errorf("Terminate(\"\") = %v, want nil", err) + } + if len(f.terminated) != 1 { + t.Errorf("Terminate(\"\") called the API: %v", f.terminated) + } +} + +func TestToState(t *testing.T) { + cases := []struct { + name string + pod Pod + want provider.InstanceState + }{{ + // The subtlety of the whole adapter: desiredStatus is DESIRED. RunPod says RUNNING + // from the moment it accepts the Pod, while the image may still be pulling — + // reporting Running then would mark a Deployment's replica ready before anything is + // listening. LastStartedAt is the one field that only appears once the container has + // actually started. + name: "RUNNING without LastStartedAt is still Pending", + pod: Pod{DesiredStatus: "RUNNING"}, + want: provider.InstancePending, + }, { + name: "RUNNING with LastStartedAt is Running", + pod: Pod{DesiredStatus: "RUNNING", LastStartedAt: "2026-08-29T00:00:00Z"}, + want: provider.InstanceRunning, + }, { + name: "EXITED is Terminated", + pod: Pod{DesiredStatus: "EXITED", LastStartedAt: "t"}, + want: provider.InstanceTerminated, + }, { + // Our own Terminate, or a spot reclaim — indistinguishable here, and both mean gone. + name: "TERMINATED is Terminated", + pod: Pod{DesiredStatus: "TERMINATED"}, + want: provider.InstanceTerminated, + }, { + name: "casing is not load-bearing", + pod: Pod{DesiredStatus: "running", LastStartedAt: "t"}, + want: provider.InstanceRunning, + }, { + // A status this adapter has never seen keeps the poll loop WATCHING rather than + // going terminal on a live, billing Pod. + name: "an unknown status is Pending, not Terminated", + pod: Pod{DesiredStatus: "SOMETHING_NEW"}, + want: provider.InstancePending, + }, { + name: "an empty status is Pending", + pod: Pod{}, + want: provider.InstancePending, + }} + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := toState(tc.pod); got != tc.want { + t.Errorf("toState = %v, want %v", got, tc.want) + } + }) + } +} + +func TestEndpointOf_PrefersDirectAddress(t *testing.T) { + // A public IP with an assigned port reaches the Pod directly, so it wins over the + // proxy. The mapping is walked in sorted key order because Go randomizes map iteration + // and this value is written to the Pod: an unsorted pick would rewrite the endpoint on + // alternating poll ticks, which reads as flapping. + pd := Pod{ + ID: "pod-1", + Ports: []string{"8000/http"}, + PublicIP: "1.2.3.4", + PortMappings: map[string]int{"22": 40022, "8000": 41234}, + } + for range 8 { + if got := endpointOf(pd); got != "1.2.3.4:40022" { + t.Fatalf("endpointOf = %q, want the lowest-keyed mapping 1.2.3.4:40022", got) + } + } + // No port at all: no address to publish, and a guess would advertise something that + // answers nothing. + if got := endpointOf(Pod{ID: "pod-2"}); got != "" { + t.Errorf("endpointOf(no ports) = %q, want empty", got) + } +} + +func TestClassifyProvisionError(t *testing.T) { + p := newTestProvider(&fakeClient{}) + const accel, region = "H100:8", "EU-RO-1" + + t.Run("nil error blocks nothing", func(t *testing.T) { + // ClassifyError already returns the zero scope here, but the region decoration + // below would repopulate it into a scope recordBlock installs. + if got := p.ClassifyProvisionError(nil, accel, region); got != (provider.BlockScope{}) { + t.Errorf("scope = %+v, want the zero scope", got) + } + }) + + t.Run("capacity is scoped to this accelerator, tier and region", func(t *testing.T) { + err := fmt.Errorf("no instances available: %w", provider.ErrNoCapacity) + got := p.ClassifyProvisionError(err, accel, region) + if got.DenyAll { + t.Error("DenyAll = true for a capacity shortage; only this candidate ran out") + } + if got.Accelerator == nil || *got.Accelerator != accel { + t.Errorf("Accelerator = %v, want %q", got.Accelerator, accel) + } + if got.CapacityType != nebulav1alpha1.CapacityOnDemand { + t.Errorf("CapacityType = %q, want OnDemand", got.CapacityType) + } + if got.Region == nil || *got.Region != region { + t.Errorf("Region = %v, want %q — a shortage in one DC must not disqualify another", + got.Region, region) + } + }) + + t.Run("a spot shortage does not block OnDemand", func(t *testing.T) { + // The failing tier is not on the error's face, so the Client marks an interruptible + // shortage with ErrSpotCapacity. Blocking it as OnDemand would disable capacity that + // is still purchasable at the higher price. + err := fmt.Errorf("spot gone: %w: %w", provider.ErrNoCapacity, ErrSpotCapacity) + got := p.ClassifyProvisionError(err, accel, region) + if got.CapacityType != nebulav1alpha1.CapacitySpot { + t.Errorf("CapacityType = %q, want Spot", got.CapacityType) + } + }) + + t.Run("auth denies the whole provider and is not narrowed", func(t *testing.T) { + // A bad API key fails in every region, so narrowing DenyAll to one would contradict + // the category and keep trying the other candidates against the same dead key. + err := fmt.Errorf("401: %w", provider.ErrAuth) + got := p.ClassifyProvisionError(err, accel, region) + if !got.DenyAll { + t.Fatal("DenyAll = false for an auth failure") + } + if got.Region != nil { + t.Errorf("Region = %v on a DenyAll scope, want nil", got.Region) + } + }) + + t.Run("a request rejection is not decorated into a block", func(t *testing.T) { + // An unusable pull credential says nothing about the CANDIDATE. Stamping a region + // onto the zero scope would make it non-empty, and recordBlock would then install a + // region-wide block across every accelerator — excluding that DC for every other Pod + // until the TTL lapsed. + err := fmt.Errorf("bad credential: %w", provider.ErrImagePull) + if got := p.ClassifyProvisionError(err, accel, region); got != (provider.BlockScope{}) { + t.Errorf("scope = %+v, want the zero scope so nothing is blocklisted", got) + } + }) + + t.Run("an empty region leaves Region nil", func(t *testing.T) { + // nil matches only candidates that carry no region either — the unconstrained pool — + // so the block never leaks onto region-pinned candidates. + err := fmt.Errorf("no capacity: %w", provider.ErrNoCapacity) + if got := p.ClassifyProvisionError(err, accel, ""); got.Region != nil { + t.Errorf("Region = %v, want nil", got.Region) + } + }) +} + +func TestCapabilities(t *testing.T) { + c := newTestProvider(&fakeClient{}).Capabilities() + // Spot is real here (`interruptible`), which is what makes the catalog's Spot rows and + // the ErrSpotCapacity marker meaningful — RunPod is the first adapter where both matter. + if !c.SupportsSpot { + t.Error("SupportsSpot = false") + } + // A stopped RunPod Pod still bills for its disk and releases its GPU, so it is neither + // free nor resumable in the sense the capability promises. + if c.SupportsStop { + t.Error("SupportsStop = true; a stopped Pod still bills and has lost its GPU") + } + // No outbound-allowlist knob exists in the API, so placement must skip RunPod for an + // egress-restricted pool rather than have Provision refuse it after the fact. + if c.SupportsEgressPolicy { + t.Error("SupportsEgressPolicy = true; RunPod exposes no outbound policy") + } + // No tags: identity rides the Pod name, which is what List filters on. + if c.NativeTags { + t.Error("NativeTags = true; RunPod Pods have no tags") + } + // Reclaims arrive with no notice pushed to us, so polling is the only detector — hence + // a zero notice window and a faster-than-default cadence. + if c.PreemptionNotice != 0 { + t.Errorf("PreemptionNotice = %v, want 0 (abrupt)", c.PreemptionNotice) + } + if c.PollInterval != spotPollInterval { + t.Errorf("PollInterval = %v, want %v", c.PollInterval, spotPollInterval) + } +} + +func TestSplitRegion(t *testing.T) { + // The split is by SHAPE, not by a lookup table: RunPod's data-center ids are compound + // ("EU-RO-1") while a bare two-letter token is an ISO country code its countryCodes + // field takes. That is what keeps a static data-center list — which rots silently — out + // of this package, and why ExpandRegions stays catalog.Base's pass-through. + cases := []struct { + region string + wantDCs []string + wantCodes []string + }{ + {region: "", wantDCs: nil, wantCodes: nil}, + {region: " ", wantDCs: nil, wantCodes: nil}, + {region: "us", wantDCs: nil, wantCodes: []string{"US"}}, + {region: "SE", wantDCs: nil, wantCodes: []string{"SE"}}, + {region: "US-KS-2", wantDCs: []string{"US-KS-2"}, wantCodes: nil}, + {region: "EU-RO-1", wantDCs: []string{"EU-RO-1"}, wantCodes: nil}, + } + for _, tc := range cases { + t.Run(fmt.Sprintf("%q", tc.region), func(t *testing.T) { + dcs, codes := splitRegion(tc.region) + if strings.Join(dcs, ",") != strings.Join(tc.wantDCs, ",") || + strings.Join(codes, ",") != strings.Join(tc.wantCodes, ",") { + t.Errorf("splitRegion(%q) = %v, %v; want %v, %v", + tc.region, dcs, codes, tc.wantDCs, tc.wantCodes) + } + }) + } +}