Files
Privatae 081763a675
ci / personal-data (push) Successful in 4s
ci / test (push) Successful in 4m30s
ci / fuzz (push) Successful in 3m25s
bastion 0.57.0
2026-10-28 20:35:45 +00:00

2002 lines
68 KiB
Go

package main
// coders.go — CODER MODE: one sealed coding run per container on this machine.
//
// A Host can place a coding run (codex, claude-code, ...) on a machine its
// user owns instead of the platform pool. The run reaches the internet over
// this machine's own connection, and nothing else: the container has no
// network at all, and its only way out is egress.sock, a proxy in this bastion
// that refuses this machine, its LAN and every private range (coder_egress.go).
// The Host reaches the run's control plane through one tunnel port per run
// (coder_forward.go), which accepts only the owning pairing's tunnel address.
//
// THE VERBS (mode "coder" on the harness lane, coder_op):
// - run-up start one container for one run and return as soon as it
// answers /health. The harness worker is released then; the
// run lives on without holding one, so the 2-worker pool
// (harness.go harnessWorkerCount) does not bound how many runs
// this machine holds.
// - run-down stop it, report what it used (cgroup memory.peak, cpu.stat),
// remove it and its files.
// - run-status what this pairing has running here.
//
// WHAT THE OWNER DECIDES. `privatae-bastion coders set|on|off` stores the
// machine-wide limits (how many runs, how big each one is, which engines). It
// is off by default. A run that does not fit is answered "full" — a capacity
// answer with exit 0, never a refusal, so a burst cannot trip the quarantine.
//
// Linux only in this cut (the socket forward pins the inode through /proc,
// service_sock_linux.go), and the bastion must run as root because the run's
// files are handed to the engine's uid instead of being opened 0777.
import (
"context"
"crypto/sha256"
_ "embed"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"math"
"net"
"net/http"
"os"
"os/exec"
"path/filepath"
"regexp"
"runtime"
"sort"
"strconv"
"strings"
"sync"
"time"
)
const (
coderCapability = "coders"
coderGateOp = "mode:coder"
coderContainerPrefix = "privatae-run-"
// Tunnel-side ports, one per run. The service slot owns 18100-18199
// (service.go); these sit right after it.
coderPortBase = 18200
coderPortMax = 18299
coderLabelRun = "privatae.coder.run"
coderLabelOwner = "privatae.coder.owner"
coderLabelEngine = "privatae.coder.engine"
coderLabelDeadline = "privatae.coder.deadline"
// Paths inside the container. The sockets are mounted as DIRECTORIES: a
// listener's socket file does not exist before the container starts, so it
// cannot be bind-mounted itself.
coderCtlDirIn = "/run/privatae/ctl"
coderEgressDirIn = "/run/privatae/egress"
coderControlSock = "control.sock"
coderEgressSock = "egress.sock"
coderWorkspaceIn = "/var/lib/skema/projects"
coderHomeIn = "/home/privatae"
coderSeccompFile = "seccomp-personal.json"
coderDefaultMins = 240
coderDeadlineSlop = 15 * time.Minute
// Defaults until the load test (plan section 4) sets them.
coderDefaultMemMB = 2048
coderDefaultCPUs = 2.0
// THE MACHINE GUARD. Runs must never take the memory the machine itself
// needs: on a 2-CPU 3.7 GiB machine, twelve 512 MiB runs were each inside
// their own limit and together left 0 MiB free, and sshd stopped answering
// until the machine was rebooted. So run-up counts every run's limit
// against what the machine can spare (coderReserveMB), and the watchdog
// (coderWatch) stops the newest run when the machine runs short anyway,
// because something outside the runs grew.
//
// coderEvictFloorMB is the kubelet's and GKE's hard eviction threshold
// (memory.available < 100Mi). The pressure limit and its duration are
// systemd-oomd's defaults: PSI memory "full" avg10 over 60% for 30 s.
coderEvictFloorMB = 100
coderPressureLimit = 60.0
coderPressureFor = 30 * time.Second
coderPressureSettle = 10 * time.Second // one avg10 window: what a stop did shows up after this
coderWatchEvery = 2 * time.Second
coderStartWait = 60 * time.Second
coderCredsMax = 1 << 20
coderPullTimeout = 10 * time.Minute
)
// coderEngines are the coding engines a run may name: the controller's
// _HARNESS_SPECS rows minus browser, and workspace, the harness control plane
// with a toolchain and no coding agent, which a Host's own threads build and
// test in. vscode is retired and is not one.
var coderEngines = []string{"codex", "cursor", "grok", "claude-code", "opencode", "kimi", "antigravity", "workspace"}
// coderDefaultImagePrefix is where every engine's image lives unless the
// owner names another prefix (`coders set --image-prefix`). A run names
// <prefix>harness-<engine>@sha256:<digest>.
const coderDefaultImagePrefix = gitForgeHost + "/privatae-harness/"
var (
coderRunIDRe = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{6,61}[a-z0-9]$`)
coderDigestRe = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
coderCredsDirRe = regexp.MustCompile(`^/home/privatae/\.[a-z0-9-]{1,32}$`)
coderCredNameRe = regexp.MustCompile(`^[A-Za-z0-9._-]{1,64}$`)
coderCapRe = regexp.MustCompile(`^[A-Za-z0-9_-]{32,128}$`)
)
// seccomp-personal.json is the profile every pool harness pod runs under (the
// cluster's node addon installs the same bytes; a test pins the sha256). It is
// written next to the coder config and handed to the runtime on every run.
//
//go:embed seccomp-personal.json
var coderSeccompProfile []byte
// The seams. Package-level so the tests can drive every verb without a real
// container runtime, a Linux kernel or root.
var (
coderGOOS = runtime.GOOS
coderIsRoot = processIsRoot
coderRuntime = detectRuntime
coderRun = runCmd
coderRunEnv = runCmdEnv
coderMemInfo = coderReadMemInfo
coderPSIMem = coderReadPSIMem
coderChown = os.Chown
coderNow = time.Now
coderWaitUp = coderWaitHealthy
coderCgroupFS = "/sys/fs/cgroup"
coderProcFS = "/proc"
// coderNvidiaConfig is NVIDIA's container toolkit config (coderNvidiaSwarmEnv).
coderNvidiaConfig = "/etc/nvidia-container-runtime/config.toml"
coderEnvNameRe = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
)
// ── config ──────────────────────────────────────────────────────────────────
// coderConfig is <configDir>/coders/config.json. Machine-wide: the owner sizes
// the machine once, whichever hosts are paired.
type coderConfig struct {
Enabled bool `json:"enabled"`
Max int `json:"max"` // 0 = unset
MemPerRun int `json:"mem_per_run_mb"` // 0 = unset
CPUsPerRun float64 `json:"cpus_per_run"` // 0 = unset
Engines []string `json:"engines"` // nil = unset; stored explicitly, never empty-for-all
AllowLAN []string `json:"allow_lan"` // host[:port] the egress proxy lets through to a private address
// Where a run's image may come from, set by the owner (`coders set
// --image-prefix`). "" is coderDefaultImagePrefix, and is omitted from the
// file, so a machine that never set it keeps the bytes it had.
ImagePrefix string `json:"image_prefix,omitempty"`
}
// imagePrefix is the owner's prefix, or the default.
func (c coderConfig) imagePrefix() string {
if c.ImagePrefix == "" {
return coderDefaultImagePrefix
}
return c.ImagePrefix
}
var (
coderPrefixDomainRe = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)*(:[0-9]{1,5})?$`)
coderPrefixPathRe = regexp.MustCompile(`^[a-z0-9]+((\.|_|__|-+)[a-z0-9]+)*$`)
)
// coderPrefixRefusal checks an image prefix the owner names: image-reference
// components ending in "/", the first one a registry host (with an optional
// port) or a path component, lowercase, at most 200 bytes. "" when it is good.
func coderPrefixRefusal(p string) string {
if p == "" || len(p) > 200 || !strings.HasSuffix(p, "/") || strings.ContainsAny(p, "@ \t\r\n") {
return "--image-prefix must be a registry path ending in '/', like " + coderDefaultImagePrefix
}
for i, c := range strings.Split(strings.TrimSuffix(p, "/"), "/") {
if coderPrefixPathRe.MatchString(c) || (i == 0 && coderPrefixDomainRe.MatchString(c)) {
continue
}
return fmt.Sprintf("--image-prefix: %q is not a lowercase image-path component", c)
}
return ""
}
// coderPrefixRegistry is the registry a prefix pulls from, by the container
// runtimes' own rule: a first component with a '.' or ':', or localhost, is a
// registry host; anything else is on docker.io.
func coderPrefixRegistry(p string) string {
first, _, _ := strings.Cut(p, "/")
if strings.ContainsAny(first, ".:") || first == "localhost" {
return first
}
return "docker.io"
}
// coderImageRefusal: the image must be <prefix>harness-<engine>@sha256:<digest>,
// the engine's own image under the prefix the owner set.
func coderImageRefusal(image, engine, prefix string) string {
repo, digest, ok := strings.Cut(image, "@")
if !ok || repo != prefix+"harness-"+engine || !coderDigestRe.MatchString(digest) {
return "image must be " + prefix + "harness-" + engine + "@sha256:<digest>"
}
return ""
}
func codersDir() (string, error) {
d, err := configDir()
if err != nil {
return "", err
}
p := filepath.Join(d, "coders")
return p, os.MkdirAll(p, 0o700)
}
func coderConfigPath() (string, error) {
d, err := codersDir()
if err != nil {
return "", err
}
return filepath.Join(d, "config.json"), nil
}
// loadCoderConfig: a missing file is the default (off). A file that does not
// parse is also off — a box never runs coders on a config it cannot read.
func loadCoderConfig() coderConfig {
p, err := coderConfigPathNoCreate()
if err != nil {
return coderConfig{}
}
b, err := os.ReadFile(p)
if err != nil {
return coderConfig{}
}
var c coderConfig
if json.Unmarshal(b, &c) != nil {
return coderConfig{}
}
return c
}
// coderConfigPathNoCreate reads without creating the coders dir, so a box
// that never used coder mode gets no new directory from a heartbeat.
func coderConfigPathNoCreate() (string, error) {
d, err := configDirPath()
if err != nil {
return "", err
}
return filepath.Join(d, "coders", "config.json"), nil
}
func saveCoderConfig(c coderConfig) error {
p, err := coderConfigPath()
if err != nil {
return err
}
b, err := json.MarshalIndent(c, "", " ")
if err != nil {
return err
}
return atomicWrite(p, b)
}
// ready: on, and every size is set (`coders on` fills what `set` left).
func (c coderConfig) ready() bool {
return c.Enabled && c.Max >= 1 && c.MemPerRun >= 512 && c.CPUsPerRun > 0 && len(c.Engines) > 0
}
func (c coderConfig) allowsEngine(e string) bool {
return containsStr(c.Engines, e)
}
func isCoderEngine(e string) bool { return containsStr(coderEngines, e) }
// ── run records ─────────────────────────────────────────────────────────────
// coderRecord is R/run.json, written once the container answers /health and
// removed by run-down. The forward, the egress loop, the heartbeat and the
// reaper all read it.
type coderRecord struct {
RunID string `json:"run_id"`
Owner string `json:"owner"`
Engine string `json:"engine"`
Image string `json:"image"`
Container string `json:"container"`
TunnelPort int `json:"tunnel_port"`
UID int `json:"uid"`
CreatedAt string `json:"created_at"`
Deadline string `json:"deadline"`
PullS float64 `json:"pull_s"`
MemMB int `json:"mem_mb,omitempty"` // the run's memory limit; 0 in a record from before it was kept
Cgroup string `json:"cgroup,omitempty"` // the container's cgroup, as /proc/<pid>/cgroup names it
StoppedBy string `json:"stopped_by,omitempty"` // why the watchdog stopped it, "" when it did not
}
func coderRunsDir() (string, error) {
d, err := codersDir()
if err != nil {
return "", err
}
p := filepath.Join(d, "runs")
return p, os.MkdirAll(p, 0o700)
}
func coderRunDir(runID string) (string, error) {
d, err := coderRunsDir()
if err != nil {
return "", err
}
return filepath.Join(d, runID), nil
}
func loadCoderRecord(runID string) (*coderRecord, error) {
if !coderRunIDRe.MatchString(runID) {
return nil, errors.New("bad run_id")
}
R, err := coderRunDir(runID)
if err != nil {
return nil, err
}
b, err := os.ReadFile(filepath.Join(R, "run.json"))
if err != nil {
return nil, err
}
var r coderRecord
if err := json.Unmarshal(b, &r); err != nil {
return nil, err
}
if r.RunID != runID {
return nil, errors.New("record names another run")
}
return &r, nil
}
// loadCoderRecords returns every readable run record, sorted by run_id. It
// creates nothing: a box that never ran a coder answers an empty list.
func loadCoderRecords() []coderRecord {
d, err := configDirPath()
if err != nil {
return nil
}
ents, err := os.ReadDir(filepath.Join(d, "coders", "runs"))
if err != nil {
return nil
}
var out []coderRecord
for _, e := range ents {
if !e.IsDir() || !coderRunIDRe.MatchString(e.Name()) {
continue
}
if r, err := loadCoderRecord(e.Name()); err == nil {
out = append(out, *r)
}
}
sort.Slice(out, func(i, j int) bool { return out[i].RunID < out[j].RunID })
return out
}
func saveCoderRecord(R string, r *coderRecord) error {
b, err := json.MarshalIndent(r, "", " ")
if err != nil {
return err
}
return atomicWrite(filepath.Join(R, "run.json"), b)
}
// coderRewriteRecord replaces the record of a run that already has one. Unlike
// saveCoderRecord (atomicWrite makes missing directories) it never creates R:
// `coders drop` tears runs down from its own process, which coderMu does not
// cover, so R can go between reading a record and writing it back. Writing
// the temporary file inside R fails once R is gone, and the run stays gone.
func coderRewriteRecord(R string, r *coderRecord) error {
b, err := json.MarshalIndent(r, "", " ")
if err != nil {
return err
}
f, err := os.CreateTemp(R, ".run-*.tmp")
if err != nil {
return err
}
tmp := f.Name()
defer os.Remove(tmp) // nothing to remove once renamed
_, err = f.Write(b)
if err == nil {
err = f.Sync()
}
if cerr := f.Close(); err == nil {
err = cerr
}
if err != nil {
return err
}
return os.Rename(tmp, filepath.Join(R, "run.json"))
}
// ── capacity ────────────────────────────────────────────────────────────────
// coderMu guards the reservation table and every record add/remove, so two
// run-ups cannot both take the last slot or the same tunnel port.
var (
coderMu sync.Mutex
coderStarting = map[string]coderReservation{} // run_id -> a run-up in progress
)
type coderReservation struct {
owner string
port int
memMB int
}
// coderCounts: box-wide runs and this pairing's, records plus run-ups in
// flight. Caller holds coderMu.
func coderCountsLocked(self string, recs []coderRecord) (all, mine int) {
for _, r := range recs {
all++
if r.Owner == self {
mine++
}
}
for id, s := range coderStarting {
if coderHasRecord(recs, id) {
continue
}
all++
if s.owner == self {
mine++
}
}
return all, mine
}
func coderHasRecord(recs []coderRecord, id string) bool {
for _, r := range recs {
if r.RunID == id {
return true
}
}
return false
}
// allocCoderPort takes the lowest tunnel port no record or reservation holds.
// 0 = all 100 are taken. Caller holds coderMu.
func allocCoderPortLocked(recs []coderRecord) int {
used := map[int]bool{}
for _, r := range recs {
used[r.TunnelPort] = true
}
for _, s := range coderStarting {
used[s.port] = true
}
for p := coderPortBase; p <= coderPortMax; p++ {
if !used[p] {
return p
}
}
return 0
}
// coderReserveMB is the memory a machine of totalMB keeps for itself: GKE's
// node reservation (255 MiB under 1 GiB; otherwise 25% of the first 4 GiB, 20%
// of the next 4, 10% of the next 8, 6% of the next 112 and 2% above 128 GiB)
// plus the eviction threshold. On the 3.7 GiB machine that is 1055 MiB; a Host
// and its database there, with 64 threads in flight and the operating system,
// used under 1.1 GiB.
func coderReserveMB(totalMB int) int {
if totalMB < 1024 {
return 255 + coderEvictFloorMB
}
tiers := []struct {
upTo int
frac float64
}{{4 << 10, 0.25}, {8 << 10, 0.20}, {16 << 10, 0.10}, {128 << 10, 0.06}, {math.MaxInt32, 0.02}}
r, low := 0.0, 0
for _, t := range tiers {
if totalMB <= low {
break
}
r += float64(min(totalMB, t.upTo)-low) * t.frac
low = t.upTo
}
return int(math.Ceil(r)) + coderEvictFloorMB
}
// coderBudgetMB is what the limits of every run together may add up to.
func coderBudgetMB(totalMB int) int { return totalMB - coderReserveMB(totalMB) }
// coderRecordMem is one run's memory limit; a record from before the limit
// was kept counts at today's per-run size.
func coderRecordMem(r coderRecord, perRun int) int {
if r.MemMB > 0 {
return r.MemMB
}
return perRun
}
// coderCommittedLocked is the memory promised to runs: the limit of every run
// not known to have stopped, plus every run-up in flight. A stopped run's
// memory is free again, though its record stays for run-down to read. Caller
// holds coderMu.
func coderCommittedLocked(recs []coderRecord, perRun int) int {
sum := 0
for _, r := range recs {
if up, known := coderRecordRunning(r); known && !up {
continue
}
sum += coderRecordMem(r, perRun)
}
for id, s := range coderStarting {
if !coderHasRecord(recs, id) {
sum += s.memMB
}
}
return sum
}
// coderRecordRunning reads whether a run's container still has processes,
// from its cgroup's cgroup.events, without asking the container runtime: the
// runtime is slowest exactly when the machine is short of memory. known is
// false when the record carries no cgroup (one from before it was kept, or a
// runtime whose cgroup was not found).
func coderRecordRunning(r coderRecord) (running, known bool) {
dir := coderCgroupPath(r.Cgroup)
if dir == "" {
return false, false
}
b, err := os.ReadFile(filepath.Join(dir, "cgroup.events"))
if err != nil {
// The cgroup is gone with its container.
return false, os.IsNotExist(err)
}
for _, ln := range strings.Split(string(b), "\n") {
if f := strings.Fields(ln); len(f) == 2 && f[0] == "populated" {
return f[1] == "1", true
}
}
return false, false
}
// coderCgroupPath is a cgroup path as /proc/<pid>/cgroup names it ("0::/x"
// without the prefix), under coderCgroupFS. "" when it is empty or climbs out.
func coderCgroupPath(p string) string {
if p == "" || strings.Contains(p, "..") {
return ""
}
return filepath.Join(coderCgroupFS, filepath.Clean("/"+p))
}
// coderParsePSIFull reads "full avg10" from a /proc/pressure file: the share
// of the last 10 s in which every task was stalled.
func coderParsePSIFull(s string) (float64, bool) {
for _, ln := range strings.Split(s, "\n") {
f := strings.Fields(ln)
if len(f) < 2 || f[0] != "full" {
continue
}
for _, kv := range f[1:] {
if v, ok := strings.CutPrefix(kv, "avg10="); ok {
x, err := strconv.ParseFloat(v, 64)
return x, err == nil
}
}
}
return 0, false
}
// ── the lane ────────────────────────────────────────────────────────────────
// coderPull is the per-run registry credential the Host mints for this pull.
// It is used once, in a throwaway config dir, and never stored or logged.
type coderPull struct {
Registry string `json:"registry"`
Username string `json:"username"`
Password string `json:"password"`
}
// coderSpecRefusal is everything about the request itself, checked before the
// lane spends any gate (handleHarnessConn), like serviceSpecRefusal.
func coderSpecRefusal(spec harnessSpec) string {
switch spec.CoderOp {
case "run-status":
return ""
case "run-down":
if !coderRunIDRe.MatchString(spec.RunID) {
return "run_id must be 8-63 of a-z, 0-9 and '-', starting and ending with a letter or digit"
}
return ""
case "run-up":
return validateCoderUp(spec)
}
return "coder_op must be run-up, run-down or run-status"
}
func validateCoderUp(spec harnessSpec) string {
if !coderRunIDRe.MatchString(spec.RunID) {
return "run_id must be 8-63 of a-z, 0-9 and '-', starting and ending with a letter or digit"
}
if !isCoderEngine(spec.Engine) {
return "engine must be one of " + strings.Join(coderEngines, ", ")
}
prefix := loadCoderConfig().imagePrefix()
if why := coderImageRefusal(spec.Image, spec.Engine, prefix); why != "" {
return why
}
if spec.UID < 10001 || spec.UID > 10099 {
return "uid must be the engine's own uid (10001-10099)"
}
if !coderCredsDirRe.MatchString(spec.CredsDir) {
return "creds_dir must be /home/privatae/.<name>"
}
total := 0
for name, body := range spec.Creds {
if !coderCredNameRe.MatchString(name) || name == "." || name == ".." {
return "a creds file name is not a plain file name"
}
total += len(name) + len(body)
}
if total > coderCredsMax {
return "creds exceed 1 MiB"
}
if !coderCapRe.MatchString(spec.ControlCap) {
return "control_cap must be 32-128 of A-Z, a-z, 0-9, '_' and '-'"
}
if spec.MaxMinutes != 0 && (spec.MaxMinutes < 1 || spec.MaxMinutes > 1440) {
return "max_minutes must be 1-1440"
}
if p := spec.Pull; p != nil {
if reg := coderPrefixRegistry(prefix); p.Registry != reg {
return "pull.registry must be " + reg
}
if p.Username == "" || p.Password == "" {
return "pull needs a username and a password"
}
if strings.ContainsAny(p.Username, ":\x00\r\n") || strings.ContainsAny(p.Password, "\x00\r\n") {
return "pull.username or pull.password contains a character a registry login cannot hold"
}
}
return ""
}
// coderGate is harnessGate's branch for mode=coder on an ANCHORED box (the
// caller has already returned for an unanchored one). run-status is a read and
// answers even under quarantine. run-up and run-down are writes: the
// quarantine applies, and the pairing must hold a valid standing coder grant
// (coder_grant.go) — approved once by the person, never per run.
func coderGate(self, dcnDir string, spec harnessSpec) string {
if spec.CoderOp == "run-status" {
return ""
}
why := ""
now := coderNow()
if until := quarantinedUntil(self, now); !until.IsZero() {
why = quarantineRefusal(until)
} else if _, gwhy := coderGrantFor(dcnDir, self); gwhy != "" {
why = "refused: " + coderGateOp + " — " + gwhy
}
if why != "" {
noteRefusal(self, coderGateOp, why, refusalCounts(coderGateOp, why), now)
}
return why
}
func coderBase(spec harnessSpec) map[string]any {
res := map[string]any{"mode": "coder", "coder_op": spec.CoderOp}
if spec.RunID != "" {
res["run_id"] = spec.RunID
}
return res
}
func coderFail(res map[string]any, reason, msg string) ([]byte, int) {
res["ok"] = false
if reason != "" {
res["reason"] = reason
}
res["error"] = msg
b, _ := json.Marshal(res)
return b, 1
}
// runCoder dispatches one coder job. spec.owner is the pairing it arrived on.
func runCoder(spec harnessSpec) ([]byte, int) {
switch spec.CoderOp {
case "run-up":
return coderRunUp(spec)
case "run-down":
return coderRunDown(spec)
case "run-status":
return coderRunStatus(spec)
}
return coderFail(coderBase(spec), "", "coder_op must be run-up, run-down or run-status")
}
// coderHostReady is the part of run-up that does not depend on the request:
// the machine can run coders at all. Also the heartbeat's honesty check.
func coderHostReady() (string, string) {
if coderGOOS != "linux" {
return "", "coder runs need a Linux machine"
}
if !coderIsRoot() {
return "", "coder runs need the bastion running as root (the run's files are handed to the engine's own uid)"
}
rt := coderRuntime()
if rt == "" {
return "", "no container runtime (docker/podman/nerdctl) on this machine"
}
return rt, ""
}
func coderRunUp(spec harnessSpec) ([]byte, int) {
res := coderBase(spec)
rt, why := coderHostReady()
if why != "" {
return coderFail(res, "", why)
}
res["runtime"] = rt
cfg := loadCoderConfig()
if !cfg.ready() {
return coderFail(res, "", "coder runs are off on this machine (`privatae-bastion coders on`)")
}
if !cfg.allowsEngine(spec.Engine) {
return coderFail(res, "", "engine '"+spec.Engine+"' is not allowed on this machine")
}
var grantMax *int
if currentAnchor() != nil {
g, gwhy := coderGrantFor(spec.dcnDir, spec.owner)
if gwhy != "" {
return coderFail(res, "", gwhy)
}
if !containsStr(g.Engines, spec.Engine) {
return coderFail(res, "", "engine '"+spec.Engine+"' is not in this host's coder grant")
}
m := g.Max
grantMax = &m
}
if why := validateCoderUp(spec); why != "" {
return coderFail(res, "", why)
}
if why := coderTransportRefusal(spec.owner); why != "" {
return coderFail(res, "", why)
}
// Reserve a slot and a port, or answer full. The answer is a capacity
// answer: ok:true, exit 0, never noteRefusal.
coderMu.Lock()
recs := loadCoderRecords()
if _, busy := coderStarting[spec.RunID]; busy || coderHasRecord(recs, spec.RunID) || coderContainerExists(rt, spec.RunID) {
coderMu.Unlock()
return coderFail(res, "run_exists", "a run with this run_id is already on this machine")
}
all, mine := coderCountsLocked(spec.owner, recs)
total, avail, memOK := coderMemInfo()
availMB, budget := int(avail>>20), coderBudgetMB(int(total>>20))
committed := coderCommittedLocked(recs, cfg.MemPerRun)
pressure, psiOK := coderPSIMem()
// Memory: free memory now is not enough, because runs that started
// together all see the same free memory before any of them has grown. So
// this run's limit must also fit in the budget beside the limits already
// promised, and the machine must not be stalling on memory already.
memFull := memOK && (availMB < cfg.MemPerRun || committed+cfg.MemPerRun > budget)
pressured := psiOK && pressure >= coderPressureLimit
full := all >= cfg.Max || (grantMax != nil && mine >= *grantMax) || memFull || pressured
port := 0
if !full {
port = allocCoderPortLocked(recs)
full = port == 0
}
if full {
coderMu.Unlock()
res["ok"], res["started"], res["reason"] = true, false, "full"
res["max"], res["running"], res["running_mine"] = cfg.Max, all, mine
res["mem_avail_mb"], res["mem_budget_mb"], res["mem_pressure"] = nil, nil, nil
res["mem_committed_mb"] = committed
if memOK {
res["mem_avail_mb"], res["mem_budget_mb"] = availMB, budget
}
if psiOK {
res["mem_pressure"] = round2(pressure)
}
b, _ := json.Marshal(res)
return b, 0
}
coderStarting[spec.RunID] = coderReservation{owner: spec.owner, port: port, memMB: cfg.MemPerRun}
coderMu.Unlock()
defer func() {
coderMu.Lock()
delete(coderStarting, spec.RunID)
coderMu.Unlock()
}()
pullS, runImage, err := coderPullImage(rt, spec)
if err != nil {
return coderFail(res, "pull_failed", truncate(err.Error(), 220))
}
spec.runImage = runImage
if runImage != spec.Image {
res["local_image_id"] = runImage
}
res["pull_s"] = pullS
return coderStart(rt, spec, cfg, port, pullS, res)
}
// coderPullImage makes the run's image present and returns what to start the
// container from. It skips the pull when the digest is already here, or when
// the image was built or loaded on this machine and is here by its ID
// (coderLocalImage). Otherwise it pulls with a throwaway config dir that holds
// only this run's credential, so the pull never reads (or writes) the
// machine's own registry logins.
func coderPullImage(rt string, spec harnessSpec) (float64, string, error) {
if _, err := coderRun(30*time.Second, rt, "image", "inspect", spec.Image); err == nil {
return 0, spec.Image, nil
}
if id := coderLocalImage(rt, spec.Image); id != "" {
return 0, id, nil
}
pullS, err := coderPullImageFromRegistry(rt, spec)
return pullS, spec.Image, err
}
// coderLocalImage is the image ID to start a run from when its image
// (<repo>@sha256:<d>, already validated) was built or loaded on this machine,
// or "" when it is not here that way.
//
// WHY. Docker's classic image store records a repo digest only for an image
// pulled from or pushed to a registry. Measured on Docker 29.1.3 with the
// containerd snapshotter off: after `docker load` and after `docker build`,
// RepoDigests is empty, `inspect <repo>@sha256:<id>` fails, and `run` of that
// reference tries to pull. The containerd image store does record one (the
// same version with it on), so there the inspect in coderPullImage answers.
//
// THE RULE. The runtime's image whose ID is exactly sha256:<d>, and which
// carries a tag of <repo>. The ID is the digest of the image's config, which
// names every layer by the digest of its content, so it pins the content as a
// manifest digest does. The tag keeps the owner's prefix deciding what may
// run: the ID of some other image on this machine is never enough. The run
// is then started from the ID, which no tag moved after this check changes.
func coderLocalImage(rt, image string) string {
repo, digest, ok := strings.Cut(image, "@")
if !ok || !coderDigestRe.MatchString(digest) {
return ""
}
out, err := coderRun(30*time.Second, rt, "image", "inspect", "--format", "{{.Id}}|{{json .RepoTags}}", digest)
if err != nil {
return ""
}
id, tagsJSON, ok := strings.Cut(strings.TrimSpace(out), "|")
// podman prints the ID without its "sha256:".
if !ok || (id != digest && "sha256:"+id != digest) {
return ""
}
var tags []string
if json.Unmarshal([]byte(tagsJSON), &tags) != nil {
return ""
}
for _, t := range tags {
if coderTagRepo(t) == repo {
return digest
}
}
return ""
}
// coderTagRepo is the repository of "repo:tag": the tag is after the last ':'
// that follows the last '/', so a registry's port stays in the repository.
func coderTagRepo(t string) string {
slash := strings.LastIndex(t, "/")
if i := strings.LastIndex(t, ":"); i > slash {
return t[:i]
}
return t
}
// coderPullImageFromRegistry is the pull itself.
func coderPullImageFromRegistry(rt string, spec harnessSpec) (float64, error) {
d, err := codersDir()
if err != nil {
return 0, err
}
dir := filepath.Join(d, "pull-"+spec.RunID)
_ = os.RemoveAll(dir)
if err := os.Mkdir(dir, 0o700); err != nil {
return 0, err
}
defer os.RemoveAll(dir)
auths := map[string]any{}
if p := spec.Pull; p != nil {
auths[p.Registry] = map[string]string{"auth": base64.StdEncoding.EncodeToString([]byte(p.Username + ":" + p.Password))}
}
authJSON, _ := json.Marshal(map[string]any{"auths": auths})
start := coderNow()
var out string
switch rt {
case "podman":
f := filepath.Join(dir, "auth.json")
if err := os.WriteFile(f, authJSON, 0o600); err != nil {
return 0, err
}
out, err = coderRun(coderPullTimeout, rt, "pull", "--authfile", f, spec.Image)
case "nerdctl":
if err := os.WriteFile(filepath.Join(dir, "config.json"), authJSON, 0o600); err != nil {
return 0, err
}
out, err = coderRunEnv(coderPullTimeout, []string{"DOCKER_CONFIG=" + dir}, rt, "pull", spec.Image)
default:
if err := os.WriteFile(filepath.Join(dir, "config.json"), authJSON, 0o600); err != nil {
return 0, err
}
out, err = coderRun(coderPullTimeout, rt, "--config", dir, "pull", spec.Image)
}
if err != nil {
if p := spec.Pull; p != nil && p.Password != "" {
out = strings.ReplaceAll(out, p.Password, "<password>")
}
return 0, fmt.Errorf("%s", strings.TrimSpace(out))
}
return round2(coderNow().Sub(start).Seconds()), nil
}
// coderWriteSeccomp writes the embedded profile when it is missing or differs.
func coderWriteSeccomp() (string, error) {
d, err := codersDir()
if err != nil {
return "", err
}
p := filepath.Join(d, coderSeccompFile)
if b, err := os.ReadFile(p); err == nil && sha256.Sum256(b) == sha256.Sum256(coderSeccompProfile) {
_ = os.Chmod(p, 0o644)
return p, nil
}
tmp := p + ".tmp"
if err := os.WriteFile(tmp, coderSeccompProfile, 0o644); err != nil {
return "", err
}
if err := os.Chmod(tmp, 0o644); err != nil {
return "", err
}
return p, os.Rename(tmp, p)
}
// coderMkdirOwned makes one directory handed to the engine's uid, 0700.
func coderMkdirOwned(p string, uid int) error {
if err := os.MkdirAll(p, 0o700); err != nil {
return err
}
if err := os.Chmod(p, 0o700); err != nil {
return err
}
return coderChown(p, uid, uid)
}
// coderArgs is the container's command line, exactly as the contract states.
func coderArgs(spec harnessSpec, cfg coderConfig, R, seccomp, deadline string) []string {
mem := strconv.Itoa(cfg.MemPerRun) + "m"
args := []string{"run", "-d", "--name", coderContainerPrefix + spec.RunID,
"--label", coderLabelRun + "=" + spec.RunID,
"--label", coderLabelOwner + "=" + spec.owner,
"--label", coderLabelEngine + "=" + spec.Engine,
"--label", coderLabelDeadline + "=" + deadline,
"--network", "none", "--cap-drop", "ALL", "--security-opt", "no-new-privileges",
"--security-opt", "seccomp=" + seccomp,
"--read-only", "--user", strconv.Itoa(spec.UID) + ":" + strconv.Itoa(spec.UID),
"--memory", mem, "--memory-swap", mem,
"--cpus", strconv.FormatFloat(cfg.CPUsPerRun, 'f', -1, 64),
// CPU: the lowest share there is (cgroup v2 cpu.weight 1, measured
// on runc 1.3; the machine's own services sit at the default 100),
// as the kubelet gives a BestEffort pod. A run may use any CPU
// nothing else wants, up to --cpus, and yields it the moment the
// Host, sshd or anything else on the machine needs it.
"--cpu-shares", "2",
"--pids-limit", "4096",
"--tmpfs", "/tmp:rw,nosuid,nodev,exec,mode=1777,size=" + mem,
"-v", filepath.Join(R, "work") + ":" + coderWorkspaceIn,
"-v", filepath.Join(R, "home") + ":" + coderHomeIn,
"-v", filepath.Join(R, "ctl") + ":" + coderCtlDirIn,
"-v", filepath.Join(R, "egress") + ":" + coderEgressDirIn + ":ro",
"-e", "SKEMA_APP_ID=" + spec.Engine,
"-e", "PRIVATAE_RUN_ID=" + spec.RunID,
"-e", "HOME=" + coderHomeIn,
"-e", "SKEMA_WORKSPACE=" + coderWorkspaceIn,
"-e", "PRIVATAE_HARNESS_CAP=" + spec.ControlCap,
"-e", "PRIVATAE_HARNESS_CAP_ENFORCE=1",
"-e", "PRIVATAE_SIDECAR_CONTROL_SOCK=" + coderCtlDirIn + "/" + coderControlSock,
"-e", "PRIVATAE_SIDECAR_EGRESS_SOCK=" + coderEgressDirIn + "/" + coderEgressSock,
// GPU: none. A run gets no --gpus and no --device, and where NVIDIA's
// runtime is the default one, "void" makes it behave like runc and
// expose no GPU, even for an image built on a CUDA base, which
// sets NVIDIA_VISIBLE_DEVICES=all.
"-e", "NVIDIA_VISIBLE_DEVICES=void",
}
for _, n := range coderNvidiaSwarmEnv() {
args = append(args, "-e", n+"=void")
}
img := spec.Image
if spec.runImage != "" {
img = spec.runImage // a local image, by its ID (coderLocalImage)
}
return append(args, img)
}
// coderNvidiaSwarmEnv is every variable name the machine's NVIDIA toolkit
// config lists in swarm-resource. The toolkit reads those names in place of
// NVIDIA_VISIBLE_DEVICES whenever the container sets one of them, so an image
// that set one would get GPUs past NVIDIA_VISIBLE_DEVICES=void. The toolkit
// ships the setting commented out, with DOCKER_RESOURCE_GPU as its example
// value, for Docker Swarm. A run sets each listed name to "void" itself, and
// the run's value replaces the image's. No config, or no such setting, adds
// nothing.
func coderNvidiaSwarmEnv() []string {
b, err := os.ReadFile(coderNvidiaConfig)
if err != nil {
return nil
}
var names []string
for _, ln := range strings.Split(string(b), "\n") {
k, v, ok := strings.Cut(strings.TrimSpace(ln), "=")
if !ok || strings.TrimSpace(k) != "swarm-resource" {
continue // "#swarm-resource = ..." is the commented-out default
}
v, _, _ = strings.Cut(v, "#")
for _, n := range strings.Split(strings.Trim(strings.TrimSpace(v), `"'`), ",") {
if n = strings.TrimSpace(n); coderEnvNameRe.MatchString(n) && !containsStr(names, n) {
names = append(names, n)
}
}
}
return names
}
func coderStart(rt string, spec harnessSpec, cfg coderConfig, port int, pullS float64, res map[string]any) ([]byte, int) {
R, err := coderRunDir(spec.RunID)
if err != nil {
return coderFail(res, "start_failed", "no run directory: "+err.Error())
}
cname := coderContainerPrefix + spec.RunID
started := false
cleanup := func() {
if started {
_, _ = coderRun(30*time.Second, rt, "rm", "-f", cname)
}
coderCloseEgress(spec.RunID)
_ = os.RemoveAll(R)
}
fail := func(msg string) ([]byte, int) {
cleanup()
return coderFail(res, "start_failed", truncate(msg, 220))
}
_ = os.RemoveAll(R)
if err := os.MkdirAll(R, 0o700); err != nil {
return fail("run directory: " + err.Error())
}
for _, sub := range []string{"work", "home", "ctl"} {
if err := coderMkdirOwned(filepath.Join(R, sub), spec.UID); err != nil {
return fail("run directory: " + err.Error())
}
}
credDir := filepath.Join(R, "home", filepath.Base(spec.CredsDir))
if err := coderMkdirOwned(credDir, spec.UID); err != nil {
return fail("creds directory: " + err.Error())
}
names := make([]string, 0, len(spec.Creds))
for n := range spec.Creds {
names = append(names, n)
}
sort.Strings(names)
for _, n := range names {
f := filepath.Join(credDir, n)
if err := os.WriteFile(f, []byte(spec.Creds[n]), 0o600); err != nil {
return fail("creds file: " + err.Error())
}
if err := os.Chmod(f, 0o600); err != nil {
return fail("creds file: " + err.Error())
}
if err := coderChown(f, spec.UID, spec.UID); err != nil {
return fail("creds file: " + err.Error())
}
}
egDir := filepath.Join(R, "egress")
if err := os.MkdirAll(egDir, 0o755); err != nil {
return fail("egress directory: " + err.Error())
}
_ = os.Chmod(egDir, 0o755)
if err := coderListenEgress(spec.RunID, egDir, spec.UID); err != nil {
return fail("egress proxy: " + err.Error())
}
seccomp, err := coderWriteSeccomp()
if err != nil {
return fail("seccomp profile: " + err.Error())
}
maxMin := spec.MaxMinutes
if maxMin == 0 {
maxMin = coderDefaultMins
}
now := coderNow().UTC()
deadline := now.Add(time.Duration(maxMin)*time.Minute + coderDeadlineSlop).Format(time.RFC3339)
t0 := coderNow()
out, err := coderRun(60*time.Second, rt, coderArgs(spec, cfg, R, seccomp, deadline)...)
if err != nil {
started = coderContainerExists(rt, spec.RunID)
return fail("run failed: " + out)
}
started = true
if err := coderWaitUp(rt, cname, filepath.Join(R, "ctl", coderControlSock), spec.UID, coderStartWait); err != nil {
return fail(err.Error())
}
rec := &coderRecord{RunID: spec.RunID, Owner: spec.owner, Engine: spec.Engine, Image: spec.Image,
Container: cname, TunnelPort: port, UID: spec.UID, CreatedAt: now.Format(time.RFC3339),
Deadline: deadline, PullS: pullS, MemMB: cfg.MemPerRun, Cgroup: coderCgroupOf(rt, cname)}
coderMu.Lock()
err = saveCoderRecord(R, rec)
coderMu.Unlock()
if err != nil {
return fail("could not record the run, so it was removed: " + err.Error())
}
// A Host on this machine reaches the run through a socket in its runs
// directory (coder_socket.go), open before this answer goes back.
sock, err := coderOpenRunSocket(spec.owner, spec.RunID)
if err != nil {
return fail("could not open the run's control socket, so it was removed: " + err.Error())
}
if sock != "" {
res["control_socket"] = sock
}
res["ok"], res["started"] = true, true
res["container"], res["tunnel_port"] = cname, port
res["pull_s"], res["start_s"] = pullS, round2(coderNow().Sub(t0).Seconds())
res["deadline"], res["ran_on"] = deadline, orUnknown(hostnameSafe())
b, _ := json.Marshal(res)
return b, 0
}
// coderWaitHealthy waits for the sidecar to create its control socket, owned
// by the engine's uid, then asks /health over it. A container that stops
// running on the way is a failure at once, not after the full wait.
func coderWaitHealthy(rt, cname, sock string, uid int, within time.Duration) error {
deadline := time.Now().Add(within)
lastCheck := time.Time{}
for time.Now().Before(deadline) {
if owner, isSock := coderSocketOwner(sock); isSock && owner == uid {
if err := coderHealth(sock, uid); err == nil {
return nil
}
}
if time.Since(lastCheck) > 2*time.Second {
lastCheck = time.Now()
if coderContainerState(rt, cname) != "running" {
return errors.New("the container stopped before it answered /health")
}
}
time.Sleep(250 * time.Millisecond)
}
return errors.New("the container did not answer /health on its control socket within 60 s")
}
// coderDialControl connects to a run's control socket, pinned to the inode
// and only when the far end is the engine's uid (the spliceUnix walls).
func coderDialControl(sock string, uid int) (net.Conn, error) {
c, err := serviceDialSocket(sock, 5*time.Second)
if err != nil {
return nil, err
}
if got, err := servicePeerUID(c); err != nil || int(got) != uid {
_ = c.Close()
return nil, errors.New("the control socket is not the engine's")
}
return c, nil
}
func coderHealth(sock string, uid int) error {
cl := &http.Client{Timeout: 5 * time.Second, Transport: &http.Transport{
DialContext: func(context.Context, string, string) (net.Conn, error) { return coderDialControl(sock, uid) },
}}
resp, err := cl.Get("http://run/health")
if err != nil {
return err
}
resp.Body.Close()
if resp.StatusCode/100 != 2 {
return fmt.Errorf("/health answered %d", resp.StatusCode)
}
return nil
}
// coderContainerState is running | exited | missing.
func coderContainerState(rt, cname string) string {
out, err := coderRun(10*time.Second, rt, "inspect", "-f", "{{.State.Running}}", cname)
if err != nil {
return "missing"
}
if strings.TrimSpace(out) == "true" {
return "running"
}
return "exited"
}
func coderContainerExists(rt, runID string) bool {
_, err := coderRun(10*time.Second, rt, "inspect", "-f", "{{.Id}}", coderContainerPrefix+runID)
return err == nil
}
// coderContainerGone is true only when the runtime says the container does not
// exist. Any other failure is unknown, never absence (serviceContainerGone).
func coderContainerGone(rt, cname string) bool {
out, err := coderRun(10*time.Second, rt, "inspect", "-f", "{{.Id}}", cname)
if err == nil {
return false
}
o := strings.ToLower(out)
return strings.Contains(o, "no such") || strings.Contains(o, "not found")
}
// ── run-down ────────────────────────────────────────────────────────────────
// coderOwnerRefusal: only the pairing that started a run may stop it, or the
// anchor pairing on an anchored box (serviceOwnerRefusal's rule).
// An unknown owner proves nothing about self, so it gets the anchor-only rule
// too: every run this bastion starts records and labels its owner.
//
// A local machine has no anchor, so anchorOnly lets every pairing through
// there; once it serves several Hosts (localServesSeveralHosts), each stops
// only the runs it started, as with the servers each built.
func coderOwnerRefusal(self, owner string) string {
if owner != "" && owner == self {
return ""
}
if localServesSeveralHosts() {
return "refused: that run was started by " + orUnknown(owner) + "; on a local machine that serves several Hosts each stops only the runs it started"
}
if why := anchorOnly(self, "stopping a run another host started"); why != "" {
return "refused: that run was started by " + orUnknown(owner) + "; " + why
}
return ""
}
func coderRunDown(spec harnessSpec) ([]byte, int) {
res := coderBase(spec)
if !coderRunIDRe.MatchString(spec.RunID) {
return coderFail(res, "", "run_id is not a valid run id")
}
rt := coderRuntime()
rec, _ := loadCoderRecord(spec.RunID)
owner := ""
notFound := func() ([]byte, int) {
res["ok"], res["stopped"], res["reason"] = true, false, "not_found"
b, _ := json.Marshal(res)
return b, 0
}
if rec != nil {
owner = rec.Owner
} else {
cname := coderContainerPrefix + spec.RunID
if rt == "" || coderContainerGone(rt, cname) {
return notFound()
}
// No record: only the container's own labels say whose run it is. A
// container that merely has the name, without this run's label, is
// not a coder run and run-down never stops it.
run, err := coderLabelErr(rt, cname, coderLabelRun)
if err != nil {
return coderFail(res, "", "the runtime could not say whose container "+cname+" is; nothing was changed")
}
if run != spec.RunID {
return notFound()
}
owner = coderLabel(rt, cname, coderLabelOwner)
}
if why := coderOwnerRefusal(spec.owner, owner); why != "" {
return coderFail(res, "", why)
}
if rt == "" {
return coderFail(res, "", "no container runtime answers on this machine, so the run cannot be confirmed stopped; nothing was changed")
}
m, err := coderTeardown(rt, spec.RunID, rec)
if err != nil {
return coderFail(res, "", err.Error())
}
res["ok"], res["stopped"] = true, true
for k, v := range m {
res[k] = v
}
res["stopped_by"] = nil
if rec != nil && rec.StoppedBy != "" {
res["stopped_by"] = rec.StoppedBy
}
b, _ := json.Marshal(res)
return b, 0
}
func coderLabel(rt, cname, key string) string {
v, _ := coderLabelErr(rt, cname, key)
return v
}
// coderLabelErr is one label of a container; "" when it is not set, an error
// when the runtime did not answer.
func coderLabelErr(rt, cname, key string) (string, error) {
out, err := coderRun(10*time.Second, rt, "inspect", "-f", "{{index .Config.Labels \""+key+"\"}}", cname)
if err != nil {
return "", err
}
v := strings.TrimSpace(out)
if v == "<no value>" {
return "", nil
}
return v, nil
}
// coderTeardown is run-down without the reply: measure, stop, remove, confirm,
// then close the listeners and delete the run's files. The reaper calls it too.
// A container that cannot be confirmed gone keeps its record so the reaper
// retries.
func coderTeardown(rt, runID string, rec *coderRecord) (map[string]any, error) {
cname := coderContainerPrefix + runID
m := map[string]any{"peak_mem_bytes": nil, "cpu_s": nil, "oom_killed": nil, "exit_code": nil}
peak, cpu := coderMeasure(rt, cname)
if peak != nil {
m["peak_mem_bytes"] = *peak
}
if cpu != nil {
m["cpu_s"] = *cpu
}
// The start time lives only in the run record. Without it (run-down by
// the container's labels, or a record with no readable created_at) the
// wall time is not known: null, never 0.
m["wall_s"] = nil
if rec != nil {
if t, err := time.Parse(time.RFC3339, rec.CreatedAt); err == nil {
m["wall_s"] = round2(coderNow().Sub(t).Seconds())
}
}
_, _ = coderRun(30*time.Second, rt, "stop", "-t", "10", cname)
// One inspect per field: nerdctl's inspect has no State.OOMKilled and
// fails the whole template that names it, so a joint template would
// lose the exit code there too. On nerdctl oom_killed stays null.
if out, err := coderRun(10*time.Second, rt, "inspect", "-f", "{{.State.ExitCode}}", cname); err == nil {
if n, err := strconv.Atoi(strings.TrimSpace(out)); err == nil {
m["exit_code"] = n
}
}
if out, err := coderRun(10*time.Second, rt, "inspect", "-f", "{{.State.OOMKilled}}", cname); err == nil {
if b, err := strconv.ParseBool(strings.TrimSpace(out)); err == nil {
m["oom_killed"] = b
}
}
if out, err := coderRun(30*time.Second, rt, "rm", "-f", cname); err != nil || !coderContainerGone(rt, cname) {
if err == nil {
out = "the runtime still lists it"
}
return nil, fmt.Errorf("could not remove the run's container; it may still be there: %s", truncate(out, 200))
}
m["egress_refused"] = coderCloseEgress(runID)
coderMu.Lock()
if R, err := coderRunDir(runID); err == nil {
_ = os.RemoveAll(R)
}
coderMu.Unlock()
// After the record is gone: a reconcile that read it before then and
// opens the run's socket meanwhile has it closed here, and one that
// opens later finds no record (coderOpenRunSocket).
coderCloseRunSocket(runID)
return m, nil
}
// coderMeasure reads the container's cgroup from the host side before it is
// stopped: memory.peak (bytes) and cpu.stat usage_usec (as seconds). Either
// is nil when it cannot be read — never 0, never estimated.
func coderMeasure(rt, cname string) (*int64, *float64) {
dir := coderCgroupPath(coderCgroupOf(rt, cname))
if dir == "" {
return nil, nil
}
var peak *int64
if b, err := os.ReadFile(filepath.Join(dir, "memory.peak")); err == nil {
if n, err := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64); err == nil && n >= 0 {
peak = &n
}
}
var cpu *float64
if b, err := os.ReadFile(filepath.Join(dir, "cpu.stat")); err == nil {
for _, ln := range strings.Split(string(b), "\n") {
f := strings.Fields(ln)
if len(f) == 2 && f[0] == "usage_usec" {
if n, err := strconv.ParseInt(f[1], 10, 64); err == nil && n >= 0 {
v := math.Round(float64(n)/1e6*1000) / 1000
cpu = &v
}
}
}
}
return peak, cpu
}
// coderCgroupOf is the container's cgroup as /proc/<pid>/cgroup names it
// (cgroup v2), "" when the runtime gives no pid, there is no v2 line, or the
// line does not name a cgroup that exists under coderCgroupFS.
//
// That last check matters on cgroup v1 and hybrid machines. There the "0::"
// line names a path in a v2 tree mounted elsewhere (or "/"), so the path is
// absent under coderCgroupFS. Recorded anyway, the run would read as stopped
// (coderRecordRunning), its memory would not count against the budget, and
// the watchdog would never stop it. Recorded as "", the run counts at its
// limit and the watchdog stops it through the runtime. The root cgroup is
// refused as well: it holds every process on the machine.
func coderCgroupOf(rt, cname string) string {
out, err := coderRun(10*time.Second, rt, "inspect", "-f", "{{.State.Pid}}", cname)
if err != nil {
return ""
}
pid, err := strconv.Atoi(strings.TrimSpace(out))
if err != nil || pid <= 0 {
return ""
}
raw, err := os.ReadFile(filepath.Join(coderProcFS, strconv.Itoa(pid), "cgroup"))
if err != nil {
return ""
}
for _, ln := range strings.Split(string(raw), "\n") {
if !strings.HasPrefix(ln, "0::") {
continue
}
p := strings.TrimSpace(strings.TrimPrefix(ln, "0::"))
dir := coderCgroupPath(p)
if dir == "" || dir == filepath.Clean(coderCgroupFS) {
return ""
}
if _, err := os.Stat(filepath.Join(dir, "cgroup.events")); err != nil {
return ""
}
return p
}
return ""
}
// ── the watchdog ────────────────────────────────────────────────────────────
// coderGuard is what the watchdog remembers between rounds.
type coderGuard struct {
overSince time.Time // when memory pressure went over the limit; zero while under it
lastStop time.Time
lastRun string // the run stopped at lastStop
}
// exiting is true while the run stopped last still has processes and has had
// under coderPressureSettle to exit. Its memory is not free until they exit,
// and on a machine short of memory a killed process can take more than one
// round to finish exiting. Stopping the next run then would stop two where
// one was enough. The kubelet likewise waits for an evicted pod to be cleaned
// up before it evicts another. Bounded by the settle time, so a run that
// never empties cannot hold the watchdog back for long.
func (g *coderGuard) exiting(recs []coderRecord, now time.Time) bool {
if g.lastRun == "" || now.Sub(g.lastStop) >= coderPressureSettle {
return false
}
for _, r := range recs {
if r.RunID == g.lastRun {
up, known := coderRecordRunning(r)
return known && up
}
}
return false
}
// coderWatch runs a watchdog round every 2 s for as long as the daemon runs.
func coderWatch(ctx context.Context) {
g := &coderGuard{}
t := time.NewTicker(coderWatchEvery)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
g.round()
}
}
}
// round stops at most one run, the newest one still running, when this
// machine has under 100 MiB free, or when memory pressure has stayed over the
// limit for 30 s and the last stop has had 10 s to show. Run-up already keeps
// the runs' limits inside the budget, so this acts when something outside the
// runs grew into the machine's reserve. The newest run has done the least
// work, so stopping it loses the least.
//
// The run is killed through its cgroup (cgroup.kill), not through the
// container runtime, which can take minutes to answer on a machine that is
// short of memory. Its record stays, marked with why, so the Host's run-down
// reads what happened, and the reaper removes it like any stopped run.
func (g *coderGuard) round() (stopped, why string) {
recs := loadCoderRecords()
now := coderNow()
p, psiOK := coderPSIMem()
over := psiOK && p >= coderPressureLimit
if !over {
g.overSince = time.Time{}
} else if g.overSince.IsZero() {
g.overSince = now
}
if len(recs) == 0 {
return "", ""
}
_, avail, memOK := coderMemInfo()
availMB := int(avail >> 20)
switch {
case memOK && availMB < coderEvictFloorMB && g.exiting(recs, now):
return "", ""
case memOK && availMB < coderEvictFloorMB:
why = fmt.Sprintf("this machine had %d MiB of memory free, under the %d MiB it keeps", availMB, coderEvictFloorMB)
case over && now.Sub(g.overSince) >= coderPressureFor && now.Sub(g.lastStop) >= coderPressureSettle:
why = fmt.Sprintf("this machine stalled on memory %.0f%% of the time for %s (the limit is %.0f%%)",
p, coderPressureFor, coderPressureLimit)
default:
return "", ""
}
created := func(r coderRecord) time.Time {
t, _ := time.Parse(time.RFC3339, r.CreatedAt)
return t
}
sort.SliceStable(recs, func(i, j int) bool {
ti, tj := created(recs[i]), created(recs[j])
if !ti.Equal(tj) {
return ti.After(tj)
}
return recs[i].RunID > recs[j].RunID
})
rt := ""
for _, r := range recs {
if r.StoppedBy != "" {
continue
}
if up, known := coderRecordRunning(r); known && !up {
continue
}
if !coderKillCgroup(r.Cgroup) {
if rt == "" {
if rt = coderRuntime(); rt == "" {
return "", ""
}
}
if r.Cgroup == "" && coderContainerState(rt, r.Container) != "running" {
continue
}
if _, err := coderRun(30*time.Second, rt, "kill", r.Container); err != nil {
continue
}
}
g.lastStop, g.lastRun = now, r.RunID
coderMu.Lock()
if R, err := coderRunDir(r.RunID); err == nil {
if cur, err := loadCoderRecord(r.RunID); err == nil {
cur.StoppedBy = why
_ = coderRewriteRecord(R, cur)
}
}
coderMu.Unlock()
// A stopped run's control plane is gone with it: its socket, if a
// Host on this machine owns it, is removed now, not at run-down.
coderCloseRunSocket(r.RunID)
fmt.Printf("coders: stopped run %s to protect this machine: %s\n", r.RunID, why)
return r.RunID, why
}
return "", ""
}
// coderKillCgroup kills every process in a run's cgroup at once (cgroup.kill,
// Linux 5.14 and later). false when there is no cgroup to kill this way.
func coderKillCgroup(cg string) bool {
dir := coderCgroupPath(cg)
if dir == "" {
return false
}
// Never O_CREATE: on an older kernel the file is absent, not to be made.
f, err := os.OpenFile(filepath.Join(dir, "cgroup.kill"), os.O_WRONLY, 0)
if err != nil {
return false
}
_, err = f.WriteString("1")
if cerr := f.Close(); err == nil {
err = cerr
}
return err == nil
}
// ── run-status ──────────────────────────────────────────────────────────────
func coderIsAnchor(self string) bool {
a := currentAnchor()
return a != nil && a.AnchorHost == self
}
type coderRunReport struct {
RunID string `json:"run_id"`
Engine string `json:"engine"`
Container string `json:"container"`
TunnelPort int `json:"tunnel_port"`
State string `json:"state"`
CreatedAt string `json:"created_at"`
Deadline string `json:"deadline"`
StoppedBy string `json:"stopped_by,omitempty"`
}
func coderReports(self string, all bool) []coderRunReport {
recs := loadCoderRecords()
out := []coderRunReport{}
if len(recs) == 0 {
return out
}
rt := coderRuntime()
for _, r := range recs {
if !all && r.Owner != self {
continue
}
st := "missing"
if rt != "" {
st = coderContainerState(rt, r.Container)
}
out = append(out, coderRunReport{RunID: r.RunID, Engine: r.Engine, Container: r.Container,
TunnelPort: r.TunnelPort, State: st, CreatedAt: r.CreatedAt, Deadline: r.Deadline, StoppedBy: r.StoppedBy})
}
return out
}
func coderRunStatus(spec harnessSpec) ([]byte, int) {
res := coderBase(spec)
res["ok"] = true
res["runs"] = coderReports(spec.owner, coderIsAnchor(spec.owner))
b, _ := json.Marshal(res)
return b, 0
}
// ── heartbeat ───────────────────────────────────────────────────────────────
// coderHeartbeat is the `coders` capability block for pairing self, or nil
// when ANY of the conditions fails: Linux, root, coders on, a runtime, the
// pairing may be asked for mode:coder, and (anchored) it holds a valid coder
// grant. Never sent empty. The cheap checks run first, so a machine with
// coders off never runs its container runtime for a heartbeat.
func coderHeartbeat(self string) map[string]any {
if coderGOOS != "linux" || !coderIsRoot() {
return nil
}
cfg := loadCoderConfig()
if !cfg.ready() {
return nil
}
if capsAllow(self, coderGateOp) != "" {
return nil
}
var grantMax any
if currentAnchor() != nil {
g, why := coderGrantFor(coderDcnDirFor(self), self)
if why != "" {
return nil
}
grantMax = g.Max
}
rt := coderRuntime()
if rt == "" {
return nil
}
recs := loadCoderRecords()
coderMu.Lock()
all, mine := coderCountsLocked(self, recs)
coderMu.Unlock()
total, avail, ok := coderMemInfo()
var memTotal, memAvail, memBudget any
if ok {
memTotal, memAvail, memBudget = int(total>>20), int(avail>>20), coderBudgetMB(int(total>>20))
}
return map[string]any{
"max": cfg.Max, "running": all, "running_mine": mine, "grant_max": grantMax,
"mem_per_run_mb": cfg.MemPerRun, "cpus_per_run": cfg.CPUsPerRun,
"mem_total_mb": memTotal, "mem_avail_mb": memAvail, "mem_budget_mb": memBudget,
"cpus": runtime.NumCPU(), "arch": runtime.GOARCH, "runtime": rt,
"engines": append([]string(nil), cfg.Engines...),
}
}
// coderDcnDirFor is the grant directory of pairing self, "" when unknown.
func coderDcnDirFor(self string) string {
cfgs, err := loadPairings()
if err != nil {
return ""
}
if c := pairingFor(cfgs, self); c != nil && c.HostName == self {
return c.dcnDir()
}
return ""
}
// ── the box-wide loop and the reaper ────────────────────────────────────────
var coderInitOnce sync.Once
// coderInit starts the box-wide coder loop once per daemon: it re-opens the
// egress proxy for runs that outlived a restart, closes it for runs whose
// record is gone (a `coders drop` from the CLI), reaps once at start, and
// runs the watchdog (coderWatch). A machine that never used coder mode does
// nothing here: no files, no runtime.
func coderInit(ctx context.Context) {
coderInitOnce.Do(func() {
go func() {
go coderWatch(ctx)
coderReconcileEgress()
coderReap("", nil, false)
t := time.NewTicker(serviceForwardPoll)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
coderReconcileEgress()
}
}
}()
})
}
// coderMisses counts, per run, consecutive heartbeat replies from its owner
// that carried coder_runs and did not list it.
//
// coderStoppedSeen holds, per run, when the reaper first saw it not running
// without an exit time from the runtime: a record whose container is missing,
// or a runtime that keeps no exit time for it. It is held in memory, not in
// the run record: the reaper reads it on every heartbeat, and a daemon restart
// that loses it only restarts that run's 600 s grace, which delays a reap and
// never brings one forward. It also holds the first sighting of a run whose
// exit time was after now (the clock stepped back); coderExitAhead keeps that
// exit time, so the count stays on the first sighting once the clock passes
// it.
var (
coderReapMu sync.Mutex
coderMisses = map[string]int{}
coderStoppedSeen = map[string]time.Time{}
coderExitAhead = map[string]time.Time{}
)
// coderHeartbeatReply reads `coder_runs` from one pairing's heartbeat reply
// and runs the reaper. An absent key is no information: only the deadline and
// the other local rules apply.
func coderHeartbeatReply(self string, body []byte) {
var hb struct {
CoderRuns *[]string `json:"coder_runs"`
}
if json.Unmarshal(body, &hb) != nil || hb.CoderRuns == nil {
coderReap(self, nil, false)
return
}
coderReap(self, *hb.CoderRuns, true)
}
type coderSeen struct {
runID, owner, deadline string
running bool
created time.Time
finished time.Time // the runtime's exit time; zero when it gave none
}
// coderStoppedGrace is how long an exited or missing run is kept after it
// stopped, so the Host's run-down can still read its exit code and OOM flag.
const coderStoppedGrace = 600 * time.Second
// coderReap removes a coder container (run-down without a reply) when it is
// past its deadline, its owner pairing no longer exists, it has been exited or
// missing for more than 600 s, or it is older than 120 s and two consecutive
// replies from its owner that carried coder_runs left it out. The 600 s count
// from the runtime's exit time; when the runtime gives none (the container is
// missing, or the runtime keeps no exit time) or one after now (the clock
// stepped back), from when this daemon first saw the run not running; that
// exit time keeps counting from the first sighting after the clock passes
// it. Deadlines, owners and creation times are rebuilt from the containers'
// labels and the records on every round, so they survive a restart; the
// first-seen time does not (coderStoppedSeen).
func coderReap(replyFrom string, listed []string, hasKey bool) {
recs := loadCoderRecords()
if len(recs) == 0 && !loadCoderConfig().Enabled {
return
}
coderReapMu.Lock()
defer coderReapMu.Unlock()
rt := coderRuntime()
if rt == "" {
return
}
now := coderNow()
listedNow, ok := coderListContainers(rt)
if !ok {
// The runtime did not answer (a slow daemon under many runs): no
// information, never "everything exited". Next round retries.
return
}
seen := map[string]*coderSeen{}
for _, c := range listedNow {
c := c
seen[c.runID] = &c
}
byID := map[string]*coderRecord{}
for i := range recs {
r := &recs[i]
byID[r.RunID] = r
s, ok := seen[r.RunID]
if !ok {
s = &coderSeen{runID: r.RunID}
seen[r.RunID] = s
}
if s.owner == "" {
s.owner = r.Owner
}
// The earlier of the label's and the record's deadline applies.
if s.deadline == "" || coderBefore(r.Deadline, s.deadline) {
s.deadline = r.Deadline
}
if t, err := time.Parse(time.RFC3339, r.CreatedAt); err == nil && (s.created.IsZero() || t.Before(s.created)) {
s.created = t
}
}
cfgs, _ := loadPairings()
listedSet := map[string]bool{}
for _, id := range listed {
listedSet[id] = true
}
coderMu.Lock()
starting := map[string]bool{}
for id := range coderStarting {
starting[id] = true
}
coderMu.Unlock()
for id, s := range seen {
if starting[id] {
continue // a run-up in progress owns it
}
age := time.Duration(-1)
if !s.created.IsZero() {
age = now.Sub(s.created)
}
// How long the run has been exited or missing; -1 while it runs.
stopped := time.Duration(-1)
if s.running {
delete(coderStoppedSeen, id)
delete(coderExitAhead, id)
} else {
// An exit time after now (the clock stepped back) counts as
// none: the run has exited by now, so counting from when this
// daemon first saw it stopped never reaps it early. Read as a
// negative time stopped, it would hold the run for the skew;
// so would counting from it once the clock passes it, hence
// coderExitAhead remembers it.
since := s.finished
if since.After(now) {
coderExitAhead[id] = since
}
if ahead, ok := coderExitAhead[id]; ok && !ahead.Equal(since) {
delete(coderExitAhead, id) // a different exit: its own time
}
if _, ahead := coderExitAhead[id]; since.IsZero() || ahead {
first, ok := coderStoppedSeen[id]
if !ok {
first = now
coderStoppedSeen[id] = now
}
since = first
}
stopped = now.Sub(since)
}
if hasKey && s.owner == replyFrom {
if listedSet[id] {
delete(coderMisses, id)
} else {
coderMisses[id]++
}
}
why := ""
if dl, err := time.Parse(time.RFC3339, s.deadline); err == nil && now.After(dl) {
why = "past its deadline"
} else if s.owner != "" && cfgs != nil && pairingFor(cfgs, s.owner) == nil {
why = "its host is no longer paired"
} else if stopped > coderStoppedGrace {
why = "it stopped more than 600 s ago (exited or missing)"
} else if age > 120*time.Second && coderMisses[id] >= 2 {
why = "its host no longer holds it"
}
if why == "" {
continue
}
if _, err := coderTeardown(rt, id, byID[id]); err == nil {
delete(coderMisses, id)
delete(coderStoppedSeen, id)
delete(coderExitAhead, id)
fmt.Printf("coders: reaped run %s (%s)\n", id, why)
}
}
for id := range coderMisses {
if _, ok := seen[id]; !ok {
delete(coderMisses, id)
}
}
for id := range coderStoppedSeen {
if _, ok := seen[id]; !ok {
delete(coderStoppedSeen, id)
}
}
for id := range coderExitAhead {
if _, ok := seen[id]; !ok {
delete(coderExitAhead, id)
}
}
}
// coderBefore: a parses as RFC 3339 and is before b (or b does not parse).
func coderBefore(a, b string) bool {
ta, err := time.Parse(time.RFC3339, a)
if err != nil {
return false
}
tb, err := time.Parse(time.RFC3339, b)
return err != nil || ta.Before(tb)
}
// coderListContainers lists every container carrying the coder label, from
// the runtime itself (the records may be gone). ok is false when the runtime
// did not answer: that is unknown, never an empty list.
func coderListContainers(rt string) ([]coderSeen, bool) {
out, err := coderRun(15*time.Second, rt, "ps", "-aq", "--filter", "label="+coderLabelRun)
if err != nil {
return nil, false
}
ids := strings.Fields(out)
if len(ids) == 0 {
return nil, true
}
tmpl := `{{index .Config.Labels "` + coderLabelRun + `"}}|{{index .Config.Labels "` + coderLabelOwner +
`"}}|{{index .Config.Labels "` + coderLabelDeadline + `"}}|{{.State.Running}}|{{.Created}}|{{.State.FinishedAt}}`
got, err := coderRun(15*time.Second, rt, append([]string{"inspect", "-f", tmpl}, ids...)...)
if err != nil {
return nil, false
}
var res []coderSeen
for _, ln := range strings.Split(strings.TrimSpace(got), "\n") {
p := strings.SplitN(strings.TrimSpace(ln), "|", 6)
if len(p) != 6 || !coderRunIDRe.MatchString(p[0]) {
continue
}
s := coderSeen{runID: p[0], owner: p[1], deadline: p[2], running: p[3] == "true",
created: coderInspectTime(p[4]), finished: coderInspectTime(p[5])}
if s.owner == "<no value>" {
s.owner = ""
}
res = append(res, s)
}
return res, true
}
// coderInspectTime reads a time the runtime's inspect template printed:
// RFC 3339 from docker and nerdctl, Go's time.Time String form from podman.
// Zero when it is empty, does not parse, or is the runtime's "never" (docker
// and podman print year 1 for a container that has not exited, which is Go's
// zero time; nerdctl prints nothing). podman's form ends in the zone's
// abbreviation, and a zone without one prints its offset there instead
// ("+0330 +0330" in Tehran, "+0545 +0545" in Kathmandu), which the MST
// layout does not take; the offset before it already fixes the instant, so
// the form is read again without that last word.
func coderInspectTime(v string) time.Time {
v = strings.TrimSpace(v)
for _, layout := range []string{time.RFC3339Nano, "2006-01-02 15:04:05.999999999 -0700 MST"} {
if t, err := time.Parse(layout, v); err == nil {
return t
}
}
if i := strings.LastIndexByte(v, ' '); i > 0 {
if t, err := time.Parse("2006-01-02 15:04:05.999999999 -0700", v[:i]); err == nil {
return t
}
}
return time.Time{}
}
// runCmdEnv is runCmd with extra environment (nerdctl's DOCKER_CONFIG).
func runCmdEnv(timeout time.Duration, env []string, name string, args ...string) (string, error) {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
cmd := exec.CommandContext(ctx, name, args...)
hideConsole(cmd)
cmd.Env = append(os.Environ(), env...)
out, err := cmd.CombinedOutput()
return strings.TrimSpace(string(out)), err
}