diff --git a/README.md b/README.md index 6532860..8c30b0e 100644 --- a/README.md +++ b/README.md @@ -134,6 +134,37 @@ 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 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 --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 # 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 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`). With `create --volume`, list offers with `--with-volume`. + ## 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..7c93ac6 --- /dev/null +++ b/internal/api/clusters.go @@ -0,0 +1,311 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/url" + "strconv" +) + +// 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 machine 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 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"` + GPUsPerNode int `json:"gpus_per_node"` + 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"` + 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. +// Volume lists only machines that may hold a volume (for a create with one). +type ClusterOfferFilters struct { + GPUType string + GPUsPerNode int + Nodes int + Region string + Stack string + IncludeCommunity bool + Volume 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"` + CountryCode string `json:"country_code,omitempty"` + 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 { + ID int `json:"id"` + 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"` + 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. +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"` + PeakBusbwGBs float64 `json:"peak_busbw_gb_s"` + 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}. +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"` + Resize bool `json:"resize"` // workers can be added and removed +} + +// 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") + } + if f.Volume { + q.Set("volume", "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..58e1e2b --- /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, Volume: 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&volume=true", + "/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..c151ede --- /dev/null +++ b/internal/commands/clusters.go @@ -0,0 +1,714 @@ +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: "4 nodes of 8x H100". +func clusterShape(gpusPerNode int, gpuType string, nodes int) string { + 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 "-". +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\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), + c.Interconnect.Label, stackName(c.Stack), c.HourlyRate, sinceISO(c.CreatedAt, now)) + } + return tw.Flush() +} + +// 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) + } + return text +} + +func renderOfferTable(w io.Writer, offers []api.ClusterOffer) error { + tw := newTableTo(w) + 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 + 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 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) + } + 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 { + 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.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 { + fmt.Fprintf(tw, "%s\t%.1f\t%.2f\t%.2f\n", formatBytes(s.Bytes), s.TimeUs, s.AlgbwGBs, s.BusbwGBs) + } + 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, " 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) + 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 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 machine 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) +} + +// 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, 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:" + 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 + } + 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 +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. + + 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 + 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: "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 { + 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 + } + 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") + 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 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 { + 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 machine 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 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 + }, +} + +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 machine 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 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") + } + 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 machine 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, headOverlayAddress(d)) + 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 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: 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 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 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) + } + + 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..d590bc5 --- /dev/null +++ b/internal/commands/clusters_test.go @@ -0,0 +1,172 @@ +package commands + +import ( + "bytes" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/runcrate/cli/internal/api" +) + +func TestTunnelForwardsByStack(t *testing.T) { + // 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", "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, "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{ + 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{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"}}, + } + 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 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) { + passed := &api.ClusterTest{ID: "t-1", Status: "passed"} + passed.Result = &api.ClusterTestResult{ + 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 { + 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")