From 872ffcaf25e3d2954eabd94470e47f5e917067a5 Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Wed, 5 Aug 2026 18:31:25 +0000 Subject: [PATCH] =?UTF-8?q?feat(P07a/b/c):=20runtime=20abstraction=20?= =?UTF-8?q?=E2=80=94=205=20backends=20(process/podman/wasm/pve-vm/pve-ct),?= =?UTF-8?q?=20C-01=20satisfied=20(REQ-078)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P07a/b/c — Runtime abstraction interface + 5 implementations. Runtime interface (internal/runtime/runtime.go, REQ-078): - Runtime interface { Prepare, Start, Stop, Status }. Alloc struct carries Runtime field (changeable on migration per R-004). Registry keyed by runtime.one_of. DefaultRegistry(transport) registers all 5. Process (internal/runtime/process.go): - ProcessRuntime wraps os/exec (LOCAL testing only; production uses systemd emitter). SIGTERM grace 10s then SIGKILL. Podman (internal/runtime/podman.go): - PodmanRuntime via sshpush.Transport. podman pull/run/stop/rm/inspect. Wasm (internal/runtime/wasm.go, gate C-01 SATISFIED): - WasmRuntime uses wasmtime CLI (apt-installed on peer) via SSH exec. NO CGO — does NOT import bytecodealliance/wasmtime-go. CGO_ENABLED=0 build passes. D-002 cross-compile story preserved. D-187 recorded. PVE (internal/runtime/pve.go): - PveVMRuntime (qm create/start/stop/status) + PveCTRuntime (pct create/start/stop/status) via sshpush.Transport. VMID = hash(alloc.ID)%99999. C-01 evaluation: internal/runtime/C01_WASMTIME_CGO_EVAL.md. Auto-decision (full autonomy): wasmtime remains primary; CLI-via-SSH avoids CGO entirely. D-187 in PROJECT.md. 23 packages pass, 20 bats pass, gofmt clean, verify-reqs 90 consistent. 92.7% coverage on internal/runtime. ---ci--- project: orca phase: P07a/b/c milestone: v0.9 status: execute ---/ci--- --- .ciagent/PROJECT.md | 1 + internal/runtime/C01_WASMTIME_CGO_EVAL.md | 43 +++ internal/runtime/podman.go | 142 +++++++++ internal/runtime/podman_test.go | 229 +++++++++++++++ internal/runtime/process.go | 204 +++++++++++++ internal/runtime/process_test.go | 172 +++++++++++ internal/runtime/pve.go | 177 +++++++++++ internal/runtime/pve_test.go | 342 ++++++++++++++++++++++ internal/runtime/registry.go | 20 ++ internal/runtime/runtime.go | 176 +++++++++++ internal/runtime/runtime_test.go | 334 +++++++++++++++++++++ internal/runtime/wasm.go | 99 +++++++ internal/runtime/wasm_test.go | 171 +++++++++++ 13 files changed, 2110 insertions(+) create mode 100644 internal/runtime/C01_WASMTIME_CGO_EVAL.md create mode 100644 internal/runtime/podman.go create mode 100644 internal/runtime/podman_test.go create mode 100644 internal/runtime/process.go create mode 100644 internal/runtime/process_test.go create mode 100644 internal/runtime/pve.go create mode 100644 internal/runtime/pve_test.go create mode 100644 internal/runtime/registry.go create mode 100644 internal/runtime/runtime.go create mode 100644 internal/runtime/runtime_test.go create mode 100644 internal/runtime/wasm.go create mode 100644 internal/runtime/wasm_test.go diff --git a/.ciagent/PROJECT.md b/.ciagent/PROJECT.md index da68470..fc9e190 100644 --- a/.ciagent/PROJECT.md +++ b/.ciagent/PROJECT.md @@ -445,3 +445,4 @@ are recorded in `REQUIREMENTS.md`. The reordered phase plan is in | D-158 | Namespace model: single flat root or multi-namespace? | **Multi-namespace under ORCA_HOME (R-002)** | Hard multi-tenant product requirement (override ground 3). `_defaults/` implicit root; `cluster/` for cluster-wide; per-namespace `db/`, `.env`, `.env.secrets`, `jobs/`, `alloc/`, `ns.md`. No namespace column in SQLite. | 0.84 | | D-179 | Jobspec format: HCL canonical (AD-007) or Markdown? | **Markdown with YAML frontmatter canonical (R-013); HCL legacy** | PRD §8 — Markdown + body preservation is the operator-facing format. HCL adapter (REQ-064) preserves `orca job run old-spec.hcl` during migration. | 0.85 | | D-185 | Re-architecture justification: incremental additive or full re-architecture? | **Full re-architecture (overridden by user)** | Six-part evidence basis above; the grill's REPLAN mechanics (PC-01..PC-10, C-01..C-19) adopted as gates. The incremental-additive path was evaluated and rejected on grounds 1 + 5 (daemon failing; SSH-push only viable). | 0.88 | +| D-187 | wasmtime Go binding (bytecodealliance/wasmtime-go) is CGO-based — does adopting it revoke D-002 (modernc/sqlite CGO-free cross-compile story)? | **Use the wasmtime CLI (apt-installed on peer) via SSH exec; do NOT import wasmtime-go.** | The Go binding links libwasmtime via cgo and would revoke D-002's CGO-free cross-compile story. The CLI-via-SSH approach (same pattern as podman/qm/pct) avoids CGO entirely. `internal/runtime/wasm.go` imports only stdlib + sshpush. `CGO_ENABLED=0 go build ./...` succeeds. C-01 grill gate SATISFIED; D-002 NOT revoked. Full evaluation in `internal/runtime/C01_WASMTIME_CGO_EVAL.md`. | 0.90 | diff --git a/internal/runtime/C01_WASMTIME_CGO_EVAL.md b/internal/runtime/C01_WASMTIME_CGO_EVAL.md new file mode 100644 index 0000000..353ad9f --- /dev/null +++ b/internal/runtime/C01_WASMTIME_CGO_EVAL.md @@ -0,0 +1,43 @@ +# C-01 Grill Gate Evaluation — wasmtime / CGO + +**Gate**: C-01 (wasmtime/CGO evaluation), gating P07b. + +**Question**: The wasmtime Go binding +(`github.com/bytecodealliance/wasmtime-go`) is CGO-based. Does adopting +it revoke D-002 (modernc/sqlite CGO-free cross-compile story)? + +## Evaluation + +| Option | CGO required? | Cross-compile impact | Decision | +|--------|---------------|----------------------|----------| +| A. Use `bytecodealliance/wasmtime-go` (Go binding) | **YES** — the binding links libwasmtime via cgo | Revokes D-002 — Go cross-compile (`GOOS=linux GOARCH=arm64 go build`) breaks; CGO toolchain needed on every build host; static-binary story lost | REJECTED | +| B. Use the `wasmtime` CLI (apt-installed on the peer) via SSH exec | **NO** — pure Go code, shells out to a CLI over SSH (same pattern as podman/qm/pct) | None — D-002 preserved | **ACCEPTED** | +| C. Use an alternative pure-Go WASM runtime (e.g. wazero) | No CGO | Pure-Go alternative exists; but wazero's wasmtime-compat is incomplete (component model, WASI 0.2); different runtime semantics than the "wasmtime" operator surface promised in D-088 | DEFERRED (v0.10 evaluation if CLI-via-SSH proves insufficient) | + +## Decision (full autonomy, auto-decision) + +**Option B**: `WasmRuntime` uses the `wasmtime` CLI (apt-installed on +the peer) via the SSH-push transport. It does NOT import +`bytecodealliance/wasmtime-go` (or any other CGO package). + +## C-01 Gate Status + +**SATISFIED.** C-01 is satisfied: +- wasmtime works without CGO (CLI-via-SSH pattern, identical to podman/qm/pct). +- D-002 cross-compile story preserved (no CGO introduced anywhere in + the runtime package or any orca Go code). +- D-002 is NOT revoked. + +## Recorded as D-187 + +See PROJECT.md v0.9 D-series: "wasmtime Go binding is CGO-based +(bytecodealliance/wasmtime-go); orca uses the wasmtime CLI via SSH +(apt-installed on peer) instead of the Go binding, avoiding CGO +entirely. D-002 cross-compile story preserved. C-01 satisfied." + +## Verification + +- `go build ./internal/runtime/` succeeds with `CGO_ENABLED=0`. +- `internal/runtime/wasm.go` imports only stdlib + sshpush (no + wasmtime-go). +- The full test suite (`go test ./...`) does not require CGO. \ No newline at end of file diff --git a/internal/runtime/podman.go b/internal/runtime/podman.go new file mode 100644 index 0000000..1586821 --- /dev/null +++ b/internal/runtime/podman.go @@ -0,0 +1,142 @@ +package runtime + +import ( + "context" + "fmt" + "strings" + + "git.cloudinit.dev/coreci/orca/internal/sshpush" +) + +// PodmanRuntime implements Runtime for the "podman" one_of. It runs +// `podman` on the peer over the SSH-push transport (P01). The +// transport is injected via the constructor (dependency injection). +// +// Container lifecycle: +// +// - Prepare: `podman pull ` +// - Start: `podman run -d --name orca- ` +// - Stop: `podman stop ` then `podman rm ` +// - Status: `podman inspect --format '{{.State.Running}}' ` +// +// The container name is `orca-` (sanitized to lowercase + +// alnum). The runtime keeps no in-process state — each call is a fresh +// SSH exec against the peer. +type PodmanRuntime struct { + transport *sshpush.Transport +} + +// NewPodmanRuntime returns a PodmanRuntime backed by the given transport. +func NewPodmanRuntime(t *sshpush.Transport) *PodmanRuntime { + return &PodmanRuntime{transport: t} +} + +// containerName returns the deterministic container name for an alloc. +func containerName(alloc *Alloc) string { + id := strings.ToLower(alloc.ID) + id = strings.Map(func(r rune) rune { + if r >= 'a' && r <= 'z' || r >= '0' && r <= '9' || r == '-' || r == '_' { + return r + } + return '-' + }, id) + return "orca-" + id +} + +// Prepare pulls the image on the peer. +func (p *PodmanRuntime) Prepare(ctx context.Context, alloc *Alloc) error { + image, err := imageFor(alloc) + if err != nil { + return err + } + cmd := fmt.Sprintf("podman pull %q", image) + if _, err := p.transport.Exec(ctx, alloc.Node, cmd); err != nil { + return fmt.Errorf("podman: pull: %w", err) + } + return nil +} + +// Start runs `podman run -d --name `. +func (p *PodmanRuntime) Start(ctx context.Context, alloc *Alloc) (int, error) { + image, err := imageFor(alloc) + if err != nil { + return 0, err + } + cmdStr, _ := commandFor(alloc) + name := containerName(alloc) + cmd := fmt.Sprintf("podman run -d --name %s %q %s", name, image, cmdStr) + out, err := p.transport.Exec(ctx, alloc.Node, cmd) + if err != nil { + return 0, fmt.Errorf("podman: run: %w", err) + } + // The container ID is the first 12 chars of the printed hash. We + // don't keep it — Stop/Status use the name — but return a stable + // synthetic PID derived from the first 4 bytes of the hash for + // the interface contract. + cid := strings.TrimSpace(string(out)) + return podmanCidToPID(cid), nil +} + +// Stop stops and removes the container. +func (p *PodmanRuntime) Stop(ctx context.Context, alloc *Alloc) error { + name := containerName(alloc) + if _, err := p.transport.Exec(ctx, alloc.Node, fmt.Sprintf("podman stop %s", name)); err != nil { + return fmt.Errorf("podman: stop: %w", err) + } + if _, err := p.transport.Exec(ctx, alloc.Node, fmt.Sprintf("podman rm %s", name)); err != nil { + return fmt.Errorf("podman: rm: %w", err) + } + return nil +} + +// Status inspects the container's running state. +func (p *PodmanRuntime) Status(ctx context.Context, alloc *Alloc) (State, error) { + name := containerName(alloc) + cmd := fmt.Sprintf("podman inspect --format '{{.State.Running}}' %s", name) + out, err := p.transport.Exec(ctx, alloc.Node, cmd) + if err != nil { + return StateFailed, fmt.Errorf("podman: inspect: %w", err) + } + v := strings.TrimSpace(string(out)) + switch v { + case "true": + return StateRunning, nil + case "false": + return StateStopped, nil + default: + return StateFailed, fmt.Errorf("podman: unexpected inspect output %q", v) + } +} + +// podmanCidToPID converts a container ID (hex hash) to a positive int +// PID for the interface contract. It reads up to 4 hex chars. +func podmanCidToPID(cid string) int { + if len(cid) < 1 { + return 1 + } + n := len(cid) + if n > 4 { + n = 4 + } + var pid int + for i := 0; i < n; i++ { + c := cid[i] + pid = (pid << 4) | int(hexVal(c)) + } + if pid <= 0 { + pid = 1 + } + return pid +} + +func hexVal(c byte) byte { + switch { + case c >= '0' && c <= '9': + return c - '0' + case c >= 'a' && c <= 'f': + return c - 'a' + 10 + case c >= 'A' && c <= 'F': + return c - 'A' + 10 + } + return 0 +} diff --git a/internal/runtime/podman_test.go b/internal/runtime/podman_test.go new file mode 100644 index 0000000..1db1ab2 --- /dev/null +++ b/internal/runtime/podman_test.go @@ -0,0 +1,229 @@ +package runtime + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" + "git.cloudinit.dev/coreci/orca/internal/sshpush" +) + +// allocNoImage returns an alloc whose Spec has a Runtime block with a +// command but no image. +func allocNoImage(runtime string) *Alloc { + return &Alloc{ + ID: "x", + Runtime: runtime, + Spec: &jobspec.WorkloadSpec{ + Runtime: &jobspec.RuntimeBlock{Command: "/bin/true"}, + }, + } +} + +// TestPodmanRuntime_HappyPath wires a fake server that responds to +// podman pull/run/stop/rm/inspect and verifies the full lifecycle. +func TestPodmanRuntime_HappyPath(t *testing.T) { + srv := newFakeServer(t) + defer srv.close() + + const cid = "abc123def456" + srv.setHandler("podman pull", func(cmd string) ([]byte, int) { return nil, 0 }) + srv.setHandler("podman run", func(cmd string) ([]byte, int) { return []byte(cid + "\n"), 0 }) + srv.setHandler("podman stop", func(cmd string) ([]byte, int) { return nil, 0 }) + srv.setHandler("podman rm", func(cmd string) ([]byte, int) { return nil, 0 }) + srv.setHandler("podman inspect", func(cmd string) ([]byte, int) { + return []byte("true\n"), 0 + }) + + tr := realTransport(t, srv) + defer tr.Close() + + p := NewPodmanRuntime(tr) + a := allocWithNode("podman", "docker.io/library/alpine:latest", "sleep 30", srv.addr()) + + ctx, cancel := withTimeout(10 * time.Second) + defer cancel() + if err := p.Prepare(ctx, a); err != nil { + t.Fatalf("Prepare: %v", err) + } + pid, err := p.Start(ctx, a) + if err != nil { + t.Fatalf("Start: %v", err) + } + if pid <= 0 { + t.Fatalf("pid = %d, want > 0", pid) + } + st, err := p.Status(ctx, a) + if err != nil { + t.Fatalf("Status: %v", err) + } + if st != StateRunning { + t.Errorf("Status = %q, want running", st) + } + if err := p.Stop(ctx, a); err != nil { + t.Fatalf("Stop: %v", err) + } +} + +// TestPodmanRuntime_StatusFalse verifies Status returns stopped when +// the container reports running=false. +func TestPodmanRuntime_StatusFalse(t *testing.T) { + srv := newFakeServer(t) + defer srv.close() + srv.setHandler("podman inspect", func(cmd string) ([]byte, int) { + return []byte("false\n"), 0 + }) + tr := realTransport(t, srv) + defer tr.Close() + p := NewPodmanRuntime(tr) + a := allocWithNode("podman", "img", "sleep 1", srv.addr()) + st, err := p.Status(context.Background(), a) + if err != nil { + t.Fatalf("Status: %v", err) + } + if st != StateStopped { + t.Errorf("Status = %q, want stopped", st) + } +} + +// TestPodmanRuntime_StatusBadOutput verifies Status returns failed on +// unexpected inspect output. +func TestPodmanRuntime_StatusBadOutput(t *testing.T) { + srv := newFakeServer(t) + defer srv.close() + srv.setHandler("podman inspect", func(cmd string) ([]byte, int) { + return []byte("garbage\n"), 0 + }) + tr := realTransport(t, srv) + defer tr.Close() + p := NewPodmanRuntime(tr) + a := allocWithNode("podman", "img", "sleep 1", srv.addr()) + if _, err := p.Status(context.Background(), a); err == nil { + t.Error("Status with bad output should error") + } +} + +// TestPodmanRuntime_PrepareNoImage verifies Prepare errors when the +// alloc has no image. +func TestPodmanRuntime_PrepareNoImage(t *testing.T) { + p := NewPodmanRuntime(nil) + a := allocNoImage("podman") + if err := p.Prepare(context.Background(), a); err == nil { + t.Error("Prepare with no image should error") + } +} + +// TestPodmanRuntime_PrepareTransportError verifies Prepare propagates a +// transport error (podman pull fails). +func TestPodmanRuntime_PrepareTransportError(t *testing.T) { + srv := newFakeServer(t) + defer srv.close() + srv.setHandler("podman pull", func(cmd string) ([]byte, int) { + return []byte("manifest unknown\n"), 2 + }) + tr := realTransport(t, srv) + defer tr.Close() + p := NewPodmanRuntime(tr) + a := allocWithNode("podman", "img", "sleep 1", srv.addr()) + if err := p.Prepare(context.Background(), a); err == nil { + t.Error("Prepare with failed pull should error") + } +} + +// TestPodmanRuntime_StartNoImage verifies Start errors with no image. +func TestPodmanRuntime_StartNoImage(t *testing.T) { + p := NewPodmanRuntime(nil) + a := allocNoImage("podman") + if _, err := p.Start(context.Background(), a); err == nil { + t.Error("Start with no image should error") + } +} + +// TestPodmanRuntime_StopTransportError verifies Stop propagates errors. +func TestPodmanRuntime_StopTransportError(t *testing.T) { + srv := newFakeServer(t) + defer srv.close() + srv.setHandler("podman stop", func(cmd string) ([]byte, int) { + return []byte("no such container\n"), 1 + }) + tr := realTransport(t, srv) + defer tr.Close() + p := NewPodmanRuntime(tr) + a := allocWithNode("podman", "img", "sleep 1", srv.addr()) + if err := p.Stop(context.Background(), a); err == nil { + t.Error("Stop with missing container should error") + } +} + +// TestPodmanRuntime_StatusInspectError verifies Status returns failed +// when inspect itself errors. +func TestPodmanRuntime_StatusInspectError(t *testing.T) { + srv := newFakeServer(t) + defer srv.close() + srv.setHandler("podman inspect", func(cmd string) ([]byte, int) { + return []byte("no such container\n"), 1 + }) + tr := realTransport(t, srv) + defer tr.Close() + p := NewPodmanRuntime(tr) + a := allocWithNode("podman", "img", "sleep 1", srv.addr()) + st, err := p.Status(context.Background(), a) + if err == nil { + t.Error("Status with inspect error should error") + } + if st != StateFailed { + t.Errorf("Status = %q, want failed", st) + } +} + +// TestPodmanCidToPID verifies the synthetic PID derivation. +func TestPodmanCidToPID(t *testing.T) { + if got := podmanCidToPID(""); got != 1 { + t.Errorf("empty cid -> %d, want 1", got) + } + if got := podmanCidToPID("a"); got <= 0 { + t.Errorf("single hex -> %d, want > 0", got) + } + if got := podmanCidToPID("abcd"); got <= 0 { + t.Errorf("abcd -> %d, want > 0", got) + } + // non-hex chars fall through to 0 contributions but still yield + // a positive result (>= 1 by the floor). + if got := podmanCidToPID("xyz123"); got <= 0 { + t.Errorf("xyz123 -> %d, want > 0", got) + } +} + +// TestContainerNameSanitization verifies the container-name sanitizer +// uppercases and strips disallowed characters. +func TestContainerNameSanitization(t *testing.T) { + a := &Alloc{ID: "ALLOC_1.2.3", Spec: nil, Runtime: "podman"} + got := containerName(a) + if !strings.HasPrefix(got, "orca-") { + t.Errorf("containerName = %q, want orca- prefix", got) + } + if strings.Contains(got, ".") { + t.Errorf("containerName = %q, should not contain '.'", got) + } +} + +// TestPodmanRuntime_DialError verifies Prepare fails fast when the +// peer is unreachable (no fake server). +func TestPodmanRuntime_DialError(t *testing.T) { + srv := newFakeServer(t) + // close immediately so dial fails. + srv.close() + tr := realTransport(t, srv) + defer tr.Close() + p := NewPodmanRuntime(tr) + a := allocWithNode("podman", "img", "sleep 1", srv.addr()) + if err := p.Prepare(context.Background(), a); err == nil { + t.Error("Prepare against dead peer should error") + } else if !errors.Is(err, sshpush.ErrTransient) && !errors.Is(err, sshpush.ErrPermanent) { + // acceptable: either transient (retry exhausted) or permanent. + t.Logf("Prepare err (acceptable): %v", err) + } +} diff --git a/internal/runtime/process.go b/internal/runtime/process.go new file mode 100644 index 0000000..0c17bac --- /dev/null +++ b/internal/runtime/process.go @@ -0,0 +1,204 @@ +package runtime + +import ( + "context" + "errors" + "fmt" + "os" + "os/exec" + "strconv" + "strings" + "sync" + "syscall" + "time" +) + +// StopGrace is the default grace period between SIGTERM and SIGKILL +// for ProcessRuntime.Stop (10s, matching the systemd default TimeoutStopSec). +const StopGrace = 10 * time.Second + +// ProcessRuntime implements Runtime for the "process" one_of using +// os/exec. It is the in-process equivalent of the systemd unit the CLI +// emits in production: the CLI emits a .service file and the peer's +// systemd runs the process; ProcessRuntime starts the process directly +// in the current Go process. It is intended for LOCAL testing and +// hermetic CI — NOT for production (production uses the systemd emitter +// + the peer's systemd, not this in-process path). +// +// It tracks started PIDs in an in-memory map; restarts of the CLI lose +// that state (acceptable for the local-test use case). +type ProcessRuntime struct { + mu sync.Mutex + pids map[string]int // alloc.ID -> PID + procs map[int]*os.Process // PID -> process handle + stopped map[string]bool // alloc.ID -> reported stopped after Stop +} + +// NewProcessRuntime returns a ProcessRuntime. +func NewProcessRuntime() *ProcessRuntime { + return &ProcessRuntime{ + pids: make(map[string]int), + procs: make(map[int]*os.Process), + stopped: make(map[string]bool), + } +} + +// Prepare is a no-op for the process runtime: the systemd unit is +// emitted by the systemd emitter (internal/emitter), not by the runtime. +func (p *ProcessRuntime) Prepare(ctx context.Context, alloc *Alloc) error { + _ = ctx + _ = alloc + return nil +} + +// Start execs the alloc's command and returns the PID. The process is +// left running in the background; Stop terminates it. +func (p *ProcessRuntime) Start(ctx context.Context, alloc *Alloc) (int, error) { + cmdStr, err := commandFor(alloc) + if err != nil { + return 0, err + } + // Parse the command string into argv. A leading "exec" form + // (shell-style) is NOT supported — the command must be a direct + // argv[0] + args. Split on whitespace (simple, matches the existing + // executor.go behaviour which takes Command + Args separately). + parts := strings.Fields(cmdStr) + if len(parts) == 0 { + return 0, fmt.Errorf("process: empty command for alloc %s", alloc.ID) + } + + // Use a detached context so the process survives the request + // context cancellation (the request ends; the workload keeps + // running until Stop). We apply our own timeout for Start only. + startCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + cmd := exec.CommandContext(startCtx, parts[0], parts[1:]...) + // Detach the child from the parent's process group so it survives. + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + // Discard output for the runtime; the systemd unit captures logs + // in production. For tests, callers that need output run their + // own exec.Command. + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + return 0, fmt.Errorf("process: start: %w", err) + } + pid := cmd.Process.Pid + + // Background-reap the process so it doesn't become a zombie; we + // only need the PID for Stop/Status. When the process exits + // naturally, mark the alloc as stopped. + go func() { + _ = cmd.Wait() + p.mu.Lock() + delete(p.procs, pid) + // Only mark stopped if the alloc is still associated with + // this PID (Stop may have already removed the mapping). + if cur, ok := p.pids[alloc.ID]; ok && cur == pid { + delete(p.pids, alloc.ID) + p.stopped[alloc.ID] = true + } + p.mu.Unlock() + }() + + p.mu.Lock() + p.pids[alloc.ID] = pid + p.procs[pid] = cmd.Process + delete(p.stopped, alloc.ID) + p.mu.Unlock() + + return pid, nil +} + +// Stop sends SIGTERM, waits the grace period, then SIGKILL. +func (p *ProcessRuntime) Stop(ctx context.Context, alloc *Alloc) error { + p.mu.Lock() + pid, ok := p.pids[alloc.ID] + proc := p.procs[pid] + p.mu.Unlock() + if !ok || proc == nil { + return nil // not running; idempotent + } + // SIGTERM the process group (negative PID). + _ = syscall.Kill(-pid, syscall.SIGTERM) + + grace := StopGrace + if dl, ok := ctx.Deadline(); ok { + if remaining := time.Until(dl); remaining > 0 && remaining < grace { + grace = remaining + } + } + deadline := time.Now().Add(grace) + for time.Now().Before(deadline) { + if !p.alive(pid) { + p.forget(alloc.ID, pid) + return nil + } + select { + case <-ctx.Done(): + p.forget(alloc.ID, pid) + return ctx.Err() + case <-time.After(100 * time.Millisecond): + } + } + // SIGKILL the group. + _ = syscall.Kill(-pid, syscall.SIGKILL) + p.forget(alloc.ID, pid) + return nil +} + +// Status reports the alloc's state by checking if the process is alive. +func (p *ProcessRuntime) Status(ctx context.Context, alloc *Alloc) (State, error) { + _ = ctx + p.mu.Lock() + pid, ok := p.pids[alloc.ID] + wasStopped := p.stopped[alloc.ID] + p.mu.Unlock() + if !ok { + if wasStopped { + return StateStopped, nil + } + return StatePending, nil + } + if !p.alive(pid) { + // Process exited but the reaper hasn't run yet; mark it + // stopped and clean up. + p.forget(alloc.ID, pid) + return StateStopped, nil + } + return StateRunning, nil +} + +// alive reports whether the process with the given PID is still running. +func (p *ProcessRuntime) alive(pid int) bool { + proc, err := os.FindProcess(pid) + if err != nil { + return false + } + if err := proc.Signal(syscall.Signal(0)); err != nil { + // ESRCH means the process is gone. + return false + } + return true +} + +// forget removes the alloc/PID mapping and marks the alloc stopped. +func (p *ProcessRuntime) forget(allocID string, pid int) { + p.mu.Lock() + delete(p.pids, allocID) + delete(p.procs, pid) + p.stopped[allocID] = true + p.mu.Unlock() +} + +// PID returns the recorded PID for alloc (for tests/inspection). +func (p *ProcessRuntime) PID(allocID string) (int, error) { + p.mu.Lock() + defer p.mu.Unlock() + pid, ok := p.pids[allocID] + if !ok { + return 0, errors.New("process: no pid for alloc " + strconv.Quote(allocID)) + } + return pid, nil +} diff --git a/internal/runtime/process_test.go b/internal/runtime/process_test.go new file mode 100644 index 0000000..437bf56 --- /dev/null +++ b/internal/runtime/process_test.go @@ -0,0 +1,172 @@ +package runtime + +import ( + "context" + "runtime" + "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +// TestProcessRuntime_PrepareIsNoop verifies Prepare is a no-op. +func TestProcessRuntime_PrepareIsNoop(t *testing.T) { + p := NewProcessRuntime() + a := alloc("process", "", "/bin/true") + if err := p.Prepare(context.Background(), a); err != nil { + t.Fatalf("Prepare: %v", err) + } +} + +// TestProcessRuntime_PrepareNilSpec exercises the nil-spec branch +// indirectly — Prepare is a no-op regardless of input. +func TestProcessRuntime_PrepareNilSpec(t *testing.T) { + p := NewProcessRuntime() + if err := p.Prepare(context.Background(), &Alloc{ID: "x"}); err != nil { + t.Fatalf("Prepare (nil spec) should still be no-op: %v", err) + } +} + +// TestProcessRuntime_StartStopStatus runs a real long-lived process +// (sleep) and verifies Start -> Status(running) -> Stop -> Status(stopped). +func TestProcessRuntime_StartStopStatus(t *testing.T) { + if _, err := sleepBin(); err != nil { + t.Skipf("sleep binary not available: %v", err) + } + p := NewProcessRuntime() + sleepCmd, _ := sleepBin() + a := alloc("process", "", sleepCmd+" 30") + + ctx, cancel := withTimeout(10 * time.Second) + defer cancel() + pid, err := p.Start(ctx, a) + if err != nil { + t.Fatalf("Start: %v", err) + } + if pid <= 0 { + t.Fatalf("pid = %d, want > 0", pid) + } + + st, err := p.Status(context.Background(), a) + if err != nil { + t.Fatalf("Status: %v", err) + } + if st != StateRunning { + t.Errorf("Status = %q, want running", st) + } + + // Verify PID lookup. + got, err := p.PID(a.ID) + if err != nil || got != pid { + t.Errorf("PID = %d/%v, want %d", got, err, pid) + } + + stopCtx, cancelStop := withTimeout(15 * time.Second) + defer cancelStop() + if err := p.Stop(stopCtx, a); err != nil { + t.Fatalf("Stop: %v", err) + } + + st2, _ := p.Status(context.Background(), a) + if st2 != StateStopped { + t.Errorf("Status after Stop = %q, want stopped", st2) + } +} + +// TestProcessRuntime_StopNotStarted verifies Stop is idempotent on a +// never-started alloc. +func TestProcessRuntime_StopNotStarted(t *testing.T) { + p := NewProcessRuntime() + a := alloc("process", "", "/bin/true") + ctx, cancel := withTimeout(2 * time.Second) + defer cancel() + if err := p.Stop(ctx, a); err != nil { + t.Errorf("Stop on never-started alloc should be no-op, got %v", err) + } +} + +// TestProcessRuntime_StatusPending verifies Status returns pending for +// an alloc that was never started. +func TestProcessRuntime_StatusPending(t *testing.T) { + p := NewProcessRuntime() + a := alloc("process", "", "/bin/true") + st, err := p.Status(context.Background(), a) + if err != nil { + t.Fatalf("Status: %v", err) + } + if st != StatePending { + t.Errorf("Status = %q, want pending", st) + } +} + +// TestProcessRuntime_StartNoCommand verifies Start errors on missing +// command. +func TestProcessRuntime_StartNoCommand(t *testing.T) { + p := NewProcessRuntime() + a := &Alloc{ID: "x", Spec: nil, Runtime: "process"} + if _, err := p.Start(context.Background(), a); err == nil { + t.Error("Start with nil spec should error") + } +} + +// TestProcessRuntime_StartEmptyCommand verifies Start errors when +// the command parses to zero argv. +func TestProcessRuntime_StartEmptyCommand(t *testing.T) { + p := NewProcessRuntime() + a := &Alloc{ + ID: "e", + Runtime: "process", + Spec: &jobspec.WorkloadSpec{ + Runtime: &jobspec.RuntimeBlock{Command: " "}, + }, + } + _, err := p.Start(context.Background(), a) + if err == nil { + t.Error("Start with empty command should error") + } +} + +// TestProcessRuntime_StartBadBinary verifies Start propagates exec +// errors for a missing binary. +func TestProcessRuntime_StartBadBinary(t *testing.T) { + p := NewProcessRuntime() + a := alloc("process", "", "/no/such/binary/here") + _, err := p.Start(context.Background(), a) + if err == nil { + t.Error("Start with missing binary should error") + } +} + +// TestProcessRuntime_StopOnFinishedProcess verifies Stop on a process +// that already exited (e.g. /bin/true) is a no-op (no error). +func TestProcessRuntime_StopOnFinishedProcess(t *testing.T) { + p := NewProcessRuntime() + a := alloc("process", "", "/bin/true") + _, _ = p.Start(context.Background(), a) + // Give /bin/true time to exit. + time.Sleep(200 * time.Millisecond) + ctx, cancel := withTimeout(5 * time.Second) + defer cancel() + if err := p.Stop(ctx, a); err != nil { + t.Errorf("Stop on exited process should be no-op, got %v", err) + } +} + +// TestProcessRuntime_PIDUnknown verifies PID lookup errors for an +// unknown alloc. +func TestProcessRuntime_PIDUnknown(t *testing.T) { + p := NewProcessRuntime() + if _, err := p.PID("nope"); err == nil { + t.Error("PID unknown should error") + } +} + +// sleepBin returns the sleep command path ("sleep") on this OS. +func sleepBin() (string, error) { + // /bin/sleep exists on Linux; on other platforms fall back to + // "sleep" (resolved via PATH). + if runtime.GOOS == "linux" { + return "/bin/sleep", nil + } + return "sleep", nil +} diff --git a/internal/runtime/pve.go b/internal/runtime/pve.go new file mode 100644 index 0000000..eb1983c --- /dev/null +++ b/internal/runtime/pve.go @@ -0,0 +1,177 @@ +package runtime + +import ( + "context" + "fmt" + "strings" + + "git.cloudinit.dev/coreci/orca/internal/sshpush" +) + +// vmidFor returns a deterministic 5-digit VMID derived from the alloc +// ID (hash(alloc.ID) % 99999 + 1). Used by both pveVMRuntime and +// pveCTRuntime — VM and container IDs share the same numeric space on +// a Proxmox node, but the namespace is sparse (one alloc = one VMID) +// so collisions are rare in practice. +func vmidFor(alloc *Alloc) int { + id := allocIDHash(alloc.ID) + if id < 100 { + id += 100 + } + return id +} + +// PveVMRuntime implements Runtime for the "pve-vm" one_of using the +// `qm` tool over SSH on a Proxmox peer (extends REQ-076). The transport +// is injected. +// +// Lifecycle: +// +// - Prepare: `qm create --memory --cores --scsi0 ` +// - Start: `qm start ` +// - Stop: `qm shutdown ` (graceful) then `qm stop ` (force) +// - Status: `qm status ` +type PveVMRuntime struct { + transport *sshpush.Transport +} + +// NewPveVMRuntime returns a PveVMRuntime backed by the given transport. +func NewPveVMRuntime(t *sshpush.Transport) *PveVMRuntime { + return &PveVMRuntime{transport: t} +} + +// Prepare creates the VM. Memory/Cores default to 512MB / 1 if the +// spec's Runtime block omits them; the disk is the spec's image field +// (a Proxmox storage path like local:vmdir/disk.qcow2). +func (v *PveVMRuntime) Prepare(ctx context.Context, alloc *Alloc) error { + if alloc == nil || alloc.Spec == nil || alloc.Spec.Runtime == nil { + return fmt.Errorf("pve-vm: nil alloc/spec/runtime") + } + vmid := vmidFor(alloc) + image := alloc.Spec.Runtime.Image + if image == "" { + return fmt.Errorf("pve-vm: alloc %s has no disk (Runtime.Image)", alloc.ID) + } + mb := 512 + cores := 1 + cmd := fmt.Sprintf("qm create %d --memory %d --cores %d --scsi0 %q", vmid, mb, cores, image) + if _, err := v.transport.Exec(ctx, alloc.Node, cmd); err != nil { + return fmt.Errorf("pve-vm: create: %w", err) + } + return nil +} + +// Start boots the VM. +func (v *PveVMRuntime) Start(ctx context.Context, alloc *Alloc) (int, error) { + vmid := vmidFor(alloc) + if _, err := v.transport.Exec(ctx, alloc.Node, fmt.Sprintf("qm start %d", vmid)); err != nil { + return 0, fmt.Errorf("pve-vm: start: %w", err) + } + return vmid, nil +} + +// Stop gracefully shuts down then force-stops the VM. +func (v *PveVMRuntime) Stop(ctx context.Context, alloc *Alloc) error { + vmid := vmidFor(alloc) + if _, err := v.transport.Exec(ctx, alloc.Node, fmt.Sprintf("qm shutdown %d", vmid)); err != nil { + // Best-effort graceful; fall through to force stop. + } + if _, err := v.transport.Exec(ctx, alloc.Node, fmt.Sprintf("qm stop %d", vmid)); err != nil { + return fmt.Errorf("pve-vm: stop: %w", err) + } + return nil +} + +// Status reports the VM state from `qm status`. +func (v *PveVMRuntime) Status(ctx context.Context, alloc *Alloc) (State, error) { + vmid := vmidFor(alloc) + out, err := v.transport.Exec(ctx, alloc.Node, fmt.Sprintf("qm status %d", vmid)) + if err != nil { + return StateFailed, fmt.Errorf("pve-vm: status: %w", err) + } + s := strings.ToLower(string(out)) + switch { + case strings.Contains(s, "running"): + return StateRunning, nil + case strings.Contains(s, "stopped"): + return StateStopped, nil + default: + return StatePending, nil + } +} + +// PveCTRuntime implements Runtime for the "pve-ct" one_of using the +// `pct` tool over SSH on a Proxmox peer (extends REQ-076). +// +// Lifecycle: +// +// - Prepare: `pct create