From 40b9ed3cad96fc4d7e7172545d2c558be4d5de19 Mon Sep 17 00:00:00 2001 From: Nirupam Bhowmick <48842933+heydryft@users.noreply.github.com> Date: Sun, 25 Oct 2026 15:33:29 +0000 Subject: [PATCH 1/3] feat(clusters): runcrate clusters to create, use, test and delete GPU clusters (RUN-635) runcrate clusters list, offers, create, get, delete, ssh, kubeconfig, tunnel and test, on the /api/v1/clusters routes. - offers shows where a cluster can start, with the price per node and for the whole cluster as the API returns them; community hosts only with --include-community. - create sends an Idempotency-Key; a gone offer prints the current offers and creates nothing; no capacity says nothing was created or billed. - ssh and tunnel sign an ephemeral certificate per node instance id, the same flow as runcrate ssh (now execSSH, shared by both). tunnel forwards the Kubernetes API (6443) or the Ray dashboard (8265) from the head. - kubeconfig -o writes the file mode 600; test waits for the NCCL result unless -d. Refs: RUN-635 --- README.md | 25 ++ internal/api/clusters.go | 296 +++++++++++++ internal/api/clusters_test.go | 130 ++++++ internal/api/schedules.go | 8 +- internal/commands/clusters.go | 650 +++++++++++++++++++++++++++++ internal/commands/clusters_test.go | 138 ++++++ internal/commands/helpers.go | 4 +- internal/commands/root.go | 11 + internal/commands/ssh.go | 27 +- 9 files changed, 1275 insertions(+), 14 deletions(-) create mode 100644 internal/api/clusters.go create mode 100644 internal/api/clusters_test.go create mode 100644 internal/commands/clusters.go create mode 100644 internal/commands/clusters_test.go diff --git a/README.md b/README.md index 6532860..4c2cd55 100644 --- a/README.md +++ b/README.md @@ -134,6 +134,31 @@ runcrate schedule delete [--yes] # what it already started keeps running that key is revoked or the sign-in ends, the schedule pauses (last result `paused: key revoked`). +## Clusters + +A cluster is several GPU machines on one private network, with SSH between them, +`/shared` on every node, and optionally Slurm, Kubernetes or Ray installed across +them. It is all or nothing: every node starts or none does, and a cluster that could +not start costs nothing. Billing starts when every node is up and runs until you +delete the cluster, which stops every node together. + +```bash +runcrate clusters offers --gpu H100:8 --nodes 4 # where it can start, $/hr per node and for the cluster +runcrate clusters create --name train --offer --nodes 4 --stack slurm +runcrate clusters list [--all] [-q] [--json] +runcrate clusters get train # nodes, hostfile, torchrun / mpirun / sbatch, events +runcrate clusters ssh train [--node 2] [-- sinfo] # the head is node 0 +runcrate clusters kubeconfig train -o kubeconfig.yaml # Kubernetes clusters +runcrate clusters tunnel train # Kubernetes API on 6443, Ray dashboard on 8265 +runcrate clusters test train [-d] # NCCL all-reduce bandwidth across every GPU +runcrate clusters delete train [--yes] # terminates every node +``` + +- `--stack` is `none` (SSH, hostfile and torchrun only), `slurm`, `kubernetes` or `ray`; + `offers` lists the ones each offer can run. +- Offers are on datacenter hosts unless you pass `--include-community` (to both + `offers` and `create`). + ## Secrets Secrets are environment variables every instance and job launched in an diff --git a/internal/api/clusters.go b/internal/api/clusters.go new file mode 100644 index 0000000..ff8e852 --- /dev/null +++ b/internal/api/clusters.go @@ -0,0 +1,296 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/url" + "strconv" +) + +// ClusterInterconnect is how a cluster's nodes talk to each other. +type ClusterInterconnect struct { + Kind string `json:"kind"` // infiniband, roce or ethernet + RateGbps *float64 `json:"rate_gbps,omitempty"` + Label string `json:"label"` +} + +// ClusterNodeSpec is one node of an offer. +type ClusterNodeSpec struct { + VCPUs int `json:"vcpus"` + MemoryGB float64 `json:"memory_gb"` + StorageGB float64 `json:"storage_gb"` + VRAMGB *float64 `json:"vram_gb,omitempty"` +} + +// ClusterOffer is somewhere a cluster of a shape can be created now, from +// GET /api/v1/clusters/offers. HourlyRate is for the whole cluster at NodeCount. +type ClusterOffer struct { + OfferID string `json:"offer_id"` + GPUType string `json:"gpu_type"` + GPUsPerNode int `json:"gpus_per_node"` + NodeCount int `json:"node_count"` + MinNodes int `json:"min_nodes"` + MaxNodes int `json:"max_nodes"` + NodesAvailable *int `json:"nodes_available"` + DeploymentType string `json:"deployment_type"` + HostTier string `json:"host_tier"` + Location string `json:"location"` + Interconnect ClusterInterconnect `json:"interconnect"` + HourlyRatePerNode float64 `json:"hourly_rate_per_node"` + HourlyRate float64 `json:"hourly_rate"` + Stacks []string `json:"stacks"` + Resize bool `json:"resize"` + BootEstimateSec *int `json:"boot_estimate_sec,omitempty"` + Node ClusterNodeSpec `json:"node"` +} + +// ClusterOfferFilters are the query of GET /api/v1/clusters/offers; zero values are not sent. +type ClusterOfferFilters struct { + GPUType string + GPUsPerNode int + Nodes int + Region string + Stack string + IncludeCommunity bool +} + +// CreateClusterRequest is the body of POST /api/v1/clusters. +type CreateClusterRequest struct { + Name string `json:"name"` + OfferID string `json:"offer_id"` + NodeCount int `json:"node_count"` + Stack string `json:"stack"` + SSHKeyIDs []string `json:"ssh_key_ids,omitempty"` + StorageID string `json:"storage_id,omitempty"` + IncludeCommunity bool `json:"include_community,omitempty"` +} + +// Cluster is one row of GET /api/v1/clusters, and the top of the detail. +type Cluster struct { + ID string `json:"id"` + Name string `json:"name"` + Status string `json:"status"` + Stack string `json:"stack"` + StackStatus string `json:"stack_status"` + StackDetail *string `json:"stack_detail"` + NodeCount int `json:"node_count"` + GPUType string `json:"gpu_type"` + GPUsPerNode int `json:"gpus_per_node"` + DeploymentType string `json:"deployment_type"` + HostTier string `json:"host_tier"` + Location string `json:"location"` + Interconnect ClusterInterconnect `json:"interconnect"` + HourlyRatePerNode float64 `json:"hourly_rate_per_node"` + HourlyRate float64 `json:"hourly_rate"` + CreatedAt string `json:"created_at"` + UpAt *string `json:"up_at"` + ReadyAt *string `json:"ready_at"` + TerminatedAt *string `json:"terminated_at"` + FailureReason *string `json:"failure_reason,omitempty"` + FailureDetail *string `json:"failure_detail,omitempty"` +} + +// ClusterNode is one machine of a cluster. InstanceID works with every +// /api/v1/instances/{id} route (ssh-certificate, exec, upload, download). +type ClusterNode struct { + Rank int `json:"rank"` + Role string `json:"role"` // head (rank 0) or worker + InstanceID string `json:"instance_id"` + Name string `json:"name"` + Status string `json:"status"` + Stage string `json:"stage"` + OverlayAddress *string `json:"overlay_address"` + GPUs int `json:"gpus"` + IP *string `json:"ip"` + SSHCommand *string `json:"ssh_command"` +} + +// ClusterConnection is how to use a cluster: SSH to the head, the hostfile, +// the launcher environment and ready-made commands per stack. +type ClusterConnection struct { + HeadSSH *string `json:"head_ssh"` + Hostfile string `json:"hostfile"` + Env map[string]string `json:"env"` + Snippets map[string]string `json:"snippets"` +} + +// ClusterShared is the /shared folder every node mounts. +type ClusterShared struct { + Path string `json:"path"` + Available bool `json:"available"` + Detail string `json:"detail,omitempty"` + Volume *struct { + ID string `json:"id"` + Name string `json:"name"` + Path string `json:"path"` + } `json:"volume,omitempty"` +} + +// ClusterEvent is one line of a cluster's history. +type ClusterEvent struct { + At string `json:"at"` + Kind string `json:"kind"` + Rank *int `json:"rank"` + Message string `json:"message"` +} + +// NCCLSize is the all-reduce result at one message size. +type NCCLSize struct { + Bytes int64 `json:"bytes"` + AlgbwGbps float64 `json:"algbw_gbps"` + BusbwGbps float64 `json:"busbw_gbps"` + TimeUs float64 `json:"time_us"` +} + +// ClusterTest is an NCCL all-reduce test across every GPU of a cluster. +type ClusterTest struct { + ID string `json:"id"` + Kind string `json:"kind"` + Status string `json:"status"` // running, passed or failed + StartedAt string `json:"started_at"` + FinishedAt *string `json:"finished_at"` + Result *ClusterTestResult `json:"result"` +} + +// ClusterTestResult is a finished test: bandwidth per message size, or Error. +type ClusterTestResult struct { + Sizes []NCCLSize `json:"sizes"` + PeakBusbwGbps float64 `json:"peak_busbw_gbps"` + Nodes int `json:"nodes"` + GPUs int `json:"gpus"` + Error string `json:"error,omitempty"` +} + +// ClusterDetail is GET /api/v1/clusters/{id}. +type ClusterDetail struct { + Cluster + Nodes []ClusterNode `json:"nodes"` + Connection ClusterConnection `json:"connection"` + Shared ClusterShared `json:"shared"` + Events []ClusterEvent `json:"events"` + LastTest *ClusterTest `json:"last_test"` + Billing struct { + BilledSince *string `json:"billed_since"` + NodesBilled int `json:"nodes_billed"` + } `json:"billing"` +} + +// Kubeconfig is a Kubernetes cluster's kubeconfig and the SSH tunnel it needs: +// its server is https://127.0.0.1:6443. +type Kubeconfig struct { + YAML string + Tunnel string +} + +// ListClusters returns the environment's clusters, newest first: active ones, +// or every one with all. +func (c *Client) ListClusters(all bool) ([]Cluster, error) { + path := "/api/v1/clusters" + if all { + path += "?status=all" + } + list, err := callJSON[[]Cluster](c, "GET", path, nil) + if err != nil { + return nil, err + } + return *list, nil +} + +// ListClusterOffers returns where a cluster of this shape can be created now. +func (c *Client) ListClusterOffers(f ClusterOfferFilters) ([]ClusterOffer, error) { + q := url.Values{} + if f.GPUType != "" { + q.Set("gpu_type", f.GPUType) + } + if f.GPUsPerNode > 0 { + q.Set("gpus_per_node", strconv.Itoa(f.GPUsPerNode)) + } + if f.Nodes > 0 { + q.Set("nodes", strconv.Itoa(f.Nodes)) + } + if f.Region != "" { + q.Set("region", f.Region) + } + if f.Stack != "" { + q.Set("stack", f.Stack) + } + if f.IncludeCommunity { + q.Set("include_community", "true") + } + path := "/api/v1/clusters/offers" + if len(q) > 0 { + path += "?" + q.Encode() + } + list, err := callJSON[[]ClusterOffer](c, "GET", path, nil) + if err != nil { + return nil, err + } + return *list, nil +} + +// CreateCluster creates a cluster, all or nothing. The same idempotencyKey +// answers the cluster it created; "" sends a fresh one. A gone offer is a 409 +// OFFER_GONE whose details carry the current offers (see OffersOf). +func (c *Client) CreateCluster(req CreateClusterRequest, idempotencyKey string) (*ClusterDetail, error) { + if idempotencyKey == "" { + idempotencyKey = NewIdempotencyKey() + } + return callJSONHeaders[ClusterDetail](c, "POST", "/api/v1/clusters", req, map[string]string{"Idempotency-Key": idempotencyKey}) +} + +// GetCluster returns one cluster with its nodes and how to connect. +func (c *Client) GetCluster(id string) (*ClusterDetail, error) { + return callJSON[ClusterDetail](c, "GET", clusterPath(id), nil) +} + +// DeleteCluster terminates every node of a cluster together. +func (c *Client) DeleteCluster(id string) (*ClusterDetail, error) { + return callJSON[ClusterDetail](c, "DELETE", clusterPath(id), nil) +} + +// StartClusterTest starts an NCCL all-reduce test; a running one is a 409 TEST_RUNNING. +func (c *Client) StartClusterTest(id string) (*ClusterTest, error) { + return callJSON[ClusterTest](c, "POST", clusterPath(id)+"/tests", nil) +} + +// GetClusterTest returns one NCCL test. +func (c *Client) GetClusterTest(id, testID string) (*ClusterTest, error) { + return callJSON[ClusterTest](c, "GET", clusterPath(id)+"/tests/"+url.PathEscape(testID), nil) +} + +// GetClusterKubeconfig returns a Kubernetes cluster's kubeconfig (text/yaml, +// not the JSON envelope) and the tunnel command from X-Runcrate-Tunnel. +func (c *Client) GetClusterKubeconfig(id string) (*Kubeconfig, error) { + resp, err := c.doCtxHeaders(context.Background(), "GET", clusterPath(id)+"/kubeconfig", nil, map[string]string{"Accept": "text/yaml"}) + if err != nil { + return nil, fmt.Errorf("network error: %w", err) + } + defer resp.Body.Close() + if err := checkHTTPStatus(resp); err != nil { + return nil, err + } + yaml, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("reading kubeconfig: %w", err) + } + return &Kubeconfig{YAML: string(yaml), Tunnel: resp.Header.Get("X-Runcrate-Tunnel")}, nil +} + +// OffersOf returns the current offers a 409 OFFER_GONE carries in details.offers. +func OffersOf(apiErr *APIError) []ClusterOffer { + raw, err := json.Marshal(apiErr.Details["offers"]) + if err != nil { + return nil + } + var offers []ClusterOffer + if json.Unmarshal(raw, &offers) != nil { + return nil + } + return offers +} + +func clusterPath(id string) string { + return "/api/v1/clusters/" + url.PathEscape(id) +} diff --git a/internal/api/clusters_test.go b/internal/api/clusters_test.go new file mode 100644 index 0000000..6cc02c1 --- /dev/null +++ b/internal/api/clusters_test.go @@ -0,0 +1,130 @@ +package api + +import ( + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" +) + +const testOffer = `{"offer_id":"off_1","gpu_type":"H100","gpus_per_node":8,"node_count":2,"min_nodes":2,"max_nodes":8,` + + `"nodes_available":null,"deployment_type":"vm","host_tier":"datacenter","location":"Dallas, TX, United States",` + + `"interconnect":{"kind":"ethernet","label":"Ethernet, private overlay"},"hourly_rate_per_node":20,"hourly_rate":40,` + + `"stacks":["none","slurm"],"resize":true,"node":{"vcpus":96,"memory_gb":1024,"storage_gb":2000}}` + +func TestCreateClusterSendsBodyAndIdempotencyKey(t *testing.T) { + var keys []string + var body map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/api/v1/clusters" { + w.WriteHeader(http.StatusNotFound) + return + } + keys = append(keys, r.Header.Get("Idempotency-Key")) + json.NewDecoder(r.Body).Decode(&body) + w.WriteHeader(http.StatusCreated) + w.Write([]byte(`{"data":{"id":"cl-1","name":"train","status":"provisioning","stack":"slurm","node_count":2,` + + `"hourly_rate":40,"nodes":[{"rank":0,"role":"head","instance_id":"dep-0","ip":null}]}}`)) + })) + defer server.Close() + client := NewClient("rc_live_test", server.URL) + + req := CreateClusterRequest{Name: "train", OfferID: "off_1", NodeCount: 2, Stack: "slurm"} + d, err := client.CreateCluster(req, "") + if err != nil { + t.Fatalf("CreateCluster() error: %v", err) + } + if d.ID != "cl-1" || d.Status != "provisioning" || len(d.Nodes) != 1 || d.Nodes[0].InstanceID != "dep-0" || d.Nodes[0].IP != nil { + t.Errorf("cluster = %+v", d) + } + if body["name"] != "train" || body["offer_id"] != "off_1" || body["node_count"] != float64(2) || body["stack"] != "slurm" { + t.Errorf("body = %v", body) + } + for _, unset := range []string{"ssh_key_ids", "storage_id", "include_community"} { + if _, ok := body[unset]; ok { + t.Errorf("unset %q was sent: %v", unset, body[unset]) + } + } + + if _, err := client.CreateCluster(req, "my-key"); err != nil { + t.Fatal(err) + } + if len(keys[0]) != 36 || keys[1] != "my-key" { + t.Errorf("Idempotency-Keys = %q, want a fresh UUID then the given key", keys) + } +} + +func TestCreateClusterOfferGoneCarriesTheCurrentOffers(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusConflict) + w.Write([]byte(`{"error":{"code":"OFFER_GONE","message":"That offer is no longer available.","details":{"offers":[` + testOffer + `]}}}`)) + })) + defer server.Close() + + _, err := NewClient("rc_live_test", server.URL).CreateCluster(CreateClusterRequest{Name: "a", OfferID: "old", NodeCount: 2, Stack: "none"}, "") + var apiErr *APIError + if !errors.As(err, &apiErr) || apiErr.Status != http.StatusConflict || apiErr.Code != "OFFER_GONE" { + t.Fatalf("err = %v, want a 409 OFFER_GONE", err) + } + offers := OffersOf(apiErr) + if len(offers) != 1 || offers[0].OfferID != "off_1" || offers[0].HourlyRate != 40 || offers[0].NodesAvailable != nil { + t.Errorf("offers = %+v", offers) + } +} + +func TestListClusterOffersSendsOnlyTheFiltersGiven(t *testing.T) { + var queries []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + queries = append(queries, r.URL.Path+"?"+r.URL.RawQuery) + w.Write([]byte(`{"data":[` + testOffer + `]}`)) + })) + defer server.Close() + client := NewClient("rc_live_test", server.URL) + + offers, err := client.ListClusterOffers(ClusterOfferFilters{GPUType: "H100", GPUsPerNode: 8, Nodes: 4, IncludeCommunity: true}) + if err != nil || len(offers) != 1 || offers[0].Interconnect.Kind != "ethernet" { + t.Fatalf("offers = %+v, err = %v", offers, err) + } + if _, err := client.ListClusterOffers(ClusterOfferFilters{}); err != nil { + t.Fatal(err) + } + want := []string{ + "/api/v1/clusters/offers?gpu_type=H100&gpus_per_node=8&include_community=true&nodes=4", + "/api/v1/clusters/offers?", + } + if len(queries) != 2 || queries[0] != want[0] || queries[1] != want[1] { + t.Errorf("requests = %q, want %q", queries, want) + } +} + +func TestGetClusterKubeconfigReadsYAMLAndTunnel(t *testing.T) { + const yaml = "apiVersion: v1\nclusters:\n- cluster:\n server: https://127.0.0.1:6443\n" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v1/clusters/k8s/kubeconfig": + w.Header().Set("Content-Type", "text/yaml") + w.Header().Set("X-Runcrate-Tunnel", "ssh -N -L 6443:127.0.0.1:6443 root@203.0.113.5") + w.Write([]byte(yaml)) + default: + w.WriteHeader(http.StatusConflict) + w.Write([]byte(`{"error":{"code":"conflict","message":"This cluster does not run Kubernetes."}}`)) + } + })) + defer server.Close() + client := NewClient("rc_live_test", server.URL) + + kc, err := client.GetClusterKubeconfig("k8s") + if err != nil { + t.Fatalf("GetClusterKubeconfig() error: %v", err) + } + if kc.YAML != yaml || kc.Tunnel != "ssh -N -L 6443:127.0.0.1:6443 root@203.0.113.5" { + t.Errorf("kubeconfig = %+v", kc) + } + + _, err = client.GetClusterKubeconfig("slurm") + var apiErr *APIError + if !errors.As(err, &apiErr) || apiErr.Status != http.StatusConflict || apiErr.Message != "This cluster does not run Kubernetes." { + t.Errorf("err = %v, want the 409 message", err) + } +} diff --git a/internal/api/schedules.go b/internal/api/schedules.go index 9c1fb60..058dbce 100644 --- a/internal/api/schedules.go +++ b/internal/api/schedules.go @@ -2,6 +2,7 @@ package api import ( "bytes" + "context" "encoding/json" "fmt" "io" @@ -114,6 +115,11 @@ func schedulePath(id string) string { // callJSON sends body (nil for none) as JSON and decodes the { data: T } reply. func callJSON[T any](c *Client, method, path string, body any) (*T, error) { + return callJSONHeaders[T](c, method, path, body, nil) +} + +// callJSONHeaders is callJSON with extra request headers (an Idempotency-Key). +func callJSONHeaders[T any](c *Client, method, path string, body any, headers map[string]string) (*T, error) { var reader io.Reader if body != nil { raw, err := json.Marshal(body) @@ -123,7 +129,7 @@ func callJSON[T any](c *Client, method, path string, body any) (*T, error) { reader = bytes.NewReader(raw) } - resp, err := c.do(method, path, reader) + resp, err := c.doCtxHeaders(context.Background(), method, path, reader, headers) if err != nil { return nil, fmt.Errorf("network error: %w", err) } diff --git a/internal/commands/clusters.go b/internal/commands/clusters.go new file mode 100644 index 0000000..e6671d2 --- /dev/null +++ b/internal/commands/clusters.go @@ -0,0 +1,650 @@ +package commands + +import ( + "errors" + "fmt" + "io" + "net/http" + "os" + "sort" + "strings" + "time" + + "github.com/runcrate/cli/internal/api" + "github.com/spf13/cobra" +) + +// ── display ───────────────────────────────────────────────────────────────── + +var stackNames = map[string]string{"none": "None", "slurm": "Slurm", "kubernetes": "Kubernetes", "ray": "Ray"} + +func stackName(stack string) string { + if name, ok := stackNames[stack]; ok { + return name + } + return stack +} + +// clusterShape reads like the dashboard: "8x H100 x 4 nodes". +func clusterShape(gpusPerNode int, gpuType string, nodes int) string { + return fmt.Sprintf("%dx %s x %d nodes", gpusPerNode, gpuType, nodes) +} + +// sinceISO is the time from an ISO instant to now, or "-". +func sinceISO(iso string, now time.Time) string { + t, err := time.Parse(time.RFC3339, iso) + if err != nil { + return "-" + } + return formatShortDuration(now.Sub(t)) +} + +func renderClusterTable(w io.Writer, list []api.Cluster, now time.Time) error { + tw := newTableTo(w) + fmt.Fprintf(tw, "ID\tNAME\tSTATUS\tGPUS\tINTERCONNECT\tSOFTWARE\t$/HR\tAGE\n") + for _, c := range list { + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\t$%.2f\t%s\n", + c.ID, truncate(c.Name, 24), statusColor(c.Status), clusterShape(c.GPUsPerNode, c.GPUType, c.NodeCount), + c.Interconnect.Label, stackName(c.Stack), c.HourlyRate, sinceISO(c.CreatedAt, now)) + } + return tw.Flush() +} + +// nodesText is the node range of an offer, and how many can start now when known. +func nodesText(o *api.ClusterOffer) string { + text := fmt.Sprintf("%d-%d", o.MinNodes, o.MaxNodes) + if o.NodesAvailable != nil { + text += fmt.Sprintf(" (%d free)", *o.NodesAvailable) + } + return text +} + +func renderOfferTable(w io.Writer, offers []api.ClusterOffer) error { + tw := newTableTo(w) + fmt.Fprintf(tw, "OFFER\tGPUS/NODE\tLOCATION\tINTERCONNECT\tKIND\tNODES\tSOFTWARE\t$/HR/NODE\t$/HR\n") + for i := range offers { + o := &offers[i] + kind := o.DeploymentType + if o.HostTier == "community" { + kind += ", community" + } + software := make([]string, 0, len(o.Stacks)) + for _, s := range o.Stacks { + software = append(software, stackName(s)) + } + fmt.Fprintf(tw, "%s\t%dx %s\t%s\t%s\t%s\t%s\t%s\t$%.2f\t$%.2f (%d nodes)\n", + o.OfferID, o.GPUsPerNode, o.GPUType, o.Location, o.Interconnect.Label, kind, nodesText(o), + strings.Join(software, ", "), o.HourlyRatePerNode, o.HourlyRate, o.NodeCount) + } + return tw.Flush() +} + +func orDash(s *string) string { + if s == nil || *s == "" { + return "-" + } + return *s +} + +// renderTestResult writes an NCCL test's outcome and its bandwidth per size. +func renderTestResult(w io.Writer, t *api.ClusterTest) error { + switch { + case t.Status == "running": + fmt.Fprintf(w, "NCCL all-reduce test %s is running (started %s).\n", t.ID, t.StartedAt) + return nil + case t.Result == nil: + fmt.Fprintf(w, "NCCL all-reduce test %s %s with no result.\n", t.ID, t.Status) + return nil + case t.Result.Error != "": + fmt.Fprintf(w, "NCCL all-reduce test %s failed: %s\n", t.ID, t.Result.Error) + return nil + } + r := t.Result + fmt.Fprintf(w, "NCCL all-reduce test %s across %d GPUs on %d nodes: peak bus bandwidth %.1f GB/s.\n", + t.Status, r.GPUs, r.Nodes, r.PeakBusbwGbps) + tw := newTableTo(w) + fmt.Fprintf(tw, "SIZE\tTIME (us)\tALGBW (GB/s)\tBUSBW (GB/s)\n") + for _, s := range r.Sizes { + fmt.Fprintf(tw, "%s\t%.1f\t%.2f\t%.2f\n", formatBytes(s.Bytes), s.TimeUs, s.AlgbwGbps, s.BusbwGbps) + } + return tw.Flush() +} + +// printClusterDetails writes a cluster's fields, nodes, how to connect, its +// last test and its latest events. +func printClusterDetails(w io.Writer, d *api.ClusterDetail, now time.Time) error { + fmt.Fprintln(w) + fmt.Fprintf(w, " Name: %s\n", d.Name) + fmt.Fprintf(w, " ID: %s\n", d.ID) + fmt.Fprintf(w, " Status: %s\n", statusColor(d.Status)) + software := stackName(d.Stack) + if d.Stack != "none" { + software += " (" + d.StackStatus + ")" + } + fmt.Fprintf(w, " Software: %s\n", software) + if d.StackDetail != nil && *d.StackDetail != "" { + fmt.Fprintf(w, " Install: %s\n", *d.StackDetail) + } + fmt.Fprintf(w, " GPUs: %s\n", clusterShape(d.GPUsPerNode, d.GPUType, d.NodeCount)) + fmt.Fprintf(w, " Location: %s\n", d.Location) + fmt.Fprintf(w, " Interconnect: %s\n", d.Interconnect.Label) + fmt.Fprintf(w, " Price: $%.2f/hr for the cluster ($%.2f/hr per node)\n", d.HourlyRate, d.HourlyRatePerNode) + if d.Billing.BilledSince != nil { + fmt.Fprintf(w, " Billed since: %s (%s, %d nodes)\n", *d.Billing.BilledSince, sinceISO(*d.Billing.BilledSince, now), d.Billing.NodesBilled) + } else if d.Status == "provisioning" { + fmt.Fprintf(w, " Billed since: not yet: billing starts when every node is up\n") + } + switch { + case !d.Shared.Available: + fmt.Fprintf(w, " Shared: not available: %s\n", d.Shared.Detail) + case d.Shared.Volume != nil: + fmt.Fprintf(w, " Shared: %s, volume %s at %s\n", d.Shared.Path, d.Shared.Volume.Name, d.Shared.Volume.Path) + default: + fmt.Fprintf(w, " Shared: %s on every node\n", d.Shared.Path) + } + if d.FailureDetail != nil && *d.FailureDetail != "" { + fmt.Fprintf(w, " Failure: %s\n", *d.FailureDetail) + } + + if len(d.Nodes) > 0 { + fmt.Fprintln(w) + tw := newTableTo(w) + fmt.Fprintf(tw, " RANK\tROLE\tSTATUS\tSTAGE\tGPUS\tOVERLAY\tIP\n") + for _, n := range d.Nodes { + fmt.Fprintf(tw, " %d\t%s\t%s\t%s\t%d\t%s\t%s\n", + n.Rank, n.Role, statusColor(n.Status), n.Stage, n.GPUs, orDash(n.OverlayAddress), orDash(n.IP)) + } + if err := tw.Flush(); err != nil { + return err + } + } + + c := d.Connection + if c.HeadSSH != nil { + fmt.Fprintln(w) + fmt.Fprintf(w, " Connect:\n") + fmt.Fprintf(w, " runcrate clusters ssh %s # the head; --node N for another\n", d.Name) + if d.Stack == "kubernetes" { + fmt.Fprintf(w, " runcrate clusters kubeconfig %s -o kubeconfig.yaml\n", d.Name) + } + if d.Stack == "kubernetes" || d.Stack == "ray" { + fmt.Fprintf(w, " runcrate clusters tunnel %s\n", d.Name) + } + keys := make([]string, 0, len(c.Env)) + for k := range c.Env { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + fmt.Fprintf(w, " %s=%s\n", k, c.Env[k]) + } + for _, name := range []string{"torchrun", "mpirun", "deepspeed", "sbatch"} { + if snippet := c.Snippets[name]; snippet != "" { + fmt.Fprintf(w, "\n %s:\n %s\n", name, strings.ReplaceAll(strings.TrimSpace(snippet), "\n", "\n ")) + } + } + } + + if d.LastTest != nil { + fmt.Fprintln(w) + if err := renderTestResult(w, d.LastTest); err != nil { + return err + } + } + + if len(d.Events) > 0 { + fmt.Fprintln(w) + fmt.Fprintf(w, " Recent events:\n") + for i, e := range d.Events { + if i == 5 { + break + } + fmt.Fprintf(w, " %s %s\n", e.At, e.Message) + } + } + fmt.Fprintln(w) + return nil +} + +// ── resolving ─────────────────────────────────────────────────────────────── + +// resolveCluster finds a cluster by ID, else by name (an active one first). +func resolveCluster(client *api.Client, ref string) (*api.ClusterDetail, error) { + d, err := client.GetCluster(ref) + var apiErr *api.APIError + if err == nil || !errors.As(err, &apiErr) || apiErr.Status != http.StatusNotFound || apiErr.Code == "FEATURE_OFF" { + return d, err + } + list, listErr := client.ListClusters(true) + if listErr != nil { + return nil, fmt.Errorf("fetching clusters: %w", listErr) + } + ended := map[string]bool{"terminated": true, "failed": true} + var match *api.Cluster + for i := range list { + c := &list[i] + if !strings.EqualFold(c.Name, ref) { + continue + } + // Prefer a live cluster over an ended one of the same name. + if match == nil || (ended[match.Status] && !ended[c.Status]) { + match = c + } + } + if match == nil { + return nil, fmt.Errorf("cluster %q not found", ref) + } + return client.GetCluster(match.ID) +} + +// clusterNode is the node of rank, with an address to connect to. +func clusterNode(d *api.ClusterDetail, rank int) (*api.ClusterNode, error) { + for i := range d.Nodes { + n := &d.Nodes[i] + if n.Rank != rank { + continue + } + if n.IP == nil || *n.IP == "" { + return nil, fmt.Errorf("node %d of %s has no address yet (the cluster is %s)", rank, d.Name, d.Status) + } + return n, nil + } + return nil, fmt.Errorf("%s has no node %d (ranks 0 to %d)", d.Name, rank, len(d.Nodes)-1) +} + +// tunnelForwards are the ssh -L forwards from this computer to the head for a +// stack: the Kubernetes API on 6443 and the Ray dashboard on 8265. +func tunnelForwards(stack string) ([]string, string, error) { + switch stack { + case "kubernetes": + return []string{"-L", "6443:127.0.0.1:6443"}, "the Kubernetes API on https://127.0.0.1:6443", nil + case "ray": + return []string{"-L", "8265:127.0.0.1:8265"}, "the Ray dashboard on http://127.0.0.1:8265", nil + } + return nil, "", fmt.Errorf("a %s cluster has no API or dashboard to tunnel; connect with: runcrate clusters ssh", stackName(stack)) +} + +// ── commands ──────────────────────────────────────────────────────────────── + +var clustersCmd = &cobra.Command{ + Use: "clusters", + Aliases: []string{"cluster"}, + Short: "Create and use multi-node GPU clusters with Slurm, Kubernetes or Ray", + Long: `A cluster is several GPU machines on one private network, with SSH between +them, /shared on every node, and optionally Slurm, Kubernetes or Ray. + +It is all or nothing: every node starts or none does, and a cluster that +could not start costs nothing. Billing starts when every node is up and runs +until you delete the cluster, which stops every node together. + + runcrate clusters offers --gpu H100:8 --nodes 4 + runcrate clusters create --name train --offer --nodes 4 --stack slurm + runcrate clusters get train + runcrate clusters ssh train -- sinfo + runcrate clusters test train + runcrate clusters delete train`, +} + +var clustersListCmd = &cobra.Command{ + Use: "list", + Aliases: []string{"ls"}, + Short: "List clusters, newest first", + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + all, _ := cmd.Flags().GetBool("all") + list, err := client.ListClusters(all) + if err != nil { + return fmt.Errorf("fetching clusters: %w", err) + } + if jsonOutput { + return printJSON(list) + } + if quiet, _ := cmd.Flags().GetBool("quiet"); quiet { + for _, c := range list { + fmt.Println(c.ID) + } + return nil + } + if len(list) == 0 { + fmt.Println("No clusters. Find where one can start with: runcrate clusters offers --gpu H100 --nodes 2") + return nil + } + return renderClusterTable(os.Stdout, list, time.Now()) + }, +} + +var clustersOffersCmd = &cobra.Command{ + Use: "offers", + Short: "Where a cluster can be created now, with its price per hour", + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + gpuRaw, _ := cmd.Flags().GetString("gpu") + perNodeRaw, _ := cmd.Flags().GetInt("gpu-count") + gpuType, perNode, err := parseGPUFlag(gpuRaw, perNodeRaw, cmd.Flags().Changed("gpu-count")) + if err != nil { + return err + } + f := api.ClusterOfferFilters{GPUType: gpuType, GPUsPerNode: perNode} + f.Nodes, _ = cmd.Flags().GetInt("nodes") + f.Region, _ = cmd.Flags().GetString("region") + f.Stack, _ = cmd.Flags().GetString("stack") + f.IncludeCommunity, _ = cmd.Flags().GetBool("include-community") + + offers, err := client.ListClusterOffers(f) + if err != nil { + return fmt.Errorf("fetching offers: %w", err) + } + if jsonOutput { + return printJSON(offers) + } + if len(offers) == 0 { + fmt.Println("No cluster offers match right now. Try another GPU, fewer nodes, or any region.") + return nil + } + if err := renderOfferTable(os.Stdout, offers); err != nil { + return err + } + fmt.Printf("\nCreate one with: runcrate clusters create --name --offer --nodes %d --stack none|slurm|kubernetes|ray\n", offers[0].NodeCount) + return nil + }, +} + +var clustersCreateCmd = &cobra.Command{ + Use: "create", + Short: "Create a cluster from an offer (all or nothing; billed from when every node is up)", + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + req := api.CreateClusterRequest{} + req.Name, _ = cmd.Flags().GetString("name") + req.OfferID, _ = cmd.Flags().GetString("offer") + req.NodeCount, _ = cmd.Flags().GetInt("nodes") + req.Stack, _ = cmd.Flags().GetString("stack") + req.SSHKeyIDs, _ = cmd.Flags().GetStringSlice("ssh-key") + req.StorageID, _ = cmd.Flags().GetString("volume") + req.IncludeCommunity, _ = cmd.Flags().GetBool("include-community") + if _, ok := stackNames[req.Stack]; !ok { + return fmt.Errorf("--stack must be none, slurm, kubernetes or ray, not %q", req.Stack) + } + if req.NodeCount < 2 || req.NodeCount > 64 { + return fmt.Errorf("--nodes must be 2 to 64") + } + + d, err := client.CreateCluster(req, "") + if api.IsBudgetReached(err) { + return errors.New(budgetReachedMessage) + } + var apiErr *api.APIError + if errors.As(err, &apiErr) && apiErr.Code == "OFFER_GONE" { + fmt.Fprintln(os.Stderr, "That offer is no longer available, so nothing was created. Current offers:") + fmt.Fprintln(os.Stderr) + if err := renderOfferTable(os.Stderr, api.OffersOf(apiErr)); err != nil { + return err + } + return errors.New("pick another offer and run create again") + } + if errors.As(err, &apiErr) && apiErr.Status == http.StatusServiceUnavailable { + return fmt.Errorf("%s Nothing was created or billed. Try another offer, or again in a minute", apiErr.Message) + } + if err != nil { + return fmt.Errorf("creating cluster: %w", err) + } + if jsonOutput { + return printJSON(d) + } + fmt.Printf("Creating cluster %s (%s): %s, %s, %s.\n", d.Name, d.ID, + clusterShape(d.GPUsPerNode, d.GPUType, d.NodeCount), d.Location, stackName(d.Stack)) + fmt.Printf("Every node starts or none does. Billing starts when every node is up, at $%.2f/hr for the cluster.\n", d.HourlyRate) + fmt.Printf("Follow it with: runcrate clusters get %s\n", d.Name) + return nil + }, +} + +var clustersGetCmd = &cobra.Command{ + Use: "get ", + Aliases: []string{"info", "status"}, + Short: "Show a cluster: nodes, how to connect, last test and events", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + d, err := resolveCluster(client, args[0]) + if err != nil { + return fmt.Errorf("fetching cluster: %w", err) + } + if jsonOutput { + return printJSON(d) + } + return printClusterDetails(os.Stdout, d, time.Now()) + }, +} + +var clustersDeleteCmd = &cobra.Command{ + Use: "delete ", + Aliases: []string{"rm", "terminate"}, + Short: "Terminate every node of a cluster and stop its billing", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + d, err := resolveCluster(client, args[0]) + if err != nil { + return fmt.Errorf("fetching cluster: %w", err) + } + if yes, _ := cmd.Flags().GetBool("yes"); !yes { + ok, err := confirm(os.Stdin, os.Stderr, fmt.Sprintf( + "Terminate cluster %s and all %d nodes? Files on the nodes and in /shared are deleted (a volume keeps /shared/volume). [y/N] ", + d.Name, d.NodeCount)) + if err != nil { + return errors.New("no answer to the prompt; pass --yes to terminate without asking") + } + if !ok { + fmt.Println("Kept the cluster.") + return nil + } + } + res, err := client.DeleteCluster(d.ID) + if err != nil { + return fmt.Errorf("terminating cluster: %w", err) + } + if jsonOutput { + return printJSON(res) + } + fmt.Printf("Terminating cluster %s: every node stops and billing ends.\n", d.Name) + return nil + }, +} + +var clustersSSHCmd = &cobra.Command{ + Use: "ssh [-- command]", + Short: "SSH into the head of a cluster, or another node with --node", + Long: `Opens SSH to the head (rank 0) of a cluster, or to the node of --node. +Uses ephemeral SSH certificates, like runcrate ssh. + + runcrate clusters ssh train # shell on the head + runcrate clusters ssh train --node 2 # shell on node 2 + runcrate clusters ssh train -- sinfo # run a command on the head + runcrate clusters ssh train -- cat /shared/hostfile`, + Args: cobra.MinimumNArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + d, err := resolveCluster(client, args[0]) + if err != nil { + return fmt.Errorf("fetching cluster: %w", err) + } + rank, _ := cmd.Flags().GetInt("node") + node, err := clusterNode(d, rank) + if err != nil { + return err + } + var remote []string + if dashIdx := cmd.ArgsLenAtDash(); dashIdx >= 0 && dashIdx < len(args) { + remote = args[dashIdx:] + } + return execSSH(client, node.InstanceID, *node.IP, nil, remote) + }, +} + +var clustersKubeconfigCmd = &cobra.Command{ + Use: "kubeconfig ", + Short: "Print or save the kubeconfig of a Kubernetes cluster", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + d, err := resolveCluster(client, args[0]) + if err != nil { + return fmt.Errorf("fetching cluster: %w", err) + } + kc, err := client.GetClusterKubeconfig(d.ID) + if err != nil { + return fmt.Errorf("fetching kubeconfig: %w", err) + } + out, _ := cmd.Flags().GetString("output") + if out == "" { + fmt.Print(kc.YAML) + } else { + if err := os.WriteFile(out, []byte(kc.YAML), 0o600); err != nil { + return fmt.Errorf("saving kubeconfig: %w", err) + } + fmt.Fprintf(os.Stderr, "Saved the kubeconfig to %s. It grants admin on the cluster: keep it private.\n", out) + } + fmt.Fprintf(os.Stderr, "Its server is https://127.0.0.1:6443. In another terminal run: runcrate clusters tunnel %s\n", d.Name) + if out != "" { + fmt.Fprintf(os.Stderr, "Then: export KUBECONFIG=%s && kubectl get nodes\n", out) + } + return nil + }, +} + +var clustersTunnelCmd = &cobra.Command{ + Use: "tunnel ", + Short: "Forward the Kubernetes API (6443) or Ray dashboard (8265) from the head to this computer", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + d, err := resolveCluster(client, args[0]) + if err != nil { + return fmt.Errorf("fetching cluster: %w", err) + } + forwards, what, err := tunnelForwards(d.Stack) + if err != nil { + return err + } + head, err := clusterNode(d, 0) + if err != nil { + return err + } + fmt.Fprintf(os.Stderr, "Forwarding %s from the head of %s. Press Ctrl-C to stop.\n", what, d.Name) + options := append([]string{"-N", "-o", "ExitOnForwardFailure=yes"}, forwards...) + return execSSH(client, head.InstanceID, *head.IP, options, nil) + }, +} + +// clusterTestPoll is how often test waits between reads of a running test. +const clusterTestPoll = 5 * time.Second + +// clusterTestDeadline is how long test waits: the test itself stops at 10 minutes. +const clusterTestDeadline = 12 * time.Minute + +var clustersTestCmd = &cobra.Command{ + Use: "test ", + Short: "Run an NCCL all-reduce test across every GPU and show its bandwidth", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + d, err := resolveCluster(client, args[0]) + if err != nil { + return fmt.Errorf("fetching cluster: %w", err) + } + t, err := client.StartClusterTest(d.ID) + var apiErr *api.APIError + if errors.As(err, &apiErr) && apiErr.Code == "TEST_RUNNING" && d.LastTest != nil { + fmt.Fprintf(os.Stderr, "A test is already running on %s; waiting for it.\n", d.Name) + t, err = d.LastTest, nil + } + if err != nil { + return fmt.Errorf("starting test: %w", err) + } + + if detach, _ := cmd.Flags().GetBool("detach"); !detach { + if !jsonOutput { + fmt.Fprintf(os.Stderr, "Testing all-reduce across every GPU of %s; this takes a few minutes.\n", d.Name) + } + deadline := time.Now().Add(clusterTestDeadline) + for t.Status == "running" && time.Now().Before(deadline) { + time.Sleep(clusterTestPoll) + if t, err = client.GetClusterTest(d.ID, t.ID); err != nil { + return fmt.Errorf("reading test: %w", err) + } + } + } + if jsonOutput { + return printJSON(t) + } + if err := renderTestResult(os.Stdout, t); err != nil { + return err + } + if t.Status == "running" { + fmt.Printf("Read it later with: runcrate clusters get %s\n", d.Name) + } + return nil + }, +} + +func init() { + clustersCmd.PersistentFlags().BoolVar(&jsonOutput, "json", false, "Output as JSON") + + clustersListCmd.Flags().Bool("all", false, "Include terminated and failed clusters") + clustersListCmd.Flags().BoolP("quiet", "q", false, "Only show cluster IDs") + + clustersOffersCmd.Flags().String("gpu", "", "GPU type, optionally with GPUs per node (e.g. H100 or H100:8)") + clustersOffersCmd.Flags().Int("gpu-count", 0, "GPUs on each node") + clustersOffersCmd.Flags().Int("nodes", 2, "Number of nodes the price is for (2 to 64)") + clustersOffersCmd.Flags().String("region", "", "Only offers in this region, city or country") + clustersOffersCmd.Flags().String("stack", "", "Only offers that can run none, slurm, kubernetes or ray") + clustersOffersCmd.Flags().Bool("include-community", false, "Also show community hosts (datacenter only by default)") + + clustersCreateCmd.Flags().String("name", "", "Cluster name; nodes are named - (required)") + clustersCreateCmd.Flags().String("offer", "", "Offer ID from 'runcrate clusters offers' (required)") + clustersCreateCmd.Flags().Int("nodes", 0, "Number of nodes, 2 to 64, within the offer's range (required)") + clustersCreateCmd.Flags().String("stack", "none", "Software across the nodes: none, slurm, kubernetes or ray") + clustersCreateCmd.Flags().StringSlice("ssh-key", nil, "SSH key ID from 'runcrate ssh-keys list' to add on every node (repeatable)") + clustersCreateCmd.Flags().String("volume", "", "Volume ID synced into /shared/volume on the head") + clustersCreateCmd.Flags().Bool("include-community", false, "Needed when the offer is on community hosts") + for _, name := range []string{"name", "offer", "nodes"} { + _ = clustersCreateCmd.MarkFlagRequired(name) + } + + clustersDeleteCmd.Flags().BoolP("yes", "y", false, "Terminate without asking") + clustersSSHCmd.Flags().Int("node", 0, "Rank of the node to connect to (0 is the head)") + clustersKubeconfigCmd.Flags().StringP("output", "o", "", "Save to this file (mode 600) instead of printing") + clustersTestCmd.Flags().BoolP("detach", "d", false, "Start the test and return without waiting for the result") + + clustersCmd.AddCommand(clustersListCmd, clustersOffersCmd, clustersCreateCmd, clustersGetCmd, clustersDeleteCmd, + clustersSSHCmd, clustersKubeconfigCmd, clustersTunnelCmd, clustersTestCmd) + rootCmd.AddCommand(clustersCmd) +} diff --git a/internal/commands/clusters_test.go b/internal/commands/clusters_test.go new file mode 100644 index 0000000..db0b345 --- /dev/null +++ b/internal/commands/clusters_test.go @@ -0,0 +1,138 @@ +package commands + +import ( + "bytes" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/runcrate/cli/internal/api" +) + +func TestTunnelForwardsByStack(t *testing.T) { + k8s, _, err := tunnelForwards("kubernetes") + if err != nil || strings.Join(k8s, " ") != "-L 6443:127.0.0.1:6443" { + t.Errorf("kubernetes = %q, %v", k8s, err) + } + ray, _, err := tunnelForwards("ray") + if err != nil || strings.Join(ray, " ") != "-L 8265:127.0.0.1:8265" { + t.Errorf("ray = %q, %v", ray, err) + } + for _, stack := range []string{"none", "slurm"} { + if _, _, err := tunnelForwards(stack); err == nil || !strings.Contains(err.Error(), "runcrate clusters ssh") { + t.Errorf("%s: err = %v, want a pointer to clusters ssh", stack, err) + } + } +} + +func TestClusterNodePicksTheRankWithAnAddress(t *testing.T) { + ip := "203.0.113.7" + d := &api.ClusterDetail{ + Cluster: api.Cluster{Name: "train", Status: "provisioning"}, + Nodes: []api.ClusterNode{ + {Rank: 0, Role: "head", InstanceID: "dep-0", IP: &ip}, + {Rank: 1, Role: "worker", InstanceID: "dep-1"}, + }, + } + if n, err := clusterNode(d, 0); err != nil || n.InstanceID != "dep-0" { + t.Errorf("rank 0 = %+v, %v", n, err) + } + if _, err := clusterNode(d, 1); err == nil || !strings.Contains(err.Error(), "no address yet") { + t.Errorf("rank 1 without an ip: err = %v", err) + } + if _, err := clusterNode(d, 5); err == nil || !strings.Contains(err.Error(), "ranks 0 to 1") { + t.Errorf("rank 5: err = %v", err) + } +} + +func TestResolveClusterByNamePrefersALiveOne(t *testing.T) { + var gets []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/api/v1/clusters" && r.URL.Query().Get("status") == "all": + w.Write([]byte(`{"data":[{"id":"cl-old","name":"train","status":"terminated"},{"id":"cl-new","name":"Train","status":"ready"}]}`)) + case r.URL.Path == "/api/v1/clusters/cl-new": + gets = append(gets, "cl-new") + w.Write([]byte(`{"data":{"id":"cl-new","name":"Train","status":"ready","nodes":[]}}`)) + default: + gets = append(gets, r.URL.Path) + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error":{"code":"not_found","message":"Cluster not found"}}`)) + } + })) + defer server.Close() + client := api.NewClient("rc_live_test", server.URL) + + d, err := resolveCluster(client, "train") + if err != nil || d.ID != "cl-new" { + t.Fatalf("resolveCluster = %+v, %v", d, err) + } + if _, err := resolveCluster(client, "nope"); err == nil || !strings.Contains(err.Error(), `"nope" not found`) { + t.Errorf("unknown name: err = %v", err) + } +} + +func TestResolveClusterStopsWhenClustersAreOff(t *testing.T) { + lists := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/api/v1/clusters" { + lists++ + } + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"error":{"code":"FEATURE_OFF","message":"Self-serve clusters are not on for this workspace."}}`)) + })) + defer server.Close() + + _, err := resolveCluster(api.NewClient("rc_live_test", server.URL), "train") + if err == nil || !strings.Contains(err.Error(), "not on for this workspace") || lists != 0 { + t.Errorf("err = %v, lists = %d; want the FEATURE_OFF message and no list call", err, lists) + } +} + +func TestRenderOfferTableShowsRatesAndCommunity(t *testing.T) { + free := 3 + offers := []api.ClusterOffer{ + {OfferID: "off_1", GPUType: "H100", GPUsPerNode: 8, NodeCount: 4, MinNodes: 2, MaxNodes: 8, DeploymentType: "vm", + HostTier: "datacenter", Location: "Dallas, TX, United States", Interconnect: api.ClusterInterconnect{Label: "Ethernet, private overlay"}, + HourlyRatePerNode: 20, HourlyRate: 80, Stacks: []string{"none", "kubernetes"}}, + {OfferID: "off_2", GPUType: "H100", GPUsPerNode: 8, NodeCount: 4, MinNodes: 2, MaxNodes: 4, NodesAvailable: &free, + DeploymentType: "container", HostTier: "community", Location: "Oslo, Norway", Interconnect: api.ClusterInterconnect{Label: "InfiniBand 400 Gb/s"}, + HourlyRatePerNode: 18.5, HourlyRate: 74, Stacks: []string{"none", "ray"}}, + } + var buf bytes.Buffer + if err := renderOfferTable(&buf, offers); err != nil { + t.Fatal(err) + } + out := buf.String() + for _, want := range []string{"off_1", "8x H100", "$20.00", "$80.00 (4 nodes)", "None, Kubernetes", "2-8", + "container, community", "2-4 (3 free)", "$74.00 (4 nodes)"} { + if !strings.Contains(out, want) { + t.Errorf("table lacks %q:\n%s", want, out) + } + } +} + +func TestRenderTestResult(t *testing.T) { + passed := &api.ClusterTest{ID: "t-1", Status: "passed"} + passed.Result = &api.ClusterTestResult{ + Sizes: []api.NCCLSize{{Bytes: 1 << 30, AlgbwGbps: 40.1, BusbwGbps: 75.2, TimeUs: 26800}}, + PeakBusbwGbps: 75.2, Nodes: 2, GPUs: 16, + } + var buf bytes.Buffer + if err := renderTestResult(&buf, passed); err != nil { + t.Fatal(err) + } + for _, want := range []string{"passed across 16 GPUs on 2 nodes", "peak bus bandwidth 75.2 GB/s", "1.00 GB", "75.20"} { + if !strings.Contains(buf.String(), want) { + t.Errorf("output lacks %q:\n%s", want, buf.String()) + } + } + + passed.Result.Error = "NCCL timed out" + buf.Reset() + renderTestResult(&buf, passed) + if !strings.Contains(buf.String(), "failed: NCCL timed out") { + t.Errorf("error result = %q", buf.String()) + } +} diff --git a/internal/commands/helpers.go b/internal/commands/helpers.go index 6c1bf56..e22d9d4 100644 --- a/internal/commands/helpers.go +++ b/internal/commands/helpers.go @@ -86,9 +86,9 @@ func execOrRun(bin string, argv []string) error { // statusColor returns the status with a simple text indicator. func statusColor(status string) string { switch strings.ToLower(status) { - case "running", "deployed", "succeeded": + case "running", "deployed", "succeeded", "ready", "passed": return "● " + status - case "creating", "deploying", "awaiting_provider", "queued", "starting": + case "creating", "deploying", "awaiting_provider", "queued", "starting", "provisioning", "installing", "terminating": return "◌ " + status case "terminated", "failed", "cancelled", "timed_out": return "✕ " + status diff --git a/internal/commands/root.go b/internal/commands/root.go index dc1b0da..bdb4ee5 100644 --- a/internal/commands/root.go +++ b/internal/commands/root.go @@ -77,6 +77,17 @@ func printHelp() { fmt.Println(" jobs logs [-f] Print output (-f to follow)") fmt.Println(" jobs cancel Cancel and stop the machine") fmt.Println() + fmt.Println(" Clusters: Several GPU machines on one network, all or nothing") + fmt.Println(" clusters offers --gpu H100:8 --nodes 4") + fmt.Println(" Where a cluster can start, with its price per hour") + fmt.Println(" clusters create --name X --offer Y --nodes 4 [--stack slurm|kubernetes|ray]") + fmt.Println(" clusters list | get Nodes, how to connect, last test") + fmt.Println(" clusters ssh [--node N] SSH to the head or a node") + fmt.Println(" clusters kubeconfig [-o file] The kubeconfig of a Kubernetes cluster") + fmt.Println(" clusters tunnel Kubernetes API (6443) or Ray dashboard (8265) on this computer") + fmt.Println(" clusters test NCCL all-reduce bandwidth across every GPU") + fmt.Println(" clusters delete Terminate every node and stop billing") + fmt.Println() fmt.Println(" Access:") fmt.Println(" ssh Interactive shell (keyless)") fmt.Println(" ssh -- Run command and exit") diff --git a/internal/commands/ssh.go b/internal/commands/ssh.go index ab4f4c3..3780824 100644 --- a/internal/commands/ssh.go +++ b/internal/commands/ssh.go @@ -35,22 +35,31 @@ Use -- to run a remote command without opening a shell: }, } -// runSSH handles the SSH certificate flow for both ssh and cp commands. -// Returns the temp dir path and SSH args prefix for cp to reuse, or execs ssh directly. +// runSSH opens ssh to an instance by name or ID, running the command after -- +// when there is one. func runSSH(cmd *cobra.Command, args []string, dryRun bool) error { client, err := loadClient() if err != nil { return err } - instanceRef := args[0] - // Resolve instance ID (need the UUID for the certificate API) - instanceID, ip, err := resolveInstance(client, instanceRef) + instanceID, ip, err := resolveInstance(client, args[0]) if err != nil { return err } + var remote []string + if dashIdx := cmd.ArgsLenAtDash(); dashIdx >= 0 && dashIdx < len(args) { + remote = args[dashIdx:] + } + return execSSH(client, instanceID, ip, nil, remote) +} + +// execSSH runs ssh to one instance (by UUID, at ip unless the certificate names +// another address) with an ephemeral key signed for it: options go before the +// destination, remote after it. +func execSSH(client *api.Client, instanceID, ip string, options, remote []string) error { // Generate ephemeral Ed25519 key pair pubKey, privKey, err := generateEphemeralKeyPair() if err != nil { @@ -90,13 +99,9 @@ func runSSH(cmd *cobra.Command, args []string, dryRun bool) error { sshArgs = append(sshArgs, "-o", "CertificateFile="+certPath) sshArgs = append(sshArgs, "-i", keyPath) sshArgs = append(sshArgs, "-p", port) + sshArgs = append(sshArgs, options...) sshArgs = append(sshArgs, fmt.Sprintf("%s@%s", user, ip)) - - // If there are args after --, pass them as the remote command - dashIdx := cmd.ArgsLenAtDash() - if dashIdx >= 0 && dashIdx < len(args) { - sshArgs = append(sshArgs, args[dashIdx:]...) - } + sshArgs = append(sshArgs, remote...) // Find ssh binary sshBin, err := exec.LookPath("ssh") From 90d395c04fb902ecd24e659d182cb0d8ee125948 Mon Sep 17 00:00:00 2001 From: Nirupam Bhowmick <48842933+heydryft@users.noreply.github.com> Date: Sun, 25 Oct 2026 15:47:19 +0000 Subject: [PATCH 2/3] fix(clusters): read NCCL bandwidth as algbw_gb_s, busbw_gb_s, peak_busbw_gb_s (RUN-635) Refs: RUN-635 --- internal/api/clusters.go | 18 +++++++++--------- internal/commands/clusters.go | 4 ++-- internal/commands/clusters_test.go | 4 ++-- 3 files changed, 13 insertions(+), 13 deletions(-) diff --git a/internal/api/clusters.go b/internal/api/clusters.go index ff8e852..69b5e6b 100644 --- a/internal/api/clusters.go +++ b/internal/api/clusters.go @@ -138,10 +138,10 @@ type ClusterEvent struct { // NCCLSize is the all-reduce result at one message size. type NCCLSize struct { - Bytes int64 `json:"bytes"` - AlgbwGbps float64 `json:"algbw_gbps"` - BusbwGbps float64 `json:"busbw_gbps"` - TimeUs float64 `json:"time_us"` + Bytes int64 `json:"bytes"` + AlgbwGBs float64 `json:"algbw_gb_s"` + BusbwGBs float64 `json:"busbw_gb_s"` + TimeUs float64 `json:"time_us"` } // ClusterTest is an NCCL all-reduce test across every GPU of a cluster. @@ -156,11 +156,11 @@ type ClusterTest struct { // ClusterTestResult is a finished test: bandwidth per message size, or Error. type ClusterTestResult struct { - Sizes []NCCLSize `json:"sizes"` - PeakBusbwGbps float64 `json:"peak_busbw_gbps"` - Nodes int `json:"nodes"` - GPUs int `json:"gpus"` - Error string `json:"error,omitempty"` + Sizes []NCCLSize `json:"sizes"` + PeakBusbwGBs float64 `json:"peak_busbw_gb_s"` + Nodes int `json:"nodes"` + GPUs int `json:"gpus"` + Error string `json:"error,omitempty"` } // ClusterDetail is GET /api/v1/clusters/{id}. diff --git a/internal/commands/clusters.go b/internal/commands/clusters.go index e6671d2..ff76c02 100644 --- a/internal/commands/clusters.go +++ b/internal/commands/clusters.go @@ -101,11 +101,11 @@ func renderTestResult(w io.Writer, t *api.ClusterTest) error { } r := t.Result fmt.Fprintf(w, "NCCL all-reduce test %s across %d GPUs on %d nodes: peak bus bandwidth %.1f GB/s.\n", - t.Status, r.GPUs, r.Nodes, r.PeakBusbwGbps) + t.Status, r.GPUs, r.Nodes, r.PeakBusbwGBs) tw := newTableTo(w) fmt.Fprintf(tw, "SIZE\tTIME (us)\tALGBW (GB/s)\tBUSBW (GB/s)\n") for _, s := range r.Sizes { - fmt.Fprintf(tw, "%s\t%.1f\t%.2f\t%.2f\n", formatBytes(s.Bytes), s.TimeUs, s.AlgbwGbps, s.BusbwGbps) + fmt.Fprintf(tw, "%s\t%.1f\t%.2f\t%.2f\n", formatBytes(s.Bytes), s.TimeUs, s.AlgbwGBs, s.BusbwGBs) } return tw.Flush() } diff --git a/internal/commands/clusters_test.go b/internal/commands/clusters_test.go index db0b345..1f48ae4 100644 --- a/internal/commands/clusters_test.go +++ b/internal/commands/clusters_test.go @@ -116,8 +116,8 @@ func TestRenderOfferTableShowsRatesAndCommunity(t *testing.T) { func TestRenderTestResult(t *testing.T) { passed := &api.ClusterTest{ID: "t-1", Status: "passed"} passed.Result = &api.ClusterTestResult{ - Sizes: []api.NCCLSize{{Bytes: 1 << 30, AlgbwGbps: 40.1, BusbwGbps: 75.2, TimeUs: 26800}}, - PeakBusbwGbps: 75.2, Nodes: 2, GPUs: 16, + Sizes: []api.NCCLSize{{Bytes: 1 << 30, AlgbwGBs: 40.1, BusbwGBs: 75.2, TimeUs: 26800}}, + PeakBusbwGBs: 75.2, Nodes: 2, GPUs: 16, } var buf bytes.Buffer if err := renderTestResult(&buf, passed); err != nil { From 6212cc78a89132517be5add6f5f306b697fb3c1c Mon Sep 17 00:00:00 2001 From: Nirupam Bhowmick <48842933+heydryft@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:20:59 +0000 Subject: [PATCH 3/3] fix(clusters): sell clusters by the machine, list volume-ready offers, tunnel k8s to the head Prices read per machine and for the cluster, GPUs per machine are 1, 2, 4 or 8 (refused before the call), offers show allowed node counts and the interconnect note once, test results show why a slow link stopped early, and offers take --with-volume (volume=true) for create --volume. The Kubernetes tunnel forwards 6443 to the head's private address, where k3s listens, instead of its loopback. API types gain allowed_node_counts, country_code, interconnect.note, result.note, resize and event id. --- README.md | 20 +++-- internal/api/clusters.go | 23 ++++- internal/api/clusters_test.go | 4 +- internal/commands/clusters.go | 132 +++++++++++++++++++++-------- internal/commands/clusters_test.go | 48 +++++++++-- 5 files changed, 173 insertions(+), 54 deletions(-) diff --git a/README.md b/README.md index 4c2cd55..8c30b0e 100644 --- a/README.md +++ b/README.md @@ -138,26 +138,32 @@ runcrate schedule delete [--yes] # what it already started keeps running A cluster is several GPU machines on one private network, with SSH between them, `/shared` on every node, and optionally Slurm, Kubernetes or Ray installed across -them. It is all or nothing: every node starts or none does, and a cluster that could -not start costs nothing. Billing starts when every node is up and runs until you -delete the cluster, which stops every node together. +them. It is bought by the machine: each machine has 1, 2, 4 or 8 GPUs, and the +price is per hour for the whole cluster, with the price per machine under it. It is +all or nothing: every machine starts or none does, and a cluster that could not +start costs nothing. Billing starts when every machine is up and runs until you +delete the cluster, which stops every machine together. ```bash -runcrate clusters offers --gpu H100:8 --nodes 4 # where it can start, $/hr per node and for the cluster +runcrate clusters offers --gpu H100 --nodes 4 # the machines it can start on: $/hr per machine and for the cluster +runcrate clusters offers --gpu H100:8 --nodes 4 # only machines with 8 GPUs each runcrate clusters create --name train --offer --nodes 4 --stack slurm runcrate clusters list [--all] [-q] [--json] -runcrate clusters get train # nodes, hostfile, torchrun / mpirun / sbatch, events +runcrate clusters get train # machines, hostfile, torchrun / mpirun / sbatch, events runcrate clusters ssh train [--node 2] [-- sinfo] # the head is node 0 runcrate clusters kubeconfig train -o kubeconfig.yaml # Kubernetes clusters runcrate clusters tunnel train # Kubernetes API on 6443, Ray dashboard on 8265 runcrate clusters test train [-d] # NCCL all-reduce bandwidth across every GPU -runcrate clusters delete train [--yes] # terminates every node +runcrate clusters delete train [--yes] # terminates every machine ``` - `--stack` is `none` (SSH, hostfile and torchrun only), `slurm`, `kubernetes` or `ray`; `offers` lists the ones each offer can run. +- Self-serve clusters run on Ethernet: good for independent jobs, Slurm job arrays + and Ray tasks, not for training one model across machines. Clusters on a fast + InfiniBand or RoCE network are coming; ask us for one. - Offers are on datacenter hosts unless you pass `--include-community` (to both - `offers` and `create`). + `offers` and `create`). With `create --volume`, list offers with `--with-volume`. ## Secrets diff --git a/internal/api/clusters.go b/internal/api/clusters.go index 69b5e6b..7c93ac6 100644 --- a/internal/api/clusters.go +++ b/internal/api/clusters.go @@ -9,14 +9,16 @@ import ( "strconv" ) -// ClusterInterconnect is how a cluster's nodes talk to each other. +// ClusterInterconnect is how a cluster's machines talk to each other. Note says +// what the link is good for when it is not a training fabric (every ethernet one). type ClusterInterconnect struct { Kind string `json:"kind"` // infiniband, roce or ethernet RateGbps *float64 `json:"rate_gbps,omitempty"` Label string `json:"label"` + Note string `json:"note,omitempty"` } -// ClusterNodeSpec is one node of an offer. +// ClusterNodeSpec is one machine of an offer. type ClusterNodeSpec struct { VCPUs int `json:"vcpus"` MemoryGB float64 `json:"memory_gb"` @@ -24,8 +26,10 @@ type ClusterNodeSpec struct { VRAMGB *float64 `json:"vram_gb,omitempty"` } -// ClusterOffer is somewhere a cluster of a shape can be created now, from -// GET /api/v1/clusters/offers. HourlyRate is for the whole cluster at NodeCount. +// ClusterOffer is one kind of machine a cluster can be created on now, from +// GET /api/v1/clusters/offers. A cluster is bought by the machine: +// HourlyRatePerNode is one machine, HourlyRate the whole cluster at NodeCount. +// When AllowedNodeCounts is set, only those numbers of machines can be created. type ClusterOffer struct { OfferID string `json:"offer_id"` GPUType string `json:"gpu_type"` @@ -33,10 +37,12 @@ type ClusterOffer struct { NodeCount int `json:"node_count"` MinNodes int `json:"min_nodes"` MaxNodes int `json:"max_nodes"` + AllowedNodeCounts []int `json:"allowed_node_counts,omitempty"` NodesAvailable *int `json:"nodes_available"` DeploymentType string `json:"deployment_type"` HostTier string `json:"host_tier"` Location string `json:"location"` + CountryCode string `json:"country_code,omitempty"` Interconnect ClusterInterconnect `json:"interconnect"` HourlyRatePerNode float64 `json:"hourly_rate_per_node"` HourlyRate float64 `json:"hourly_rate"` @@ -47,6 +53,7 @@ type ClusterOffer struct { } // ClusterOfferFilters are the query of GET /api/v1/clusters/offers; zero values are not sent. +// Volume lists only machines that may hold a volume (for a create with one). type ClusterOfferFilters struct { GPUType string GPUsPerNode int @@ -54,6 +61,7 @@ type ClusterOfferFilters struct { Region string Stack string IncludeCommunity bool + Volume bool } // CreateClusterRequest is the body of POST /api/v1/clusters. @@ -81,6 +89,7 @@ type Cluster struct { DeploymentType string `json:"deployment_type"` HostTier string `json:"host_tier"` Location string `json:"location"` + CountryCode string `json:"country_code,omitempty"` Interconnect ClusterInterconnect `json:"interconnect"` HourlyRatePerNode float64 `json:"hourly_rate_per_node"` HourlyRate float64 `json:"hourly_rate"` @@ -130,6 +139,7 @@ type ClusterShared struct { // ClusterEvent is one line of a cluster's history. type ClusterEvent struct { + ID int `json:"id"` At string `json:"at"` Kind string `json:"kind"` Rank *int `json:"rank"` @@ -161,6 +171,7 @@ type ClusterTestResult struct { Nodes int `json:"nodes"` GPUs int `json:"gpus"` Error string `json:"error,omitempty"` + Note string `json:"note,omitempty"` // why it stopped early on a slow link } // ClusterDetail is GET /api/v1/clusters/{id}. @@ -175,6 +186,7 @@ type ClusterDetail struct { BilledSince *string `json:"billed_since"` NodesBilled int `json:"nodes_billed"` } `json:"billing"` + Resize bool `json:"resize"` // workers can be added and removed } // Kubeconfig is a Kubernetes cluster's kubeconfig and the SSH tunnel it needs: @@ -219,6 +231,9 @@ func (c *Client) ListClusterOffers(f ClusterOfferFilters) ([]ClusterOffer, error if f.IncludeCommunity { q.Set("include_community", "true") } + if f.Volume { + q.Set("volume", "true") + } path := "/api/v1/clusters/offers" if len(q) > 0 { path += "?" + q.Encode() diff --git a/internal/api/clusters_test.go b/internal/api/clusters_test.go index 6cc02c1..58e1e2b 100644 --- a/internal/api/clusters_test.go +++ b/internal/api/clusters_test.go @@ -82,7 +82,7 @@ func TestListClusterOffersSendsOnlyTheFiltersGiven(t *testing.T) { defer server.Close() client := NewClient("rc_live_test", server.URL) - offers, err := client.ListClusterOffers(ClusterOfferFilters{GPUType: "H100", GPUsPerNode: 8, Nodes: 4, IncludeCommunity: true}) + offers, err := client.ListClusterOffers(ClusterOfferFilters{GPUType: "H100", GPUsPerNode: 8, Nodes: 4, IncludeCommunity: true, Volume: true}) if err != nil || len(offers) != 1 || offers[0].Interconnect.Kind != "ethernet" { t.Fatalf("offers = %+v, err = %v", offers, err) } @@ -90,7 +90,7 @@ func TestListClusterOffersSendsOnlyTheFiltersGiven(t *testing.T) { t.Fatal(err) } want := []string{ - "/api/v1/clusters/offers?gpu_type=H100&gpus_per_node=8&include_community=true&nodes=4", + "/api/v1/clusters/offers?gpu_type=H100&gpus_per_node=8&include_community=true&nodes=4&volume=true", "/api/v1/clusters/offers?", } if len(queries) != 2 || queries[0] != want[0] || queries[1] != want[1] { diff --git a/internal/commands/clusters.go b/internal/commands/clusters.go index ff76c02..c151ede 100644 --- a/internal/commands/clusters.go +++ b/internal/commands/clusters.go @@ -25,9 +25,25 @@ func stackName(stack string) string { return stack } -// clusterShape reads like the dashboard: "8x H100 x 4 nodes". +// clusterShape reads like the dashboard: "4 nodes of 8x H100". func clusterShape(gpusPerNode int, gpuType string, nodes int) string { - return fmt.Sprintf("%dx %s x %d nodes", gpusPerNode, gpuType, nodes) + return fmt.Sprintf("%d nodes of %dx %s", nodes, gpusPerNode, gpuType) +} + +// gpusPerMachine are the GPU counts a machine comes with; a cluster is bought by the machine. +var gpusPerMachine = []int{1, 2, 4, 8} + +// checkGPUsPerMachine refuses a count no machine comes with; 0 means any. +func checkGPUsPerMachine(n int) error { + if n == 0 { + return nil + } + for _, allowed := range gpusPerMachine { + if n == allowed { + return nil + } + } + return fmt.Errorf("machines come with 1, 2, 4 or 8 GPUs, not %d; leave the count out to see every size", n) } // sinceISO is the time from an ISO instant to now, or "-". @@ -41,7 +57,7 @@ func sinceISO(iso string, now time.Time) string { func renderClusterTable(w io.Writer, list []api.Cluster, now time.Time) error { tw := newTableTo(w) - fmt.Fprintf(tw, "ID\tNAME\tSTATUS\tGPUS\tINTERCONNECT\tSOFTWARE\t$/HR\tAGE\n") + fmt.Fprintf(tw, "ID\tNAME\tSTATUS\tMACHINES\tINTERCONNECT\tSOFTWARE\t$/HR\tAGE\n") for _, c := range list { fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\t$%.2f\t%s\n", c.ID, truncate(c.Name, 24), statusColor(c.Status), clusterShape(c.GPUsPerNode, c.GPUType, c.NodeCount), @@ -50,9 +66,17 @@ func renderClusterTable(w io.Writer, list []api.Cluster, now time.Time) error { return tw.Flush() } -// nodesText is the node range of an offer, and how many can start now when known. +// nodesText is how many machines an offer can be created with (its range, or +// the only counts it takes), and how many can start now when known. func nodesText(o *api.ClusterOffer) string { text := fmt.Sprintf("%d-%d", o.MinNodes, o.MaxNodes) + if len(o.AllowedNodeCounts) > 0 { + counts := make([]string, len(o.AllowedNodeCounts)) + for i, n := range o.AllowedNodeCounts { + counts[i] = fmt.Sprint(n) + } + text = strings.Join(counts, ", ") + } if o.NodesAvailable != nil { text += fmt.Sprintf(" (%d free)", *o.NodesAvailable) } @@ -61,7 +85,7 @@ func nodesText(o *api.ClusterOffer) string { func renderOfferTable(w io.Writer, offers []api.ClusterOffer) error { tw := newTableTo(w) - fmt.Fprintf(tw, "OFFER\tGPUS/NODE\tLOCATION\tINTERCONNECT\tKIND\tNODES\tSOFTWARE\t$/HR/NODE\t$/HR\n") + fmt.Fprintf(tw, "OFFER\tGPUS/MACHINE\tLOCATION\tINTERCONNECT\tKIND\tMACHINES\tSOFTWARE\t$/HR/MACHINE\t$/HR\n") for i := range offers { o := &offers[i] kind := o.DeploymentType @@ -72,11 +96,23 @@ func renderOfferTable(w io.Writer, offers []api.ClusterOffer) error { for _, s := range o.Stacks { software = append(software, stackName(s)) } - fmt.Fprintf(tw, "%s\t%dx %s\t%s\t%s\t%s\t%s\t%s\t$%.2f\t$%.2f (%d nodes)\n", + fmt.Fprintf(tw, "%s\t%dx %s\t%s\t%s\t%s\t%s\t%s\t$%.2f\t$%.2f (%d machines)\n", o.OfferID, o.GPUsPerNode, o.GPUType, o.Location, o.Interconnect.Label, kind, nodesText(o), strings.Join(software, ", "), o.HourlyRatePerNode, o.HourlyRate, o.NodeCount) } - return tw.Flush() + if err := tw.Flush(); err != nil { + return err + } + // What each kind of link is good for, once per distinct note. + seen := map[string]bool{} + for i := range offers { + ic := offers[i].Interconnect + if ic.Note != "" && !seen[ic.Note] { + seen[ic.Note] = true + fmt.Fprintf(w, "\n%s: %s\n", ic.Label, ic.Note) + } + } + return nil } func orDash(s *string) string { @@ -102,6 +138,9 @@ func renderTestResult(w io.Writer, t *api.ClusterTest) error { r := t.Result fmt.Fprintf(w, "NCCL all-reduce test %s across %d GPUs on %d nodes: peak bus bandwidth %.1f GB/s.\n", t.Status, r.GPUs, r.Nodes, r.PeakBusbwGBs) + if r.Note != "" { + fmt.Fprintln(w, r.Note) + } tw := newTableTo(w) fmt.Fprintf(tw, "SIZE\tTIME (us)\tALGBW (GB/s)\tBUSBW (GB/s)\n") for _, s := range r.Sizes { @@ -125,14 +164,17 @@ func printClusterDetails(w io.Writer, d *api.ClusterDetail, now time.Time) error if d.StackDetail != nil && *d.StackDetail != "" { fmt.Fprintf(w, " Install: %s\n", *d.StackDetail) } - fmt.Fprintf(w, " GPUs: %s\n", clusterShape(d.GPUsPerNode, d.GPUType, d.NodeCount)) + fmt.Fprintf(w, " Machines: %s\n", clusterShape(d.GPUsPerNode, d.GPUType, d.NodeCount)) fmt.Fprintf(w, " Location: %s\n", d.Location) fmt.Fprintf(w, " Interconnect: %s\n", d.Interconnect.Label) - fmt.Fprintf(w, " Price: $%.2f/hr for the cluster ($%.2f/hr per node)\n", d.HourlyRate, d.HourlyRatePerNode) + if d.Interconnect.Note != "" { + fmt.Fprintf(w, " %s\n", d.Interconnect.Note) + } + fmt.Fprintf(w, " Price: $%.2f/hr for the cluster ($%.2f/hr per machine)\n", d.HourlyRate, d.HourlyRatePerNode) if d.Billing.BilledSince != nil { - fmt.Fprintf(w, " Billed since: %s (%s, %d nodes)\n", *d.Billing.BilledSince, sinceISO(*d.Billing.BilledSince, now), d.Billing.NodesBilled) + fmt.Fprintf(w, " Billed since: %s (%s, %d machines)\n", *d.Billing.BilledSince, sinceISO(*d.Billing.BilledSince, now), d.Billing.NodesBilled) } else if d.Status == "provisioning" { - fmt.Fprintf(w, " Billed since: not yet: billing starts when every node is up\n") + fmt.Fprintf(w, " Billed since: not yet: billing starts when every machine is up\n") } switch { case !d.Shared.Available: @@ -252,12 +294,26 @@ func clusterNode(d *api.ClusterDetail, rank int) (*api.ClusterNode, error) { return nil, fmt.Errorf("%s has no node %d (ranks 0 to %d)", d.Name, rank, len(d.Nodes)-1) } +// headOverlayAddress is the head's address on the cluster's private network. +func headOverlayAddress(d *api.ClusterDetail) string { + for _, n := range d.Nodes { + if n.Rank == 0 && n.OverlayAddress != nil && *n.OverlayAddress != "" { + return *n.OverlayAddress + } + } + if addr := d.Connection.Env["MASTER_ADDR"]; addr != "" { + return addr + } + return "10.77.0.1" +} + // tunnelForwards are the ssh -L forwards from this computer to the head for a -// stack: the Kubernetes API on 6443 and the Ray dashboard on 8265. -func tunnelForwards(stack string) ([]string, string, error) { +// stack: the Kubernetes API on 6443, which listens on the head's private +// address (headAddress), and the Ray dashboard on 8265, on the head's loopback. +func tunnelForwards(stack, headAddress string) ([]string, string, error) { switch stack { case "kubernetes": - return []string{"-L", "6443:127.0.0.1:6443"}, "the Kubernetes API on https://127.0.0.1:6443", nil + return []string{"-L", "6443:" + headAddress + ":6443"}, "the Kubernetes API on https://127.0.0.1:6443", nil case "ray": return []string{"-L", "8265:127.0.0.1:8265"}, "the Ray dashboard on http://127.0.0.1:8265", nil } @@ -271,13 +327,15 @@ var clustersCmd = &cobra.Command{ Aliases: []string{"cluster"}, Short: "Create and use multi-node GPU clusters with Slurm, Kubernetes or Ray", Long: `A cluster is several GPU machines on one private network, with SSH between -them, /shared on every node, and optionally Slurm, Kubernetes or Ray. +them, /shared on every node, and optionally Slurm, Kubernetes or Ray. It is +bought by the machine: each machine has 1, 2, 4 or 8 GPUs, and the price is per +hour for the whole cluster, with the price per machine under it. -It is all or nothing: every node starts or none does, and a cluster that -could not start costs nothing. Billing starts when every node is up and runs -until you delete the cluster, which stops every node together. +It is all or nothing: every machine starts or none does, and a cluster that +could not start costs nothing. Billing starts when every machine is up and runs +until you delete the cluster, which stops every machine together. - runcrate clusters offers --gpu H100:8 --nodes 4 + runcrate clusters offers --gpu H100 --nodes 4 runcrate clusters create --name train --offer --nodes 4 --stack slurm runcrate clusters get train runcrate clusters ssh train -- sinfo @@ -318,7 +376,7 @@ var clustersListCmd = &cobra.Command{ var clustersOffersCmd = &cobra.Command{ Use: "offers", - Short: "Where a cluster can be created now, with its price per hour", + Short: "The machines a cluster can be created on now, priced per machine and for the cluster", RunE: func(cmd *cobra.Command, args []string) error { client, err := loadClient() if err != nil { @@ -330,7 +388,11 @@ var clustersOffersCmd = &cobra.Command{ if err != nil { return err } + if err := checkGPUsPerMachine(perNode); err != nil { + return err + } f := api.ClusterOfferFilters{GPUType: gpuType, GPUsPerNode: perNode} + f.Volume, _ = cmd.Flags().GetBool("with-volume") f.Nodes, _ = cmd.Flags().GetInt("nodes") f.Region, _ = cmd.Flags().GetString("region") f.Stack, _ = cmd.Flags().GetString("stack") @@ -344,7 +406,7 @@ var clustersOffersCmd = &cobra.Command{ return printJSON(offers) } if len(offers) == 0 { - fmt.Println("No cluster offers match right now. Try another GPU, fewer nodes, or any region.") + fmt.Println("No machines for this cluster right now. Try another GPU, fewer machines, any GPU count, or any region.") return nil } if err := renderOfferTable(os.Stdout, offers); err != nil { @@ -357,7 +419,7 @@ var clustersOffersCmd = &cobra.Command{ var clustersCreateCmd = &cobra.Command{ Use: "create", - Short: "Create a cluster from an offer (all or nothing; billed from when every node is up)", + Short: "Create a cluster from an offer (all or nothing; billed from when every machine is up)", RunE: func(cmd *cobra.Command, args []string) error { client, err := loadClient() if err != nil { @@ -402,7 +464,8 @@ var clustersCreateCmd = &cobra.Command{ } fmt.Printf("Creating cluster %s (%s): %s, %s, %s.\n", d.Name, d.ID, clusterShape(d.GPUsPerNode, d.GPUType, d.NodeCount), d.Location, stackName(d.Stack)) - fmt.Printf("Every node starts or none does. Billing starts when every node is up, at $%.2f/hr for the cluster.\n", d.HourlyRate) + fmt.Printf("Every machine starts or none does. Billing starts when every machine is up, at $%.2f/hr for the cluster ($%.2f/hr per machine).\n", + d.HourlyRate, d.HourlyRatePerNode) fmt.Printf("Follow it with: runcrate clusters get %s\n", d.Name) return nil }, @@ -432,7 +495,7 @@ var clustersGetCmd = &cobra.Command{ var clustersDeleteCmd = &cobra.Command{ Use: "delete ", Aliases: []string{"rm", "terminate"}, - Short: "Terminate every node of a cluster and stop its billing", + Short: "Terminate every machine of a cluster and stop its billing", Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { client, err := loadClient() @@ -445,7 +508,7 @@ var clustersDeleteCmd = &cobra.Command{ } if yes, _ := cmd.Flags().GetBool("yes"); !yes { ok, err := confirm(os.Stdin, os.Stderr, fmt.Sprintf( - "Terminate cluster %s and all %d nodes? Files on the nodes and in /shared are deleted (a volume keeps /shared/volume). [y/N] ", + "Terminate cluster %s and all %d machines? Files on the machines and in /shared are deleted (a volume keeps /shared/volume). [y/N] ", d.Name, d.NodeCount)) if err != nil { return errors.New("no answer to the prompt; pass --yes to terminate without asking") @@ -462,7 +525,7 @@ var clustersDeleteCmd = &cobra.Command{ if jsonOutput { return printJSON(res) } - fmt.Printf("Terminating cluster %s: every node stops and billing ends.\n", d.Name) + fmt.Printf("Terminating cluster %s: every machine stops and billing ends.\n", d.Name) return nil }, } @@ -547,7 +610,7 @@ var clustersTunnelCmd = &cobra.Command{ if err != nil { return fmt.Errorf("fetching cluster: %w", err) } - forwards, what, err := tunnelForwards(d.Stack) + forwards, what, err := tunnelForwards(d.Stack, headOverlayAddress(d)) if err != nil { return err } @@ -621,19 +684,20 @@ func init() { clustersListCmd.Flags().Bool("all", false, "Include terminated and failed clusters") clustersListCmd.Flags().BoolP("quiet", "q", false, "Only show cluster IDs") - clustersOffersCmd.Flags().String("gpu", "", "GPU type, optionally with GPUs per node (e.g. H100 or H100:8)") - clustersOffersCmd.Flags().Int("gpu-count", 0, "GPUs on each node") - clustersOffersCmd.Flags().Int("nodes", 2, "Number of nodes the price is for (2 to 64)") + clustersOffersCmd.Flags().String("gpu", "", "GPU type, optionally with GPUs per machine (e.g. H100 or H100:8)") + clustersOffersCmd.Flags().Int("gpu-count", 0, "GPUs on each machine: 1, 2, 4 or 8 (default: every size)") + clustersOffersCmd.Flags().Int("nodes", 2, "Number of machines the price is for (2 to 64)") clustersOffersCmd.Flags().String("region", "", "Only offers in this region, city or country") clustersOffersCmd.Flags().String("stack", "", "Only offers that can run none, slurm, kubernetes or ray") clustersOffersCmd.Flags().Bool("include-community", false, "Also show community hosts (datacenter only by default)") + clustersOffersCmd.Flags().Bool("with-volume", false, "Only machines that may hold a volume (for create --volume)") - clustersCreateCmd.Flags().String("name", "", "Cluster name; nodes are named - (required)") + clustersCreateCmd.Flags().String("name", "", "Cluster name: letters, digits and hyphens; machines are named - (required)") clustersCreateCmd.Flags().String("offer", "", "Offer ID from 'runcrate clusters offers' (required)") - clustersCreateCmd.Flags().Int("nodes", 0, "Number of nodes, 2 to 64, within the offer's range (required)") + clustersCreateCmd.Flags().Int("nodes", 0, "Number of machines, 2 to 64, within the offer's range (required)") clustersCreateCmd.Flags().String("stack", "none", "Software across the nodes: none, slurm, kubernetes or ray") - clustersCreateCmd.Flags().StringSlice("ssh-key", nil, "SSH key ID from 'runcrate ssh-keys list' to add on every node (repeatable)") - clustersCreateCmd.Flags().String("volume", "", "Volume ID synced into /shared/volume on the head") + clustersCreateCmd.Flags().StringSlice("ssh-key", nil, "SSH key ID from 'runcrate ssh-keys list' to add on every machine (repeatable)") + clustersCreateCmd.Flags().String("volume", "", "Volume ID synced into /shared/volume on the head (list offers with --with-volume)") clustersCreateCmd.Flags().Bool("include-community", false, "Needed when the offer is on community hosts") for _, name := range []string{"name", "offer", "nodes"} { _ = clustersCreateCmd.MarkFlagRequired(name) diff --git a/internal/commands/clusters_test.go b/internal/commands/clusters_test.go index 1f48ae4..d590bc5 100644 --- a/internal/commands/clusters_test.go +++ b/internal/commands/clusters_test.go @@ -11,21 +11,47 @@ import ( ) func TestTunnelForwardsByStack(t *testing.T) { - k8s, _, err := tunnelForwards("kubernetes") - if err != nil || strings.Join(k8s, " ") != "-L 6443:127.0.0.1:6443" { + // k3s listens on the head's private address only, never its loopback. + k8s, _, err := tunnelForwards("kubernetes", "10.77.0.1") + if err != nil || strings.Join(k8s, " ") != "-L 6443:10.77.0.1:6443" { t.Errorf("kubernetes = %q, %v", k8s, err) } - ray, _, err := tunnelForwards("ray") + ray, _, err := tunnelForwards("ray", "10.77.0.1") if err != nil || strings.Join(ray, " ") != "-L 8265:127.0.0.1:8265" { t.Errorf("ray = %q, %v", ray, err) } for _, stack := range []string{"none", "slurm"} { - if _, _, err := tunnelForwards(stack); err == nil || !strings.Contains(err.Error(), "runcrate clusters ssh") { + if _, _, err := tunnelForwards(stack, "10.77.0.1"); err == nil || !strings.Contains(err.Error(), "runcrate clusters ssh") { t.Errorf("%s: err = %v, want a pointer to clusters ssh", stack, err) } } } +func TestHeadOverlayAddressPrefersTheHeadsOwn(t *testing.T) { + head, worker := "10.42.0.5", "10.42.0.6" + d := &api.ClusterDetail{Nodes: []api.ClusterNode{{Rank: 1, OverlayAddress: &worker}, {Rank: 0, OverlayAddress: &head}}} + if got := headOverlayAddress(d); got != head { + t.Errorf("headOverlayAddress = %q, want the head's %q", got, head) + } + d = &api.ClusterDetail{Connection: api.ClusterConnection{Env: map[string]string{"MASTER_ADDR": "10.77.0.1"}}} + if got := headOverlayAddress(d); got != "10.77.0.1" { + t.Errorf("without a head address = %q, want MASTER_ADDR", got) + } +} + +func TestCheckGPUsPerMachine(t *testing.T) { + for _, n := range []int{0, 1, 2, 4, 8} { + if err := checkGPUsPerMachine(n); err != nil { + t.Errorf("%d: %v", n, err) + } + } + for _, n := range []int{3, 6, 16} { + if err := checkGPUsPerMachine(n); err == nil { + t.Errorf("%d GPUs per machine was accepted", n) + } + } +} + func TestClusterNodePicksTheRankWithAnAddress(t *testing.T) { ip := "203.0.113.7" d := &api.ClusterDetail{ @@ -94,8 +120,13 @@ func TestRenderOfferTableShowsRatesAndCommunity(t *testing.T) { free := 3 offers := []api.ClusterOffer{ {OfferID: "off_1", GPUType: "H100", GPUsPerNode: 8, NodeCount: 4, MinNodes: 2, MaxNodes: 8, DeploymentType: "vm", - HostTier: "datacenter", Location: "Dallas, TX, United States", Interconnect: api.ClusterInterconnect{Label: "Ethernet, private overlay"}, + HostTier: "datacenter", Location: "Dallas, TX, United States", + Interconnect: api.ClusterInterconnect{Kind: "ethernet", Label: "Ethernet", Note: "No fast link between nodes."}, HourlyRatePerNode: 20, HourlyRate: 80, Stacks: []string{"none", "kubernetes"}}, + {OfferID: "off_3", GPUType: "H100", GPUsPerNode: 4, NodeCount: 4, MinNodes: 2, MaxNodes: 8, AllowedNodeCounts: []int{2, 4, 8}, + DeploymentType: "vm", HostTier: "datacenter", Location: "Dallas, TX, United States", + Interconnect: api.ClusterInterconnect{Kind: "ethernet", Label: "Ethernet", Note: "No fast link between nodes."}, + HourlyRatePerNode: 10, HourlyRate: 40, Stacks: []string{"none"}}, {OfferID: "off_2", GPUType: "H100", GPUsPerNode: 8, NodeCount: 4, MinNodes: 2, MaxNodes: 4, NodesAvailable: &free, DeploymentType: "container", HostTier: "community", Location: "Oslo, Norway", Interconnect: api.ClusterInterconnect{Label: "InfiniBand 400 Gb/s"}, HourlyRatePerNode: 18.5, HourlyRate: 74, Stacks: []string{"none", "ray"}}, @@ -105,12 +136,15 @@ func TestRenderOfferTableShowsRatesAndCommunity(t *testing.T) { t.Fatal(err) } out := buf.String() - for _, want := range []string{"off_1", "8x H100", "$20.00", "$80.00 (4 nodes)", "None, Kubernetes", "2-8", - "container, community", "2-4 (3 free)", "$74.00 (4 nodes)"} { + for _, want := range []string{"off_1", "8x H100", "$20.00", "$80.00 (4 machines)", "None, Kubernetes", "2-8", + "container, community", "2-4 (3 free)", "$74.00 (4 machines)", "4x H100", "2, 4, 8"} { if !strings.Contains(out, want) { t.Errorf("table lacks %q:\n%s", want, out) } } + if strings.Count(out, "Ethernet: No fast link between nodes.") != 1 { + t.Errorf("the ethernet note should show once under the table:\n%s", out) + } } func TestRenderTestResult(t *testing.T) {