From 2debb76ab68824e4f431d59892ffbf78370f40e7 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Thu, 22 Oct 2026 21:07:04 -0700 Subject: [PATCH 1/2] feat(runtime): opt-in per-job memory budget; stop leaking dind parent cgroups Per-job memory budget ([runtime] memory_limit_mb, memory_high_percent): sharks (40 GB, 5 slots) had host-wide OOMs on 6 of 14 days; on 2026-10-23 03:00Z one killed processes in three concurrent ephpm jobs at once. A job cgroup memory.max confines an OOM to the job that caused it. The budget is per JOB, not per container: the heavy work of a container: job runs in a dind sibling, so a per-container limit would allow N x the limit. Siblings already share a parent cgroup (/ephemerd-dind-, which containerd derives from the job's dind namespace); with a budget set the runner container moves under it too (/ephemerd-dind-/runner) and the parent carries memory.max plus memory.high (default 90%), so page cache is reclaimed and the job throttled before anything is killed. Nesting the siblings under the runner's own cgroup is not possible: cgroup v2 forbids a cgroup with processes from having controller-enabled children. Off by default (0 = no limit, runner keeps /ephemerd/). Fail-open: if the parent cannot be created or limited, the job runs unlimited as before. The parent's limits are rewritten before every container of the job, so they survive the sweep below removing an empty parent mid-job. Leaked cgroups: runc creates the job parent with MkdirAll and removes only container leaves, so every finished job left an empty /sys/fs/cgroup/ephemerd-dind- behind (477 on ducks, 541 on sharks). destroySteps now removes it after the task delete and Dind.Stop, and the node disk sweeper (startup and every tick) removes any that hold no processes and no child cgroups and have been idle for 10 minutes. New package pkg/jobcgroup owns the layout; a dind test pins that its prefix equals DindNamespacePrefix. --- cmd/ephemerd/main.go | 18 +++ pkg/config/config.go | 67 ++++++++++ pkg/config/config_test.go | 39 +++++- pkg/dind/containers.go | 4 + pkg/dind/dind.go | 12 +- pkg/dind/jobcgroup_prefix_test.go | 22 ++++ pkg/dind/pidslimit_linux.go | 36 ++++++ pkg/dind/pidslimit_linux_test.go | 56 ++++++++ pkg/dind/pidslimit_other.go | 12 +- pkg/jobcgroup/jobcgroup.go | 54 ++++++++ pkg/jobcgroup/jobcgroup_linux.go | 176 +++++++++++++++++++++++++ pkg/jobcgroup/jobcgroup_linux_test.go | 178 ++++++++++++++++++++++++++ pkg/jobcgroup/jobcgroup_other.go | 20 +++ pkg/runtime/rlimits_linux.go | 34 +++++ pkg/runtime/rlimits_linux_test.go | 35 +++++ pkg/runtime/rlimits_other.go | 9 ++ pkg/runtime/runtime.go | 17 +++ 17 files changed, 786 insertions(+), 3 deletions(-) create mode 100644 pkg/dind/jobcgroup_prefix_test.go create mode 100644 pkg/dind/pidslimit_linux_test.go create mode 100644 pkg/jobcgroup/jobcgroup.go create mode 100644 pkg/jobcgroup/jobcgroup_linux.go create mode 100644 pkg/jobcgroup/jobcgroup_linux_test.go create mode 100644 pkg/jobcgroup/jobcgroup_other.go diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index 288038f..5953e18 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -22,6 +22,7 @@ import ( "github.com/ephpm/ephemerd/pkg/dind" "github.com/ephpm/ephemerd/pkg/github" "github.com/ephpm/ephemerd/pkg/imagegc" + "github.com/ephpm/ephemerd/pkg/jobcgroup" "github.com/ephpm/ephemerd/pkg/metrics" "github.com/ephpm/ephemerd/pkg/networking" "github.com/ephpm/ephemerd/pkg/providers" @@ -361,6 +362,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP DindAllowPrivileged: cfg.Dind.ResolvedAllowPrivileged(), Rlimits: cfg.Runtime.Rlimits.Resolved(), PidsLimit: cfg.Runtime.ResolvedPidsLimit(), + MemoryLimit: memoryLimit(cfg.Runtime), AllowNewPrivileges: cfg.Runtime.ResolvedAllowNewPrivileges(), LinuxRuntime: cfg.Runner.Linux.ContainerdRuntime(), Network: net, @@ -685,6 +687,7 @@ func serve(ctx context.Context, configFile, imagesDirFlag string, containerdTCPP CacheProxyMounts: cacheProxyMounts, Rlimits: cfg.Runtime.Rlimits.Resolved(), PidsLimit: cfg.Runtime.ResolvedPidsLimit(), + MemoryLimit: memoryLimit(cfg.Runtime), AllowNewPrivileges: cfg.Runtime.ResolvedAllowNewPrivileges(), LinuxRuntime: cfg.Runner.Linux.ContainerdRuntime(), Network: net, @@ -1143,9 +1146,23 @@ func buildkitGCConfig(cfg *config.Config) buildkit.GCConfig { // // Errors from one pass are logged and the loop continues; the next tick // retries. +// memoryLimit resolves [runtime] memory_limit_mb / memory_high_percent into +// the job cgroup budget the runtime and dind apply. +func memoryLimit(r config.RuntimeConfig) jobcgroup.Limits { + maxBytes, highBytes := r.ResolvedMemoryLimit() + return jobcgroup.Limits{MaxBytes: maxBytes, HighBytes: highBytes} +} + +// jobCgroupSweepGrace is how long an empty job parent cgroup must sit +// unchanged before the periodic sweep removes it. Teardown removes a job's +// own parent; the sweep is the backstop for jobs that never reached teardown +// (daemon crash, host reboot) and for the backlog older daemons left. +const jobCgroupSweepGrace = 10 * time.Minute + func runNodeDiskSweeper(ctx context.Context, gc *imagegc.Collector, rt *runtime.Runtime, c *containerdclient.Client, interval time.Duration, repair brokenChainRepair, log *slog.Logger) { log = log.With("component", "node-disk-sweeper", "interval", interval) log.Info("starting node disk sweeper") + jobcgroup.Sweep(jobCgroupSweepGrace, log) ticker := time.NewTicker(interval) defer ticker.Stop() for { @@ -1169,6 +1186,7 @@ func runNodeDiskSweeper(ctx context.Context, gc *imagegc.Collector, rt *runtime. } } sweepDeadBuildRecords(passCtx, c, log) + jobcgroup.Sweep(jobCgroupSweepGrace, log) // Correctness repair, not capacity policy — so it runs on // every tick whether or not image GC is enabled, and before // the collector, which would otherwise plan against records diff --git a/pkg/config/config.go b/pkg/config/config.go index d2c0a73..ebb0410 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -362,6 +362,29 @@ type RuntimeConfig struct { // others and a busy neighbour cannot fail a build. PidsLimit int64 `toml:"pids_limit"` + // MemoryLimitMB is the memory budget for one JOB: the runner container + // and every sibling it starts through dind, together, share one cgroup + // with this memory.max. 0 (unset) means no limit, which is the + // historical behaviour. + // + // It is a per-job budget, not per-container, because the heavy work + // in a `container:` job runs inside a dind sibling: a per-container + // limit would let one job spend N × the limit. When the budget is hit + // the kernel OOM-kills inside that job only — usually the largest + // process (a linker), failing the step — instead of the host-wide OOM + // that on sharks (2026-10-23 03:00Z, 40 GB, 5 slots) killed processes + // in three concurrent jobs at once. + MemoryLimitMB int64 `toml:"memory_limit_mb"` + + // MemoryHighPercent sets memory.high as a percentage of MemoryLimitMB. + // Above memory.high the kernel reclaims the job's page cache and + // throttles it before memory.max is reached, so a build that only + // looks big because of cache (target/ dirs, registry layers) slows + // down rather than getting killed. 0 (unset) means + // DefaultMemoryHighPercent; valid range 50-100, where 100 disables the + // soft limit. Ignored when MemoryLimitMB is 0. + MemoryHighPercent int `toml:"memory_high_percent"` + // AllowNewPrivileges controls whether a process in the runner // container may gain privileges through execve — i.e. whether setuid // binaries and file capabilities take effect. When true the OCI spec @@ -502,6 +525,46 @@ func (r RuntimeConfig) ResolvedPidsLimit() int64 { } } +// DefaultMemoryHighPercent is memory.high as a share of [runtime] +// memory_limit_mb when memory_high_percent is unset. +const DefaultMemoryHighPercent = 90 + +// minMemoryLimitMB is the smallest per-job budget accepted. Anything below +// it cannot hold the runner itself, and is far more likely a units mistake +// (GB written as MB) than a deliberate setting. +const minMemoryLimitMB = 512 + +// ResolvedMemoryLimit returns the job cgroup's memory.max and memory.high in +// bytes. Both are 0 when no limit is configured; high is 0 (no soft limit) +// when memory_high_percent is 100. +func (r RuntimeConfig) ResolvedMemoryLimit() (maxBytes, highBytes int64) { + if r.MemoryLimitMB <= 0 { + return 0, 0 + } + maxBytes = r.MemoryLimitMB << 20 + pct := r.MemoryHighPercent + if pct == 0 { + pct = DefaultMemoryHighPercent + } + if pct >= 100 { + return maxBytes, 0 + } + return maxBytes, maxBytes * int64(pct) / 100 +} + +func (r RuntimeConfig) validate() error { + if r.MemoryLimitMB < 0 { + return fmt.Errorf("[runtime] memory_limit_mb = %d: must be 0 (no limit) or at least %d", r.MemoryLimitMB, minMemoryLimitMB) + } + if r.MemoryLimitMB > 0 && r.MemoryLimitMB < minMemoryLimitMB { + return fmt.Errorf("[runtime] memory_limit_mb = %d: below the %d MB floor (the value is in megabytes — 16 GiB is 16384)", r.MemoryLimitMB, minMemoryLimitMB) + } + if p := r.MemoryHighPercent; p != 0 && (p < 50 || p > 100) { + return fmt.Errorf("[runtime] memory_high_percent = %d: must be between 50 and 100", p) + } + return nil +} + // defaultNproc is RLIMIT_NPROC when [runtime.rlimits] nproc is unset. It is // deliberately effectively unlimited: NPROC is enforced per UID across the // host, not per container, so as a per-job limit it only ever made @@ -2038,6 +2101,10 @@ func (c *Config) validate() error { return err } + if err := c.Runtime.validate(); err != nil { + return err + } + if err := c.Runner.Linux.validate(); err != nil { return err } diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index a57b2f3..34528fa 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -1,4 +1,4 @@ -package config +package config import ( "bytes" @@ -2518,3 +2518,40 @@ func TestRuntimeConfig_ResolvedPidsLimit(t *testing.T) { } } } + +func TestRuntimeConfig_ResolvedMemoryLimit(t *testing.T) { + for _, tc := range []struct { + name string + in RuntimeConfig + wantMax, wantHigh int64 + }{ + {"unset: no limit", RuntimeConfig{}, 0, 0}, + {"default soft limit", RuntimeConfig{MemoryLimitMB: 16384}, 16 << 30, (16 << 30) * DefaultMemoryHighPercent / 100}, + {"explicit percent", RuntimeConfig{MemoryLimitMB: 1000, MemoryHighPercent: 50}, 1000 << 20, 500 << 20}, + {"100 percent: no soft limit", RuntimeConfig{MemoryLimitMB: 2048, MemoryHighPercent: 100}, 2048 << 20, 0}, + } { + gotMax, gotHigh := tc.in.ResolvedMemoryLimit() + if gotMax != tc.wantMax || gotHigh != tc.wantHigh { + t.Errorf("%s: ResolvedMemoryLimit() = (%d, %d), want (%d, %d)", tc.name, gotMax, gotHigh, tc.wantMax, tc.wantHigh) + } + } +} + +func TestRuntimeConfig_ValidateMemoryLimit(t *testing.T) { + for _, tc := range []struct { + in RuntimeConfig + wantErr bool + }{ + {RuntimeConfig{}, false}, + {RuntimeConfig{MemoryLimitMB: 16384}, false}, + {RuntimeConfig{MemoryLimitMB: 16384, MemoryHighPercent: 85}, false}, + {RuntimeConfig{MemoryLimitMB: 16}, true}, // GB typed as MB + {RuntimeConfig{MemoryLimitMB: -1}, true}, + {RuntimeConfig{MemoryLimitMB: 4096, MemoryHighPercent: 40}, true}, + {RuntimeConfig{MemoryLimitMB: 4096, MemoryHighPercent: 101}, true}, + } { + if err := tc.in.validate(); (err != nil) != tc.wantErr { + t.Errorf("validate(%+v) = %v, wantErr %v", tc.in, err, tc.wantErr) + } + } +} diff --git a/pkg/dind/containers.go b/pkg/dind/containers.go index d7c7dc9..5c8e2d3 100644 --- a/pkg/dind/containers.go +++ b/pkg/dind/containers.go @@ -526,6 +526,10 @@ func (s *Server) handleContainerCreate(w http.ResponseWriter, r *http.Request) { // host UID with every other job's containers, so RLIMIT_NPROC cannot // bound it per job. opts = append(opts, pidsLimitOpts(s.pidsLimit)...) + // The job's memory budget lives on the job parent cgroup this sibling is + // created under; (re)write it now so it holds even if the periodic sweep + // removed an empty parent between two of the job's containers. + opts = append(opts, jobMemoryOpts(s.jobID, s.memoryLimit, s.log)...) if req.HostConfig != nil { // Privileged mode: all capabilities, all devices, disable seccomp/apparmor, diff --git a/pkg/dind/dind.go b/pkg/dind/dind.go index 0399485..836c477 100644 --- a/pkg/dind/dind.go +++ b/pkg/dind/dind.go @@ -32,6 +32,7 @@ import ( "github.com/containerd/containerd/v2/core/images" "github.com/containerd/containerd/v2/pkg/namespaces" "github.com/ephpm/ephemerd/pkg/buildkit" + "github.com/ephpm/ephemerd/pkg/jobcgroup" "github.com/ephpm/ephemerd/pkg/networking" "github.com/ephpm/ephemerd/pkg/registrymirror" ) @@ -72,6 +73,7 @@ type Server struct { runnerNetNS string // path to runner container's net namespace; used to install DNAT rules for port bindings allowPrivileged bool // gate for docker run --privileged / --cap-add; see config.DindConfig.AllowPrivileged pidsLimit int64 // cgroup pids.max for siblings; 0 = no limit + memoryLimit jobcgroup.Limits // the job's memory budget, written on the job parent cgroup // stager publishes every job-supplied bind source at a path under // /dind-binds// that only root can reach, so the path @@ -191,6 +193,13 @@ type Config struct { // config.RuntimeConfig.PidsLimit. PidsLimit int64 + // MemoryLimit is the job's memory budget, shared by the runner and every + // sibling: siblings already live under the job parent cgroup (it is the + // cgroup containerd derives from the job's namespace), so applying the + // budget means writing it on that parent before each sibling starts. + // Zero means no limit. See pkg/jobcgroup. + MemoryLimit jobcgroup.Limits + // RegistryMirror routes this job's image pulls through a LAN // pull-through cache. Nil disables mirroring, leaving the pulls // identical to what they were before the feature existed. @@ -251,7 +260,8 @@ func New(cfg Config) (*Server, error) { runnerNetNS: cfg.RunnerNetNS, allowPrivileged: cfg.AllowPrivileged, pidsLimit: cfg.PidsLimit, - stager: newBindStager(cfg.DataDir, cfg.JobID, cfg.Log), + memoryLimit: cfg.MemoryLimit, + stager: newBindStager(cfg.DataDir, cfg.JobID, cfg.Log), mirror: cfg.RegistryMirror, resultsURL: cfg.ResultsURL, log: cfg.Log.With("component", "dind", "job_id", cfg.JobID), diff --git a/pkg/dind/jobcgroup_prefix_test.go b/pkg/dind/jobcgroup_prefix_test.go new file mode 100644 index 0000000..20608e9 --- /dev/null +++ b/pkg/dind/jobcgroup_prefix_test.go @@ -0,0 +1,22 @@ +package dind + +import ( + "testing" + + "github.com/ephpm/ephemerd/pkg/jobcgroup" +) + +// TestJobCgroupPrefixIsTheDindNamespacePrefix pins the coupling the job +// memory budget and the leaked-cgroup sweep both rely on: containerd puts a +// sibling in cgroup "//", so the job parent cgroup is named +// after the job's dind namespace. If either prefix changes alone, siblings +// escape the job budget and the sweep stops matching what jobs leave behind. +func TestJobCgroupPrefixIsTheDindNamespacePrefix(t *testing.T) { + if jobcgroup.Prefix != DindNamespacePrefix { + t.Fatalf("jobcgroup.Prefix = %q, DindNamespacePrefix = %q: they must match", jobcgroup.Prefix, DindNamespacePrefix) + } + s := &Server{jobNamespace: DindNamespacePrefix + "job-1"} + if "/"+s.jobNamespace != jobcgroup.Dir("job-1") { + t.Fatalf("job namespace %q does not name the job cgroup %q", s.jobNamespace, jobcgroup.Dir("job-1")) + } +} diff --git a/pkg/dind/pidslimit_linux.go b/pkg/dind/pidslimit_linux.go index e839ca0..00b2ce8 100644 --- a/pkg/dind/pidslimit_linux.go +++ b/pkg/dind/pidslimit_linux.go @@ -4,9 +4,11 @@ package dind import ( "context" + "log/slog" "github.com/containerd/containerd/v2/core/containers" "github.com/containerd/containerd/v2/pkg/oci" + "github.com/ephpm/ephemerd/pkg/jobcgroup" ocispec "github.com/opencontainers/runtime-spec/specs-go" ) @@ -31,3 +33,37 @@ func pidsLimitOpts(limit int64) []oci.SpecOpts { }, } } + +// jobMemoryOpts applies the job's memory budget to a sibling by placing it +// under the job parent cgroup and (re)writing the parent's limits first. The +// default spec already puts it there — containerd derives CgroupsPath from +// the job's dind namespace, which is the parent's name — but the path is set +// explicitly so the budget cannot silently stop applying if that derivation +// ever changes. No budget adds nothing. +// +// Fail-open, like the runner side: a parent that cannot be limited leaves the +// sibling on its default cgroup, unlimited, rather than failing the create. +func jobMemoryOpts(jobID string, limits jobcgroup.Limits, log *slog.Logger) []oci.SpecOpts { + if !limits.Enabled() || jobID == "" { + return nil + } + return []oci.SpecOpts{ + func(_ context.Context, _ oci.Client, c *containers.Container, s *oci.Spec) error { + if err := ensureJobCgroup(jobcgroup.Root, jobID, limits); err != nil { + log.Warn("job memory limit not applied to sibling; it runs unlimited", "error", err) + return nil + } + if s.Linux == nil { + s.Linux = &ocispec.Linux{} + } + if c != nil && c.ID != "" { + s.Linux.CgroupsPath = jobcgroup.Dir(jobID) + "/" + c.ID + } + return nil + }, + } +} + +// ensureJobCgroup is jobcgroup.Ensure, a var so tests can run the spec +// option without a real cgroupfs. +var ensureJobCgroup = jobcgroup.Ensure diff --git a/pkg/dind/pidslimit_linux_test.go b/pkg/dind/pidslimit_linux_test.go new file mode 100644 index 0000000..e85b505 --- /dev/null +++ b/pkg/dind/pidslimit_linux_test.go @@ -0,0 +1,56 @@ +//go:build linux + +package dind + +import ( + "context" + "errors" + "log/slog" + "testing" + + "github.com/containerd/containerd/v2/core/containers" + "github.com/containerd/containerd/v2/pkg/oci" + "github.com/ephpm/ephemerd/pkg/jobcgroup" + ocispec "github.com/opencontainers/runtime-spec/specs-go" +) + +// TestJobMemoryOpts: a sibling of a job with a memory budget is placed under +// the job parent cgroup, whose limits are (re)written first; with no budget, +// or when the parent cannot be prepared, the sibling keeps its default. +func TestJobMemoryOpts(t *testing.T) { + old := ensureJobCgroup + t.Cleanup(func() { ensureJobCgroup = old }) + limits := jobcgroup.Limits{MaxBytes: 4 << 30, HighBytes: 3 << 30} + + var ensured []string + ensureJobCgroup = func(_ string, jobID string, l jobcgroup.Limits) error { + if l != limits { + t.Errorf("Ensure got %+v, want %+v", l, limits) + } + ensured = append(ensured, jobID) + return nil + } + apply := func(opts []oci.SpecOpts) string { + spec := &oci.Spec{Linux: &ocispec.Linux{CgroupsPath: "/default/path"}} + for _, opt := range opts { + if err := opt(context.Background(), nil, &containers.Container{ID: "abc123"}, spec); err != nil { + t.Fatalf("opt: %v", err) + } + } + return spec.Linux.CgroupsPath + } + + if got, want := apply(jobMemoryOpts("job-1", limits, slog.Default())), jobcgroup.Dir("job-1")+"/abc123"; got != want { + t.Errorf("CgroupsPath = %q, want %q", got, want) + } + if len(ensured) != 1 || ensured[0] != "job-1" { + t.Errorf("Ensure calls = %v, want one for job-1", ensured) + } + if opts := jobMemoryOpts("job-1", jobcgroup.Limits{}, slog.Default()); len(opts) != 0 { + t.Errorf("no budget produced %d opts, want none", len(opts)) + } + ensureJobCgroup = func(string, string, jobcgroup.Limits) error { return errors.New("no memory controller") } + if got := apply(jobMemoryOpts("job-1", limits, slog.Default())); got != "/default/path" { + t.Errorf("CgroupsPath after a failed Ensure = %q, want the default kept (fail-open)", got) + } +} diff --git a/pkg/dind/pidslimit_other.go b/pkg/dind/pidslimit_other.go index 51b3239..f217dba 100644 --- a/pkg/dind/pidslimit_other.go +++ b/pkg/dind/pidslimit_other.go @@ -2,10 +2,20 @@ package dind -import "github.com/containerd/containerd/v2/pkg/oci" +import ( + "log/slog" + + "github.com/containerd/containerd/v2/pkg/oci" + "github.com/ephpm/ephemerd/pkg/jobcgroup" +) // pidsLimitOpts is a no-op off Linux: pids.max is a Linux cgroup control, and // dind siblings only run on Linux (see nonPrivilegedHardeningOpts). func pidsLimitOpts(_ int64) []oci.SpecOpts { return nil } + +// jobMemoryOpts is a no-op off Linux, for the same reason. +func jobMemoryOpts(_ string, _ jobcgroup.Limits, _ *slog.Logger) []oci.SpecOpts { + return nil +} diff --git a/pkg/jobcgroup/jobcgroup.go b/pkg/jobcgroup/jobcgroup.go new file mode 100644 index 0000000..cd8e9d3 --- /dev/null +++ b/pkg/jobcgroup/jobcgroup.go @@ -0,0 +1,54 @@ +// Package jobcgroup owns the per-job parent cgroup: the one cgroup that, on +// Linux, holds every container of one job. +// +// Layout (cgroup v2, cgroupfs driver — CgroupsPath "//"): +// +// /sys/fs/cgroup/ephemerd/ runner, no job limit +// /sys/fs/cgroup/ephemerd-dind-/ dind siblings +// /sys/fs/cgroup/ephemerd-dind-/runner runner, WITH a job limit +// +// The dind siblings land under the job parent on their own: containerd's +// default spec sets CgroupsPath to "//", and each job's dind +// namespace is "ephemerd-dind-". runc creates the parent with +// MkdirAll and only ever removes the leaf, which is why every finished job +// used to leave an empty ephemerd-dind-* directory behind (477 on ducks and +// 541 on sharks, 2026-10-23). Remove and SweepStale clean that up. +// +// With [runtime] memory_limit_mb set, the runner container is moved under the +// same parent (Dir(id)/runner) so the job's memory.max covers the runner and +// its siblings together — one budget per job. It cannot simply nest the +// siblings under the runner's own cgroup instead: cgroup v2's +// no-internal-processes rule forbids a cgroup that holds processes from +// having controller-enabled children. +package jobcgroup + +// Root is the cgroup v2 mount ephemerd manages job cgroups under. +const Root = "/sys/fs/cgroup" + +// Prefix is the name prefix of every job parent cgroup. It must equal +// dind.DindNamespacePrefix — the parent IS the cgroup containerd derives +// from the job's dind namespace — and dind's tests pin that. +const Prefix = "ephemerd-dind-" + +// RunnerLeaf is the runner container's cgroup name inside the job parent +// when a job memory limit is configured. Sibling container ids are 64-hex +// digests, so it cannot collide with one. +const RunnerLeaf = "runner" + +// Dir returns the job parent cgroup's path relative to the cgroupfs root, +// in the "/" form OCI CgroupsPath takes. +func Dir(jobID string) string { return "/" + Prefix + jobID } + +// RunnerPath is the runner container's CgroupsPath when the job has a +// memory budget. +func RunnerPath(jobID string) string { return Dir(jobID) + "/" + RunnerLeaf } + +// Limits is the job's memory budget in bytes. Zero values mean no limit. +type Limits struct { + MaxBytes int64 // memory.max + HighBytes int64 // memory.high; 0 = no soft limit +} + +// Enabled reports whether the job has a memory budget at all. Without one +// nothing is created and the runner keeps its historical cgroup. +func (l Limits) Enabled() bool { return l.MaxBytes > 0 } diff --git a/pkg/jobcgroup/jobcgroup_linux.go b/pkg/jobcgroup/jobcgroup_linux.go new file mode 100644 index 0000000..58f6c85 --- /dev/null +++ b/pkg/jobcgroup/jobcgroup_linux.go @@ -0,0 +1,176 @@ +//go:build linux + +package jobcgroup + +import ( + "errors" + "fmt" + "io/fs" + "log/slog" + "os" + "path/filepath" + "strconv" + "strings" + "time" +) + +// removeDir removes one cgroup directory. On cgroupfs rmdir succeeds on a +// directory that still lists its interface files (they are virtual) and +// fails with EBUSY while it holds processes or child cgroups. Tests swap it +// for os.RemoveAll, since a temp dir standing in for cgroupfs has real files. +var removeDir = os.Remove + +// Ensure creates the job parent cgroup under root if it is missing and writes +// its memory budget. Idempotent, and called before every container of the job +// is created, so a parent that SweepStale removed between two of the job's +// containers comes back with its limits rather than without them. +func Ensure(root, jobID string, l Limits) error { + if !l.Enabled() { + return nil + } + if err := enableMemory(root); err != nil { + return err + } + dir := filepath.Join(root, Dir(jobID)) + if err := os.MkdirAll(dir, 0o755); err != nil { + return fmt.Errorf("creating job cgroup %s: %w", dir, err) + } + // max before high: high above max is allowed, a max below the current + // usage is not refused either (the kernel reclaims), so this order never + // fails on a resize in either direction. + if err := writeFile(dir, "memory.max", strconv.FormatInt(l.MaxBytes, 10)); err != nil { + return err + } + high := "max" + if l.HighBytes > 0 { + high = strconv.FormatInt(l.HighBytes, 10) + } + return writeFile(dir, "memory.high", high) +} + +// enableMemory makes sure the memory controller is delegated to root's +// children, without which the job parent has no memory.max to write. On a +// systemd host it already is; writing it again is a no-op. +func enableMemory(root string) error { + ctl := filepath.Join(root, "cgroup.subtree_control") + cur, err := os.ReadFile(ctl) + if err != nil { + return fmt.Errorf("reading %s: %w", ctl, err) + } + for _, c := range strings.Fields(string(cur)) { + if c == "memory" { + return nil + } + } + if err := os.WriteFile(ctl, []byte("+memory"), 0o644); err != nil { + return fmt.Errorf("enabling the memory controller in %s: %w", ctl, err) + } + return nil +} + +func writeFile(dir, name, value string) error { + p := filepath.Join(dir, name) + if err := os.WriteFile(p, []byte(value), 0o644); err != nil { + return fmt.Errorf("writing %s=%s: %w", p, value, err) + } + return nil +} + +// Remove deletes the job parent cgroup once its containers are gone: any +// leftover empty child first, then the parent. Missing is success. A cgroup +// that still holds processes is left alone (rmdir refuses it) and reported; +// SweepStale retries it later. +func Remove(root, jobID string) error { + dir := filepath.Join(root, Dir(jobID)) + var lastErr error + for attempt := 0; attempt < 3; attempt++ { + if attempt > 0 { + // The kernel frees a cgroup a beat after its last task exits; + // runc's own delete of a leaf can still be in flight. + time.Sleep(time.Duration(attempt) * 200 * time.Millisecond) + } + entries, err := os.ReadDir(dir) + if errors.Is(err, fs.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("reading job cgroup %s: %w", dir, err) + } + for _, e := range entries { + if e.IsDir() { + _ = removeDir(filepath.Join(dir, e.Name())) + } + } + if lastErr = removeDir(dir); lastErr == nil || errors.Is(lastErr, fs.ErrNotExist) { + return nil + } + } + return fmt.Errorf("removing job cgroup %s: %w", dir, lastErr) +} + +// SweepStale removes job parent cgroups under root that hold no processes and +// no child cgroups and have not changed for grace. It never removes one with +// either: a live job's parent always holds its containers' leaves, and with a +// job memory limit the runner's leaf is in it for the job's whole life. +// Without a limit, a live job's parent can be briefly empty between two +// sibling containers; removing it then is harmless because runc recreates it, +// and the grace period makes it unlikely in the first place. +// +// Returns how many it removed. +func SweepStale(root string, grace time.Duration, now time.Time) (int, error) { + entries, err := os.ReadDir(root) + if err != nil { + return 0, fmt.Errorf("reading %s: %w", root, err) + } + removed := 0 + for _, e := range entries { + if !e.IsDir() || !strings.HasPrefix(e.Name(), Prefix) { + continue + } + dir := filepath.Join(root, e.Name()) + if !idle(dir) { + continue + } + info, err := e.Info() + if err != nil || now.Sub(info.ModTime()) < grace { + continue + } + if removeDir(dir) == nil { + removed++ + } + } + return removed, nil +} + +// idle reports whether a cgroup has no processes and no child cgroups. +// Anything unreadable counts as busy: the sweep only removes what it can +// prove is empty. +func idle(dir string) bool { + entries, err := os.ReadDir(dir) + if err != nil { + return false + } + for _, e := range entries { + if e.IsDir() { + return false + } + } + procs, err := os.ReadFile(filepath.Join(dir, "cgroup.procs")) + if errors.Is(err, fs.ErrNotExist) { + return true + } + return err == nil && strings.TrimSpace(string(procs)) == "" +} + +// Sweep is SweepStale on the real cgroupfs, logged. A sweep failure is never +// fatal — the leak it cleans up is slow and harmless in the short term. +func Sweep(grace time.Duration, log *slog.Logger) { + n, err := SweepStale(Root, grace, time.Now()) + if err != nil { + log.Debug("job cgroup sweep failed", "error", err) + return + } + if n > 0 { + log.Info("removed stale job cgroups", "count", n) + } +} diff --git a/pkg/jobcgroup/jobcgroup_linux_test.go b/pkg/jobcgroup/jobcgroup_linux_test.go new file mode 100644 index 0000000..8dd9997 --- /dev/null +++ b/pkg/jobcgroup/jobcgroup_linux_test.go @@ -0,0 +1,178 @@ +//go:build linux + +package jobcgroup + +import ( + "errors" + "os" + "path/filepath" + "strings" + "syscall" + "testing" + "time" +) + +// fakeCgroupfs stands in for /sys/fs/cgroup: a temp dir whose root carries a +// cgroup.subtree_control, and a removeDir that behaves like rmdir on +// cgroupfs — refusing a cgroup with processes or child cgroups, removing one +// that only holds interface files. +func fakeCgroupfs(t *testing.T, subtree string) string { + t.Helper() + root := t.TempDir() + write(t, filepath.Join(root, "cgroup.subtree_control"), subtree) + old := removeDir + removeDir = func(dir string) error { + if !idle(dir) { + return &os.PathError{Op: "rmdir", Path: dir, Err: syscall.EBUSY} + } + return os.RemoveAll(dir) + } + t.Cleanup(func() { removeDir = old }) + return root +} + +func write(t *testing.T, path, data string) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte(data), 0o644); err != nil { + t.Fatal(err) + } +} + +func read(t *testing.T, path string) string { + t.Helper() + b, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return string(b) +} + +func TestEnsure_WritesTheJobBudget(t *testing.T) { + root := fakeCgroupfs(t, "cpu memory pids") + if err := Ensure(root, "job-1", Limits{MaxBytes: 16 << 30, HighBytes: 14 << 30}); err != nil { + t.Fatalf("Ensure: %v", err) + } + dir := filepath.Join(root, Dir("job-1")) + if got := read(t, filepath.Join(dir, "memory.max")); got != "17179869184" { + t.Errorf("memory.max = %q, want 16 GiB", got) + } + if got := read(t, filepath.Join(dir, "memory.high")); got != "15032385536" { + t.Errorf("memory.high = %q, want 14 GiB", got) + } + // Idempotent, and a later resize rewrites the values. + if err := Ensure(root, "job-1", Limits{MaxBytes: 8 << 30}); err != nil { + t.Fatalf("second Ensure: %v", err) + } + if got := read(t, filepath.Join(dir, "memory.high")); got != "max" { + t.Errorf("memory.high with no soft limit = %q, want max", got) + } +} + +func TestEnsure_NoBudgetCreatesNothing(t *testing.T) { + root := fakeCgroupfs(t, "memory") + if err := Ensure(root, "job-1", Limits{}); err != nil { + t.Fatalf("Ensure: %v", err) + } + if _, err := os.Stat(filepath.Join(root, Dir("job-1"))); !errors.Is(err, os.ErrNotExist) { + t.Errorf("a job with no memory budget got a parent cgroup (stat err %v)", err) + } +} + +func TestEnsure_EnablesTheMemoryController(t *testing.T) { + root := fakeCgroupfs(t, "cpu pids") + if err := Ensure(root, "job-1", Limits{MaxBytes: 1 << 30}); err != nil { + t.Fatalf("Ensure: %v", err) + } + if got := read(t, filepath.Join(root, "cgroup.subtree_control")); !strings.Contains(got, "+memory") { + t.Errorf("subtree_control = %q, want the memory controller enabled", got) + } +} + +func TestRemove_DropsTheParentAndEmptyLeaves(t *testing.T) { + root := fakeCgroupfs(t, "memory") + dir := filepath.Join(root, Dir("job-1")) + write(t, filepath.Join(dir, "memory.max"), "max") + write(t, filepath.Join(dir, "deadbeef", "cgroup.procs"), "") + if err := Remove(root, "job-1"); err != nil { + t.Fatalf("Remove: %v", err) + } + if _, err := os.Stat(dir); !errors.Is(err, os.ErrNotExist) { + t.Errorf("job cgroup still present after Remove (stat err %v)", err) + } + // Removing what is already gone is success. + if err := Remove(root, "job-1"); err != nil { + t.Errorf("Remove of a missing cgroup: %v", err) + } +} + +func TestRemove_LeavesABusyCgroup(t *testing.T) { + root := fakeCgroupfs(t, "memory") + dir := filepath.Join(root, Dir("job-1")) + write(t, filepath.Join(dir, RunnerLeaf, "cgroup.procs"), "4242\n") + if err := Remove(root, "job-1"); err == nil { + t.Fatal("Remove succeeded on a job whose runner still has processes") + } + if _, err := os.Stat(filepath.Join(dir, RunnerLeaf)); err != nil { + t.Errorf("busy leaf removed: %v", err) + } +} + +func TestSweepStale(t *testing.T) { + root := fakeCgroupfs(t, "memory") + old := time.Now().Add(-time.Hour) + mk := func(name string, files map[string]string) string { + dir := filepath.Join(root, name) + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + for f, v := range files { + write(t, filepath.Join(dir, f), v) + } + if err := os.Chtimes(dir, old, old); err != nil { + t.Fatal(err) + } + return dir + } + leaked := mk(Prefix+"done-1", map[string]string{"cgroup.procs": "", "memory.max": "max"}) + leaked2 := mk(Prefix+"done-2", nil) + busy := mk(Prefix+"running", map[string]string{"cgroup.procs": "77\n"}) + parent := mk(Prefix+"with-child", map[string]string{"cid/cgroup.procs": ""}) + if err := os.Chtimes(parent, old, old); err != nil { // the child write bumped it + t.Fatal(err) + } + other := mk("system.slice", nil) + recent := filepath.Join(root, Prefix+"just-finished") + if err := os.MkdirAll(recent, 0o755); err != nil { + t.Fatal(err) + } + + n, err := SweepStale(root, 10*time.Minute, time.Now()) + if err != nil { + t.Fatalf("SweepStale: %v", err) + } + if n != 2 { + t.Errorf("removed %d, want 2", n) + } + for _, gone := range []string{leaked, leaked2} { + if _, err := os.Stat(gone); !errors.Is(err, os.ErrNotExist) { + t.Errorf("%s: idle stale job cgroup not removed", filepath.Base(gone)) + } + } + for _, kept := range []string{busy, parent, other, recent} { + if _, err := os.Stat(kept); err != nil { + t.Errorf("%s: must be kept (procs, children, not ours, or inside the grace period): %v", filepath.Base(kept), err) + } + } +} + +func TestPaths(t *testing.T) { + if got := Dir("ephemerd-github-ephpm-x"); got != "/ephemerd-dind-ephemerd-github-ephpm-x" { + t.Errorf("Dir = %q", got) + } + if got := RunnerPath("j"); got != "/ephemerd-dind-j/runner" { + t.Errorf("RunnerPath = %q", got) + } +} diff --git a/pkg/jobcgroup/jobcgroup_other.go b/pkg/jobcgroup/jobcgroup_other.go new file mode 100644 index 0000000..1152d35 --- /dev/null +++ b/pkg/jobcgroup/jobcgroup_other.go @@ -0,0 +1,20 @@ +//go:build !linux + +package jobcgroup + +import ( + "log/slog" + "time" +) + +// Ensure is a no-op off Linux: job cgroups are a Linux cgroup v2 construct. +func Ensure(_, _ string, _ Limits) error { return nil } + +// Remove is a no-op off Linux. +func Remove(_, _ string) error { return nil } + +// SweepStale is a no-op off Linux. +func SweepStale(_ string, _ time.Duration, _ time.Time) (int, error) { return 0, nil } + +// Sweep is a no-op off Linux. +func Sweep(_ time.Duration, _ *slog.Logger) {} diff --git a/pkg/runtime/rlimits_linux.go b/pkg/runtime/rlimits_linux.go index d919fd3..91ddc2f 100644 --- a/pkg/runtime/rlimits_linux.go +++ b/pkg/runtime/rlimits_linux.go @@ -4,13 +4,47 @@ package runtime import ( "context" + "log/slog" "github.com/containerd/containerd/v2/core/containers" "github.com/containerd/containerd/v2/pkg/oci" "github.com/ephpm/ephemerd/pkg/config" + "github.com/ephpm/ephemerd/pkg/jobcgroup" ocispec "github.com/opencontainers/runtime-spec/specs-go" ) +// jobCgroupOpts places the runner container under its job's parent cgroup +// when the job has a memory budget, so the budget covers the runner and its +// dind siblings together (see pkg/jobcgroup). With no budget it adds nothing +// and the runner keeps its historical /ephemerd/ cgroup. +// +// Fail-open: if the parent cannot be created or limited, the runner keeps +// its default cgroup and the job runs unlimited, as it did before the +// setting existed. A protection that cannot be applied should not also +// cost the build. +func jobCgroupOpts(id string, limits jobcgroup.Limits, log *slog.Logger) []oci.SpecOpts { + if !limits.Enabled() { + return nil + } + return []oci.SpecOpts{ + func(_ context.Context, _ oci.Client, _ *containers.Container, s *oci.Spec) error { + if err := ensureJobCgroup(jobcgroup.Root, id, limits); err != nil { + log.Warn("job memory limit not applied; the job runs unlimited", "id", id, "error", err) + return nil + } + if s.Linux == nil { + s.Linux = &ocispec.Linux{} + } + s.Linux.CgroupsPath = jobcgroup.RunnerPath(id) + return nil + }, + } +} + +// ensureJobCgroup is jobcgroup.Ensure, a var so tests can run the spec +// option without a real cgroupfs. +var ensureJobCgroup = jobcgroup.Ensure + // pidsLimitOpts sets the container's cgroup pids.max. limit <= 0 adds // nothing (no limit). dind applies the same limit to sibling containers // (dind.pidsLimitOpt) — see config.RuntimeConfig.PidsLimit. diff --git a/pkg/runtime/rlimits_linux_test.go b/pkg/runtime/rlimits_linux_test.go index 3be21fc..71a935d 100644 --- a/pkg/runtime/rlimits_linux_test.go +++ b/pkg/runtime/rlimits_linux_test.go @@ -4,10 +4,13 @@ package runtime import ( "context" + "errors" + "log/slog" "testing" "github.com/containerd/containerd/v2/pkg/oci" "github.com/ephpm/ephemerd/pkg/config" + "github.com/ephpm/ephemerd/pkg/jobcgroup" ocispec "github.com/opencontainers/runtime-spec/specs-go" ) @@ -124,3 +127,35 @@ func TestPidsLimitOpts(t *testing.T) { t.Errorf("pidsLimitOpts(0) returned %d opts, want none", len(opts)) } } + +// TestJobCgroupOpts: with a job memory budget the runner moves under the +// job parent cgroup (so one memory.max covers it and its dind siblings); +// without one, or when the parent cannot be prepared, it keeps its default +// cgroup. +func TestJobCgroupOpts(t *testing.T) { + old := ensureJobCgroup + t.Cleanup(func() { ensureJobCgroup = old }) + limits := jobcgroup.Limits{MaxBytes: 1 << 30} + + apply := func(opts []oci.SpecOpts) *oci.Spec { + spec := &oci.Spec{Linux: &ocispec.Linux{CgroupsPath: "/ephemerd/job-1"}} + for _, opt := range opts { + if err := opt(context.Background(), nil, nil, spec); err != nil { + t.Fatalf("opt: %v", err) + } + } + return spec + } + + ensureJobCgroup = func(string, string, jobcgroup.Limits) error { return nil } + if got := apply(jobCgroupOpts("job-1", limits, slog.Default())).Linux.CgroupsPath; got != jobcgroup.RunnerPath("job-1") { + t.Errorf("CgroupsPath = %q, want %q", got, jobcgroup.RunnerPath("job-1")) + } + if opts := jobCgroupOpts("job-1", jobcgroup.Limits{}, slog.Default()); len(opts) != 0 { + t.Errorf("no budget produced %d opts, want none", len(opts)) + } + ensureJobCgroup = func(string, string, jobcgroup.Limits) error { return errors.New("read-only cgroupfs") } + if got := apply(jobCgroupOpts("job-1", limits, slog.Default())).Linux.CgroupsPath; got != "/ephemerd/job-1" { + t.Errorf("CgroupsPath after a failed Ensure = %q, want the default kept (fail-open)", got) + } +} diff --git a/pkg/runtime/rlimits_other.go b/pkg/runtime/rlimits_other.go index 7e8b4de..b48f6a6 100644 --- a/pkg/runtime/rlimits_other.go +++ b/pkg/runtime/rlimits_other.go @@ -3,8 +3,11 @@ package runtime import ( + "log/slog" + "github.com/containerd/containerd/v2/pkg/oci" "github.com/ephpm/ephemerd/pkg/config" + "github.com/ephpm/ephemerd/pkg/jobcgroup" ) // rlimitsOpts is a no-op on non-Linux platforms. Windows Hyper-V isolated @@ -18,3 +21,9 @@ func rlimitsOpts(_ config.RuntimeRlimits) []oci.SpecOpts { func pidsLimitOpts(_ int64) []oci.SpecOpts { return nil } + +// jobCgroupOpts is a no-op off Linux: the job memory budget is a Linux +// cgroup v2 control (Windows containers have their own memory limit). +func jobCgroupOpts(_ string, _ jobcgroup.Limits, _ *slog.Logger) []oci.SpecOpts { + return nil +} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index 205f6b4..170cf7e 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -26,6 +26,7 @@ import ( "github.com/ephpm/ephemerd/pkg/config" "github.com/ephpm/ephemerd/pkg/dind" "github.com/ephpm/ephemerd/pkg/imagegc" + "github.com/ephpm/ephemerd/pkg/jobcgroup" "github.com/ephpm/ephemerd/pkg/networking" "github.com/ephpm/ephemerd/pkg/proxies" "github.com/ephpm/ephemerd/pkg/registrymirror" @@ -117,6 +118,11 @@ type Config struct { // dind sibling of its job; 0 means no limit. A resolved value: callers // pass cfg.Runtime.ResolvedPidsLimit(). Linux only. PidsLimit int64 + // MemoryLimit is the per-job memory budget shared by the runner + // container and every dind sibling of its job (see pkg/jobcgroup). + // Zero means no limit and leaves the runner in its historical cgroup. + // Linux only. + MemoryLimit jobcgroup.Limits // AllowNewPrivileges permits privilege escalation via execve in the // runner container (NoNewPrivileges=false), which is what makes // `sudo` work. See config.RuntimeConfig.AllowNewPrivileges. @@ -1179,6 +1185,7 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err opts = append(opts, newPrivilegesOpts(r.cfg.AllowNewPrivileges)...) opts = append(opts, rlimitsOpts(r.cfg.Rlimits)...) opts = append(opts, pidsLimitOpts(r.cfg.PidsLimit)...) + opts = append(opts, jobCgroupOpts(id, r.cfg.MemoryLimit, r.cfg.Log)...) switch { case len(cfg.Entrypoint) > 0: // Forge mode: custom entrypoint (e.g. act_runner register + daemon). @@ -1294,6 +1301,7 @@ func (r *Runtime) Create(ctx context.Context, cfg CreateConfig) (*RunnerEnv, err BuildKit: r.cfg.BuildKit, AllowPrivileged: r.cfg.DindAllowPrivileged, PidsLimit: r.cfg.PidsLimit, + MemoryLimit: r.cfg.MemoryLimit, RegistryMirror: r.cfg.RegistryMirror, ResultsURL: resultsURL, Transport: resolveDindTransport(r.cfg.LinuxRuntime, goruntime.GOOS), @@ -1792,6 +1800,15 @@ func (r *Runtime) destroySteps(ctx context.Context, env *RunnerEnv) { r.cfg.Log.Warn("failed to delete container", "id", env.ID, "error", err) } + // The job parent cgroup (the dind siblings' parent, and the runner's too + // under a job memory limit). runc removes only container leaves, so + // without this every job left an empty directory behind. The task delete + // and Dind.Stop above have already removed the leaves; anything still + // busy is retried by the periodic sweep. + if err := jobcgroup.Remove(jobcgroup.Root, env.ID); err != nil { + r.cfg.Log.Debug("job cgroup not removed yet; the sweep will retry", "id", env.ID, "error", err) + } + // Clean up per-job runner directory copy. if env.RunnerDir != "" { if err := os.RemoveAll(env.RunnerDir); err != nil { From bf7afeeccce492dcdac847765290b38e1c25a890 Mon Sep 17 00:00:00 2001 From: Luther Monson Date: Thu, 22 Oct 2026 22:04:42 -0700 Subject: [PATCH 2/2] fix(jobcgroup): never sweep a live job's parent cgroup The leaked-cgroup sweep removed any empty ephemerd-dind-* parent whose mtime was older than the grace. cgroupfs bumps a cgroup dir's mtime only when a child cgroup is created or removed, so a job without a memory limit that went more than 10 minutes between two dind siblings had an empty, "stale" parent while still running. An rmdir landing between runc's mkdir of the parent and mkdir of the leaf fails that sibling's start; under a memory limit whose runner-side setup failed, it can also drop the limits between Ensure and runc. The sweep now takes the live job set (runner containers in the runtime namespace plus the provisioning set, via Runtime.LiveJobIDs) and skips those jobs; if the set cannot be read the pass is skipped. Also moves the memoryLimit helper off runNodeDiskSweeper's doc comment and gofmts the dind Server literal. --- cmd/ephemerd/main.go | 46 +++++++++++++++++++-------- pkg/dind/dind.go | 2 +- pkg/jobcgroup/jobcgroup_linux.go | 26 +++++++++------ pkg/jobcgroup/jobcgroup_linux_test.go | 17 ++++++++-- pkg/jobcgroup/jobcgroup_other.go | 6 ++-- pkg/runtime/runtime.go | 23 ++++++++++++++ 6 files changed, 91 insertions(+), 29 deletions(-) diff --git a/cmd/ephemerd/main.go b/cmd/ephemerd/main.go index 5953e18..112f89b 100644 --- a/cmd/ephemerd/main.go +++ b/cmd/ephemerd/main.go @@ -1134,18 +1134,6 @@ func buildkitGCConfig(cfg *config.Config) buildkit.GCConfig { } } -// runNodeDiskSweeper is the periodic half of node disk hygiene, on one -// timer: an orphan sweep, a sweep of dead jobs' BuildKit export records, -// and a disk-pressure image collection pass. -// -// All three were previously startup-only or absent, which is why long-lived -// daemons accumulated build records, snapshots and images for their entire -// uptime. Each pass is cheap on an idle node — a container list, a couple -// of directory reads and one statfs — and the image collection is a no-op -// unless a watermark is crossed. -// -// Errors from one pass are logged and the loop continues; the next tick -// retries. // memoryLimit resolves [runtime] memory_limit_mb / memory_high_percent into // the job cgroup budget the runtime and dind apply. func memoryLimit(r config.RuntimeConfig) jobcgroup.Limits { @@ -1159,10 +1147,40 @@ func memoryLimit(r config.RuntimeConfig) jobcgroup.Limits { // (daemon crash, host reboot) and for the backlog older daemons left. const jobCgroupSweepGrace = 10 * time.Minute +// sweepJobCgroups runs the leaked job cgroup sweep, skipping every job rt is +// still running or provisioning. If the live set cannot be read the pass is +// skipped: removing a live job's parent is worse than leaking one a tick +// longer. +func sweepJobCgroups(ctx context.Context, rt *runtime.Runtime, log *slog.Logger) { + var live map[string]bool + if rt != nil { + var err error + if live, err = rt.LiveJobIDs(ctx); err != nil { + log.Debug("job cgroup sweep skipped: cannot list live jobs", "error", err) + return + } + } + jobcgroup.Sweep(jobCgroupSweepGrace, live, log) +} + +// runNodeDiskSweeper is the periodic half of node disk hygiene, on one +// timer: an orphan sweep, a sweep of dead jobs' BuildKit export records, +// and a disk-pressure image collection pass. +// +// All three were previously startup-only or absent, which is why long-lived +// daemons accumulated build records, snapshots and images for their entire +// uptime. Each pass is cheap on an idle node — a container list, a couple +// of directory reads and one statfs — and the image collection is a no-op +// unless a watermark is crossed. +// +// Errors from one pass are logged and the loop continues; the next tick +// retries. func runNodeDiskSweeper(ctx context.Context, gc *imagegc.Collector, rt *runtime.Runtime, c *containerdclient.Client, interval time.Duration, repair brokenChainRepair, log *slog.Logger) { log = log.With("component", "node-disk-sweeper", "interval", interval) log.Info("starting node disk sweeper") - jobcgroup.Sweep(jobCgroupSweepGrace, log) + startCtx, startCancel := context.WithTimeout(ctx, time.Minute) + sweepJobCgroups(startCtx, rt, log) + startCancel() ticker := time.NewTicker(interval) defer ticker.Stop() for { @@ -1186,7 +1204,7 @@ func runNodeDiskSweeper(ctx context.Context, gc *imagegc.Collector, rt *runtime. } } sweepDeadBuildRecords(passCtx, c, log) - jobcgroup.Sweep(jobCgroupSweepGrace, log) + sweepJobCgroups(passCtx, rt, log) // Correctness repair, not capacity policy — so it runs on // every tick whether or not image GC is enabled, and before // the collector, which would otherwise plan against records diff --git a/pkg/dind/dind.go b/pkg/dind/dind.go index 836c477..b980a39 100644 --- a/pkg/dind/dind.go +++ b/pkg/dind/dind.go @@ -261,7 +261,7 @@ func New(cfg Config) (*Server, error) { allowPrivileged: cfg.AllowPrivileged, pidsLimit: cfg.PidsLimit, memoryLimit: cfg.MemoryLimit, - stager: newBindStager(cfg.DataDir, cfg.JobID, cfg.Log), + stager: newBindStager(cfg.DataDir, cfg.JobID, cfg.Log), mirror: cfg.RegistryMirror, resultsURL: cfg.ResultsURL, log: cfg.Log.With("component", "dind", "job_id", cfg.JobID), diff --git a/pkg/jobcgroup/jobcgroup_linux.go b/pkg/jobcgroup/jobcgroup_linux.go index 58f6c85..692889a 100644 --- a/pkg/jobcgroup/jobcgroup_linux.go +++ b/pkg/jobcgroup/jobcgroup_linux.go @@ -109,15 +109,20 @@ func Remove(root, jobID string) error { } // SweepStale removes job parent cgroups under root that hold no processes and -// no child cgroups and have not changed for grace. It never removes one with -// either: a live job's parent always holds its containers' leaves, and with a -// job memory limit the runner's leaf is in it for the job's whole life. -// Without a limit, a live job's parent can be briefly empty between two -// sibling containers; removing it then is harmless because runc recreates it, -// and the grace period makes it unlikely in the first place. +// no child cgroups, whose directory has not changed for grace, and whose job +// is not in live. It never removes one with processes or children, and never +// one belonging to a job this daemon is still running: without a memory +// limit a live job's parent is empty between two of its sibling containers, +// and an rmdir landing inside runc's mkdir-parent/mkdir-leaf window would +// fail that container's start. live is keyed by job (runner container) id; +// nil means no job is live. +// +// The grace is measured against the directory's mtime, which cgroupfs bumps +// when a child cgroup is created or removed (not when processes join or a +// limit is written). It backstops a job the live set cannot see yet. // // Returns how many it removed. -func SweepStale(root string, grace time.Duration, now time.Time) (int, error) { +func SweepStale(root string, grace time.Duration, now time.Time, live map[string]bool) (int, error) { entries, err := os.ReadDir(root) if err != nil { return 0, fmt.Errorf("reading %s: %w", root, err) @@ -127,6 +132,9 @@ func SweepStale(root string, grace time.Duration, now time.Time) (int, error) { if !e.IsDir() || !strings.HasPrefix(e.Name(), Prefix) { continue } + if live[strings.TrimPrefix(e.Name(), Prefix)] { + continue + } dir := filepath.Join(root, e.Name()) if !idle(dir) { continue @@ -164,8 +172,8 @@ func idle(dir string) bool { // Sweep is SweepStale on the real cgroupfs, logged. A sweep failure is never // fatal — the leak it cleans up is slow and harmless in the short term. -func Sweep(grace time.Duration, log *slog.Logger) { - n, err := SweepStale(Root, grace, time.Now()) +func Sweep(grace time.Duration, live map[string]bool, log *slog.Logger) { + n, err := SweepStale(Root, grace, time.Now(), live) if err != nil { log.Debug("job cgroup sweep failed", "error", err) return diff --git a/pkg/jobcgroup/jobcgroup_linux_test.go b/pkg/jobcgroup/jobcgroup_linux_test.go index 8dd9997..506a70f 100644 --- a/pkg/jobcgroup/jobcgroup_linux_test.go +++ b/pkg/jobcgroup/jobcgroup_linux_test.go @@ -148,8 +148,11 @@ func TestSweepStale(t *testing.T) { if err := os.MkdirAll(recent, 0o755); err != nil { t.Fatal(err) } + // A live job between two sibling containers: empty and old, but the + // daemon is still running it. + liveJob := mk(Prefix+"live-job", map[string]string{"cgroup.procs": ""}) - n, err := SweepStale(root, 10*time.Minute, time.Now()) + n, err := SweepStale(root, 10*time.Minute, time.Now(), map[string]bool{"live-job": true}) if err != nil { t.Fatalf("SweepStale: %v", err) } @@ -161,11 +164,19 @@ func TestSweepStale(t *testing.T) { t.Errorf("%s: idle stale job cgroup not removed", filepath.Base(gone)) } } - for _, kept := range []string{busy, parent, other, recent} { + for _, kept := range []string{busy, parent, other, recent, liveJob} { if _, err := os.Stat(kept); err != nil { - t.Errorf("%s: must be kept (procs, children, not ours, or inside the grace period): %v", filepath.Base(kept), err) + t.Errorf("%s: must be kept (procs, children, not ours, inside the grace period, or live): %v", filepath.Base(kept), err) } } + + // Once the job is no longer live, its empty parent goes. + if n, err := SweepStale(root, 10*time.Minute, time.Now(), nil); err != nil || n != 1 { + t.Errorf("SweepStale with no live jobs = (%d, %v), want (1, nil)", n, err) + } + if _, err := os.Stat(liveJob); !errors.Is(err, os.ErrNotExist) { + t.Errorf("finished job's empty parent not removed") + } } func TestPaths(t *testing.T) { diff --git a/pkg/jobcgroup/jobcgroup_other.go b/pkg/jobcgroup/jobcgroup_other.go index 1152d35..f7a638f 100644 --- a/pkg/jobcgroup/jobcgroup_other.go +++ b/pkg/jobcgroup/jobcgroup_other.go @@ -14,7 +14,9 @@ func Ensure(_, _ string, _ Limits) error { return nil } func Remove(_, _ string) error { return nil } // SweepStale is a no-op off Linux. -func SweepStale(_ string, _ time.Duration, _ time.Time) (int, error) { return 0, nil } +func SweepStale(_ string, _ time.Duration, _ time.Time, _ map[string]bool) (int, error) { + return 0, nil +} // Sweep is a no-op off Linux. -func Sweep(_ time.Duration, _ *slog.Logger) {} +func Sweep(_ time.Duration, _ map[string]bool, _ *slog.Logger) {} diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index 170cf7e..b70c4ee 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -410,6 +410,29 @@ func (r *Runtime) SweepOrphans(ctx context.Context) error { return r.sweepOrphanState(ctx, live) } +// LiveJobIDs returns the ids of the jobs this runtime is running or +// provisioning: every runner container in the runtime namespace plus the +// in-flight provisioning set. The job cgroup sweep skips these, so it never +// removes a live job's parent cgroup even while that job momentarily has no +// container in it. +func (r *Runtime) LiveJobIDs(ctx context.Context) (map[string]bool, error) { + ctx = namespaces.WithNamespace(ctx, namespace) + containers, err := r.client.Containers(ctx) + if err != nil { + return nil, fmt.Errorf("listing containers: %w", err) + } + keep := make(map[string]struct{}, len(containers)) + for _, c := range containers { + keep[c.ID()] = struct{}{} + } + r.addProvisioning(keep) + live := make(map[string]bool, len(keep)) + for id := range keep { + live[id] = true + } + return live, nil +} + // sweepOrphanState is the shared body of CleanOrphans and SweepOrphans. // live is the set of container IDs whose on-disk state must be preserved; // nil means "nothing is live".