Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,30 @@ runcrate instances auto-stop train --max-runtime 8h --idle-stop off
Point the CLI at a different API origin with `runcrate config set url <url>` or the
`RUNCRATE_API_URL` env var (defaults to `https://runcrate.ai`).

## Jobs

A job runs one command on its own GPU machine, streams its output, and stops the
machine (and its billing) when the command ends.

```bash
runcrate run --gpu A4000 -- nvidia-smi
runcrate run --gpu H100 --volume <volume-id> --max-runtime 2h -- python train.py --epochs 3
runcrate run --gpu A4000 --env HF_TOKEN -d -- 'pip install -r requirements.txt && python eval.py'
runcrate jobs list [--status running,queued] [--limit 20] [-q] [--json]
runcrate jobs status <id> # exit code, error, cost, instance, env keys
runcrate jobs logs <id> [-f] # print output; -f follows until the job ends
runcrate jobs cancel <id>
```

- The command runs with bash in `/workspace`, which holds the volume's files when
`--volume` is set; files written there are copied back when the machine stops.
- `--max-runtime` takes `90m`, `2h`, `1h30m` or minutes, from 1m to 24h (default 1h).
- `--env KEY=VALUE` is repeatable; `--env KEY` copies the value from your shell.
Values are never stored or shown.
- `runcrate run` and `runcrate jobs logs -f` exit with the job's code: the command's
own code, 124 on timeout, 130 when cancelled. Ctrl-C detaches and leaves the job
running; `-d` / `--detach` prints the job ID and returns.

## Updating

```bash
Expand Down
19 changes: 15 additions & 4 deletions internal/api/client.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package api

import (
"context"
"encoding/json"
"errors"
"fmt"
Expand All @@ -24,12 +25,16 @@ type Client struct {
// APIError is a non-2xx response from the v1 API. Status is the HTTP status code;
// Code and Message come from the { error: { code, message } } envelope when present.
type APIError struct {
Status int `json:"-"`
Code string `json:"code"`
Message string `json:"message"`
Status int `json:"-"`
Code string `json:"code"`
Message string `json:"message"`
Details map[string]any `json:"details,omitempty"`
}

func (e *APIError) Error() string {
if e.Status == http.StatusPaymentRequired && strings.EqualFold(e.Code, "BUDGET_REACHED") {
return "this environment's monthly budget is reached; an owner or admin can raise it in the dashboard (HTTP 402)"
}
if e.Message != "" {
return fmt.Sprintf("%s (HTTP %d)", e.Message, e.Status)
}
Expand Down Expand Up @@ -72,8 +77,14 @@ func NewOAuthClient(baseURL, accessToken, projectID, environment string) *Client
}

func (c *Client) do(method, path string, body io.Reader) (*http.Response, error) {
return c.doCtx(context.Background(), method, path, body)
}

// doCtx is do with a request context, for calls that can be cancelled or need
// a deadline other than the client's default timeout.
func (c *Client) doCtx(ctx context.Context, method, path string, body io.Reader) (*http.Response, error) {
url := c.baseURL + path
req, err := http.NewRequest(method, url, body)
req, err := http.NewRequestWithContext(ctx, method, url, body)
if err != nil {
return nil, err
}
Expand Down
229 changes: 229 additions & 0 deletions internal/api/jobs.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,229 @@
package api

import (
"context"
"encoding/json"
"fmt"
"net/url"
"strconv"
"strings"
"time"
)

// Job is one shell command run on its own machine (RUN-486).
type Job struct {
ID string `json:"id"`
Name string `json:"name"`
Status string `json:"status"`
Command string `json:"command"`
GPUType string `json:"gpu_type"`
GPUCount int `json:"gpu_count"`
Template string `json:"template"`
Image string `json:"image"`
EnvKeys []string `json:"env_keys"`
StorageID string `json:"storage_id"`
MaxRuntimeMinutes int `json:"max_runtime_minutes"`
ExitCode *int `json:"exit_code"`
Error string `json:"error"`
InstanceID string `json:"instance_id"`
EnvironmentID string `json:"environment_id"`
CreatedAt string `json:"created_at"`
StartedAt string `json:"started_at"`
EndedAt string `json:"ended_at"`
CostPerHour float64 `json:"cost_per_hour"`
CostUSD *float64 `json:"cost_usd"`
LogBytes int64 `json:"log_bytes"`
LogsTruncated bool `json:"logs_truncated"`
}

// LogChunk is a raw fragment of a job's combined stdout and stderr. Seq and
// TS are null on the marker chunk that stands where truncated output was dropped.
type LogChunk struct {
Seq *int64 `json:"seq"`
TS *string `json:"ts"`
Text string `json:"text"`
}

// JobLogs is one page of a job's output. Pass Cursor back as `after`; Done means
// the job has ended and every chunk was read.
type JobLogs struct {
JobID string `json:"job_id"`
Status string `json:"status"`
Chunks []LogChunk `json:"chunks"`
Cursor int64 `json:"cursor"`
Truncated bool `json:"truncated"`
Done bool `json:"done"`
}

// JobList is one page of jobs, newest first.
type JobList struct {
Jobs []Job
Cursor string
HasMore bool
}

// CreateJobRequest is the body of POST /api/v1/jobs.
type CreateJobRequest struct {
Command string `json:"command"`
GPUType string `json:"gpu_type"`
MaxRuntimeMinutes int `json:"max_runtime_minutes"`
GPUCount int `json:"gpu_count,omitempty"`
Template string `json:"template,omitempty"`
Image string `json:"image,omitempty"`
Env map[string]string `json:"env,omitempty"`
StorageID string `json:"storage_id,omitempty"`
Name string `json:"name,omitempty"`
Region string `json:"region,omitempty"`
}

// LongPollTimeout bounds a follow=true logs request. The server holds it for
// up to 25 s, so this leaves room for the response without hanging on a dead
// connection for the client's default timeout.
const LongPollTimeout = 40 * time.Second

// CreateJob submits a job. When the job was created but its machine could not
// be launched, the *APIError carries Details["job_id"].
func (c *Client) CreateJob(req CreateJobRequest) (*Job, error) {
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("encoding request: %w", err)
}

resp, err := c.do("POST", "/api/v1/jobs", strings.NewReader(string(body)))
if err != nil {
return nil, fmt.Errorf("network error: %w", err)
}
defer resp.Body.Close()

if err := checkHTTPStatus(resp); err != nil {
return nil, err
}

var result apiEnvelope[Job]
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("parsing response: %w", err)
}
return &result.Data, nil
}

// ListJobs returns one page of jobs. status is a comma list; empty means all.
// limit <= 0 uses the server default; cursor is meta.cursor from a previous page.
func (c *Client) ListJobs(status string, limit int, cursor string) (*JobList, error) {
q := url.Values{}
if status != "" {
q.Set("status", status)
}
if limit > 0 {
q.Set("limit", strconv.Itoa(limit))
}
if cursor != "" {
q.Set("cursor", cursor)
}
path := "/api/v1/jobs"
if len(q) > 0 {
path += "?" + q.Encode()
}

resp, err := c.do("GET", path, nil)
if err != nil {
return nil, fmt.Errorf("network error: %w", err)
}
defer resp.Body.Close()

if err := checkHTTPStatus(resp); err != nil {
return nil, err
}

var result struct {
Data []Job `json:"data"`
Meta struct {
Cursor any `json:"cursor"`
HasMore bool `json:"hasMore"`
} `json:"meta"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("parsing response: %w", err)
}

list := &JobList{Jobs: result.Data, HasMore: result.Meta.HasMore}
// The cursor is opaque: keep it as the string to send back.
switch v := result.Meta.Cursor.(type) {
case string:
list.Cursor = v
case float64:
list.Cursor = strconv.FormatFloat(v, 'f', -1, 64)
}
return list, nil
}

// GetJob returns one job.
func (c *Client) GetJob(id string) (*Job, error) {
resp, err := c.do("GET", "/api/v1/jobs/"+url.PathEscape(id), nil)
if err != nil {
return nil, fmt.Errorf("network error: %w", err)
}
defer resp.Body.Close()

if err := checkHTTPStatus(resp); err != nil {
return nil, err
}

var result apiEnvelope[Job]
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("parsing response: %w", err)
}
return &result.Data, nil
}

// GetJobLogs returns chunks after the seq cursor `after`. With follow, the
// server waits up to 25 s for new output or the end, and the request is bounded
// by LongPollTimeout. limit <= 0 uses the server default. Cancelling ctx aborts
// the request.
func (c *Client) GetJobLogs(ctx context.Context, id string, after int64, follow bool, limit int) (*JobLogs, error) {
q := url.Values{}
q.Set("after", strconv.FormatInt(after, 10))
if follow {
q.Set("follow", "true")
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, LongPollTimeout)
defer cancel()
}
if limit > 0 {
q.Set("limit", strconv.Itoa(limit))
}

resp, err := c.doCtx(ctx, "GET", "/api/v1/jobs/"+url.PathEscape(id)+"/logs?"+q.Encode(), nil)
if err != nil {
return nil, fmt.Errorf("network error: %w", err)
}
defer resp.Body.Close()

if err := checkHTTPStatus(resp); err != nil {
return nil, err
}

var result apiEnvelope[JobLogs]
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("parsing response: %w", err)
}
return &result.Data, nil
}

// CancelJob stops the job's machine; the job becomes cancelled.
func (c *Client) CancelJob(id string) (*Job, error) {
resp, err := c.do("POST", "/api/v1/jobs/"+url.PathEscape(id)+"/cancel", nil)
if err != nil {
return nil, fmt.Errorf("network error: %w", err)
}
defer resp.Body.Close()

if err := checkHTTPStatus(resp); err != nil {
return nil, err
}

var result apiEnvelope[Job]
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("parsing response: %w", err)
}
return &result.Data, nil
}
Loading
Loading