6 files changed,
+322,
-2
+3,
-1
1@@ -49,6 +49,8 @@ func cmdRunner(args []string) int {
2 owner := fs.String("owner-did", "", "DID of the repos' owner")
3 cloneURL := fs.String("clone-url", "", "clone URL template with {repo}, e.g. [email protected]:{repo} (required)")
4 maxConc := fs.Int("max-concurrent", 2, "segments running at once")
5+ slimAfter := fs.Duration("slim-after", runner.DefaultSlimAfter, "finished runs keep only their record (log, results, artifacts, state) after this; their checkouts and Nix GC roots go (negative: never)")
6+ deleteAfter := fs.Duration("delete-after", runner.DefaultDeleteAfter, "finished runs are deleted after this (negative: never)")
7 launcher := fs.String("launcher", "systemd", "systemd (sandboxed transient units) or local (child processes; development)")
8 shell := fs.String("shell", "devenv", "devenv, or bash (no devenv; tests)")
9 slice := fs.String("slice", "ci.slice", "systemd slice for segments")
10@@ -92,7 +94,7 @@ func cmdRunner(args []string) int {
11 }
12 cfg := runner.Config{
13 RunsDir: filepath.Join(*state, "runs"), CacheDir: filepath.Join(*state, "cache"), LocksDir: filepath.Join(*state, "locks"),
14- Knot: *knot, OwnerDid: *owner, CloneURL: *cloneURL, MaxConcurrent: *maxConc,
15+ Knot: *knot, OwnerDid: *owner, CloneURL: *cloneURL, MaxConcurrent: *maxConc, SlimAfter: *slimAfter, DeleteAfter: *deleteAfter,
16 }
17 if *ntfy != "" {
18 cfg.Notify = runner.Ntfy(*ntfy, *tokens["ntfy"], *appview, *ntfySuccess, logger)
+2,
-1
1@@ -226,8 +226,9 @@ func (r *Runner) ExpireGates(ctx context.Context) {
2 }
3 }
4
5-// Loop runs the gate expiry timer until ctx ends.
6+// Loop runs the gate expiry timer and the retention of finished runs until ctx ends.
7 func (r *Runner) Loop(ctx context.Context) {
8+ go r.pruneLoop(ctx)
9 for {
10 select {
11 case <-ctx.Done():
+148,
-0
1@@ -0,0 +1,148 @@
2+package runner
3+
4+import (
5+ "context"
6+ "os"
7+ "path/filepath"
8+ "strings"
9+ "time"
10+
11+ "git.kilimanjaro.io/ci/internal/engine"
12+)
13+
14+// Retention of finished runs. A run directory holds the run's record (small)
15+// and its checkouts, whose .devenv/ directories root Nix store paths for as
16+// long as they exist; nothing else ever frees them, so finished runs are
17+// pruned in two stages:
18+//
19+// - slimmed after Config.SlimAfter: everything but the record is removed
20+// (the checkouts and their GC roots); the run still shows its log,
21+// results and artifacts, and a manual action (rollback) can still read
22+// its state/ ($CI_STATE);
23+// - deleted after Config.DeleteAfter: the directory and the row go.
24+//
25+// Running, paused and pending runs are never touched, nor is a run with a
26+// child that has not finished.
27+
28+const (
29+ // DefaultSlimAfter and DefaultDeleteAfter apply when the Config leaves them zero.
30+ DefaultSlimAfter = 3 * 24 * time.Hour
31+ DefaultDeleteAfter = 60 * 24 * time.Hour
32+)
33+
34+// kept is what a slimmed run keeps.
35+var kept = map[string]bool{
36+ "state.json": true,
37+ "events.ndjson": true,
38+ "manifest.yml": true,
39+ "state": true,
40+ "artifacts": true,
41+ "results": true,
42+}
43+
44+func keptFile(name string) bool {
45+ if kept[name] {
46+ return true
47+ }
48+ ok, _ := filepath.Match("segment-*.log", name)
49+ return ok
50+}
51+
52+// Prune applies the retention once: it reports how many runs it slimmed and deleted.
53+func (r *Runner) Prune(now time.Time) (slimmed, deleted int) {
54+ slim, del := r.Cfg.SlimAfter, r.Cfg.DeleteAfter
55+ if slim == 0 {
56+ slim = DefaultSlimAfter
57+ }
58+ if del == 0 {
59+ del = DefaultDeleteAfter
60+ }
61+ rows, err := r.db().Query(`select ` + runCols + ` from runs`)
62+ if err != nil {
63+ r.Logger.Error("prune: list runs", "err", err)
64+ return
65+ }
66+ var runs []Run
67+ live := map[string]bool{} // parents of runs that are not finished
68+ for rows.Next() {
69+ x, err := scanRun(rows)
70+ if err != nil {
71+ break
72+ }
73+ if !engine.Terminal(x.Status) && x.Parent != "" {
74+ live[x.Parent] = true
75+ }
76+ runs = append(runs, x)
77+ }
78+ rows.Close()
79+ for _, x := range runs {
80+ if !engine.Terminal(x.Status) || live[x.ID] || !r.inRunsDir(x.Dir) {
81+ continue
82+ }
83+ t, err := time.Parse(time.RFC3339Nano, x.Updated)
84+ if err != nil {
85+ continue
86+ }
87+ age := now.Sub(t)
88+ switch {
89+ case del > 0 && age > del:
90+ // RemoveAll never follows a symlink (a working-copy run's main
91+ // checkout), so a developer's tree is safe.
92+ if err := os.RemoveAll(x.Dir); err != nil {
93+ r.Logger.Error("prune: delete run", "run", x.ID, "err", err)
94+ continue
95+ }
96+ if _, err := r.db().Exec(`delete from runs where id = ?`, x.ID); err != nil {
97+ r.Logger.Error("prune: delete row", "run", x.ID, "err", err)
98+ continue
99+ }
100+ deleted++
101+ case slim > 0 && age > slim:
102+ if r.slim(x.Dir) {
103+ slimmed++
104+ }
105+ }
106+ }
107+ return
108+}
109+
110+// slim removes everything in a run directory but its record, and reports
111+// whether it removed anything.
112+func (r *Runner) slim(dir string) bool {
113+ entries, err := os.ReadDir(dir)
114+ if err != nil {
115+ return false
116+ }
117+ removed := false
118+ for _, e := range entries {
119+ if keptFile(e.Name()) {
120+ continue
121+ }
122+ if err := os.RemoveAll(filepath.Join(dir, e.Name())); err != nil {
123+ r.Logger.Error("prune: slim run", "dir", dir, "entry", e.Name(), "err", err)
124+ continue
125+ }
126+ removed = true
127+ }
128+ return removed
129+}
130+
131+// inRunsDir guards the removals: a row's dir must be a direct child of RunsDir.
132+func (r *Runner) inRunsDir(dir string) bool {
133+ rel, err := filepath.Rel(r.Cfg.RunsDir, dir)
134+ return err == nil && rel != "." && !strings.HasPrefix(rel, "..") && !strings.ContainsRune(rel, filepath.Separator)
135+}
136+
137+// pruneLoop prunes at start and then hourly until ctx ends.
138+func (r *Runner) pruneLoop(ctx context.Context) {
139+ for {
140+ if s, d := r.Prune(time.Now()); s+d > 0 {
141+ r.Logger.Info("pruned finished runs", "slimmed", s, "deleted", d)
142+ }
143+ select {
144+ case <-ctx.Done():
145+ return
146+ case <-time.After(time.Hour):
147+ }
148+ }
149+}
+146,
-0
1@@ -0,0 +1,146 @@
2+package runner
3+
4+import (
5+ "os"
6+ "path/filepath"
7+ "testing"
8+ "time"
9+
10+ "git.kilimanjaro.io/ci/internal/engine"
11+ "git.kilimanjaro.io/ci/internal/eventlog"
12+)
13+
14+func newPruneRunner(t *testing.T, cfg Config) *Runner {
15+ t.Helper()
16+ root := t.TempDir()
17+ cfg.RunsDir = filepath.Join(root, "runs")
18+ cfg.CacheDir = filepath.Join(root, "cache")
19+ log, err := eventlog.Open(filepath.Join(root, "runner.db"))
20+ if err != nil {
21+ t.Fatal(err)
22+ }
23+ t.Cleanup(func() { log.Close() })
24+ r, err := New(cfg, log, nil, nil)
25+ if err != nil {
26+ t.Fatal(err)
27+ }
28+ return r
29+}
30+
31+// addRun makes a run directory with a record, a checkout (with a .devenv/)
32+// and a row last updated age ago.
33+func addRun(t *testing.T, r *Runner, id, status, parent string, age time.Duration) string {
34+ t.Helper()
35+ dir := engine.Dir(filepath.Join(r.Cfg.RunsDir, id))
36+ if err := dir.Create(); err != nil {
37+ t.Fatal(err)
38+ }
39+ write := func(p, s string) {
40+ t.Helper()
41+ if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
42+ t.Fatal(err)
43+ }
44+ if err := os.WriteFile(p, []byte(s), 0o644); err != nil {
45+ t.Fatal(err)
46+ }
47+ }
48+ write(dir.StateFile(), "{}")
49+ write(dir.Events(), "")
50+ write(dir.ManifestFile(), "")
51+ write(filepath.Join(dir.State(), "prev-systems.json"), "{}")
52+ write(filepath.Join(dir.Results(), "r"), "r")
53+ write(filepath.Join(dir.Artifacts(), "a"), "a")
54+ write(filepath.Join(dir.Path(), "segment-1.log"), "log")
55+ write(filepath.Join(dir.Path(), "repo", ".devenv", "gc", "shell"), "root")
56+ updated := time.Now().Add(-age).UTC().Format(time.RFC3339Nano)
57+ if _, err := r.db().Exec(`insert into runs (id, pipeline, rkey, workflow, repo, dir, parent, status, created, updated)
58+ values (?, 'p', ?, 'w', 'repo', ?, ?, ?, ?, ?)`, id, id, dir.Path(), parent, status, updated, updated); err != nil {
59+ t.Fatal(err)
60+ }
61+ return dir.Path()
62+}
63+
64+func exists(p string) bool { _, err := os.Stat(p); return err == nil }
65+
66+func TestPrune(t *testing.T) {
67+ day := 24 * time.Hour
68+ r := newPruneRunner(t, Config{SlimAfter: 3 * day, DeleteAfter: 60 * day})
69+ fresh := addRun(t, r, "fresh", "success", "", day)
70+ old := addRun(t, r, "old", "failure", "", 5*day)
71+ ancient := addRun(t, r, "ancient", "success", "", 90*day)
72+ paused := addRun(t, r, "paused", "paused", "", 90*day)
73+ running := addRun(t, r, "running", "running", "", 90*day)
74+ parent := addRun(t, r, "parent", "success", "", 90*day)
75+ addRun(t, r, "child", "running", "parent", day)
76+
77+ slimmed, deleted := r.Prune(time.Now())
78+ if slimmed != 1 || deleted != 1 {
79+ t.Fatalf("slimmed %d, deleted %d; want 1 and 1", slimmed, deleted)
80+ }
81+
82+ // fresh: untouched
83+ if !exists(filepath.Join(fresh, "repo", ".devenv", "gc", "shell")) {
84+ t.Error("a fresh run was slimmed")
85+ }
86+ // old: checkout gone, record kept (including state/ for a rollback)
87+ if exists(filepath.Join(old, "repo")) {
88+ t.Error("an old run kept its checkout")
89+ }
90+ for _, p := range []string{"state.json", "events.ndjson", "manifest.yml", "segment-1.log",
91+ "state/prev-systems.json", "results/r", "artifacts/a"} {
92+ if !exists(filepath.Join(old, p)) {
93+ t.Errorf("a slimmed run lost %s", p)
94+ }
95+ }
96+ // ancient: directory and row gone
97+ if exists(ancient) {
98+ t.Error("an ancient run kept its directory")
99+ }
100+ if _, err := r.GetRun("ancient"); err == nil {
101+ t.Error("an ancient run kept its row")
102+ }
103+ // never touched: unfinished runs, and a run with an unfinished child
104+ for name, dir := range map[string]string{"paused": paused, "running": running, "a parent with a live child": parent} {
105+ if !exists(filepath.Join(dir, "repo", ".devenv")) {
106+ t.Errorf("%s was pruned", name)
107+ }
108+ }
109+ // idempotent
110+ if s, d := r.Prune(time.Now()); s != 0 || d != 0 {
111+ t.Errorf("second pass slimmed %d, deleted %d", s, d)
112+ }
113+}
114+
115+func TestPruneDisabledAndSymlink(t *testing.T) {
116+ day := 24 * time.Hour
117+ r := newPruneRunner(t, Config{SlimAfter: -1, DeleteAfter: -1})
118+ dir := addRun(t, r, "old", "success", "", 400*day)
119+ if s, d := r.Prune(time.Now()); s != 0 || d != 0 || !exists(filepath.Join(dir, "repo")) {
120+ t.Fatalf("negative retention still pruned: %d, %d", s, d)
121+ }
122+
123+ // a working-copy run: the main checkout is a symlink to the developer's tree
124+ r = newPruneRunner(t, Config{SlimAfter: day, DeleteAfter: 2 * day})
125+ tree := t.TempDir()
126+ os.WriteFile(filepath.Join(tree, "mine"), []byte("x"), 0o644)
127+ dir = addRun(t, r, "wc", "success", "", 5*day)
128+ os.RemoveAll(filepath.Join(dir, "repo"))
129+ os.Symlink(tree, filepath.Join(dir, "repo"))
130+ r.Prune(time.Now())
131+ if !exists(filepath.Join(tree, "mine")) {
132+ t.Fatal("pruning followed a symlink into a working tree")
133+ }
134+}
135+
136+func TestPruneOutsideRunsDir(t *testing.T) {
137+ r := newPruneRunner(t, Config{})
138+ outside := t.TempDir()
139+ os.WriteFile(filepath.Join(outside, "keep"), []byte("x"), 0o644)
140+ old := time.Now().Add(-400 * 24 * time.Hour).UTC().Format(time.RFC3339Nano)
141+ r.db().Exec(`insert into runs (id, pipeline, rkey, workflow, repo, dir, status, created, updated)
142+ values ('x', 'p', 'x', 'w', 'r', ?, 'success', ?, ?)`, outside, old, old)
143+ r.Prune(time.Now())
144+ if !exists(filepath.Join(outside, "keep")) {
145+ t.Fatal("pruned a directory outside the runs dir")
146+ }
147+}
+5,
-0
1@@ -48,6 +48,11 @@ type Config struct {
2 CloneURL string
3 // MaxConcurrent limits segments running at once (default 2).
4 MaxConcurrent int
5+ // SlimAfter and DeleteAfter are the retention of finished runs (see
6+ // prune.go): after SlimAfter a run keeps only its record, after DeleteAfter
7+ // it is deleted. Zero means DefaultSlimAfter / DefaultDeleteAfter; negative
8+ // turns that stage off.
9+ SlimAfter, DeleteAfter time.Duration
10 // Notify is called for gates and finished runs (ntfy), may be nil.
11 Notify func(Notification)
12 }
+18,
-0
1@@ -152,6 +152,22 @@ in
2 type = types.ints.positive;
3 default = 2;
4 };
5+ retention = {
6+ slimAfter = mkOption {
7+ type = types.str;
8+ default = "72h";
9+ description = ''
10+ Finished runs keep only their record (log, results, artifacts, state/) after
11+ this long (a Go duration); their checkouts and the Nix GC roots under
12+ .devenv/ go, so the paths become collectable. "-1h" turns it off.
13+ '';
14+ };
15+ deleteAfter = mkOption {
16+ type = types.str;
17+ default = "1440h";
18+ description = "Finished runs are deleted entirely after this long (60 days). \"-1h\" turns it off.";
19+ };
20+ };
21 shell = mkOption {
22 type = types.enum [
23 "devenv"
24@@ -380,6 +396,8 @@ in
25 "--knot=${cfg.runner.knot}"
26 "--clone-url=${cfg.runner.cloneUrl}"
27 "--max-concurrent=${toString cfg.runner.maxConcurrent}"
28+ "--slim-after=${cfg.runner.retention.slimAfter}"
29+ "--delete-after=${cfg.runner.retention.deleteAfter}"
30 "--launcher=systemd"
31 "--shell=${cfg.runner.shell}"
32 "--slice=ci.slice"