krz/gitbay

ci: record runner polls, and admin runners !160

merged merged by cmc on 2026-09-02 03:29 UTC · krz/gitbay:stack-2-runners into main

8 files changed, +164 −0

Layout: unified · split

cmd/gitbay/main.go +1
@@ -91,6 +91,7 @@ func newRoot() *cobra.Command {
91 ), 91 ),
92 pass("invite", "issue a registration invite and mail its code: --email <address>", passOpts{server: []string{"admin", "invite"}}), 92 pass("invite", "issue a registration invite and mail its code: --email <address>", passOpts{server: []string{"admin", "invite"}}),
93 pass("stats", "instance statistics: counts and per-repository disk usage", passOpts{server: []string{"admin", "stats"}}), 93 pass("stats", "instance statistics: counts and per-repository disk usage", passOpts{server: []string{"admin", "stats"}}),
94 pass("runners", "runner accounts: last poll, scope, the build each holds", passOpts{server: []string{"admin", "runners"}}),
94 group("repo", "any repository, for moderation (audited)", 95 group("repo", "any repository, for moderation (audited)",
95 pass("list", "every repository with size and last push: [--owner o] [--visibility v] [--limit n] [--cursor c]", passOpts{server: []string{"admin", "repo", "list"}}), 96 pass("list", "every repository with size and last push: [--owner o] [--visibility v] [--limit n] [--cursor c]", passOpts{server: []string{"admin", "repo", "list"}}),
96 pass("archive", "archive a repository: <owner/name>", passOpts{server: []string{"admin", "repo", "archive"}}), 97 pass("archive", "archive a repository: <owner/name>", passOpts{server: []string{"admin", "repo", "archive"}}),
cmd/gitbayd/main.go +1
@@ -336,6 +336,7 @@ func adminCmd() *cobra.Command {
336 configCmd, 336 configCmd,
337 hostCmd("invite --email <address>", "issue a registration invite and email its code", "admin", "invite"), 337 hostCmd("invite --email <address>", "issue a registration invite and email its code", "admin", "invite"),
338 hostCmd("stats [--json]", "instance statistics: counts and per-repository disk usage", "admin", "stats"), 338 hostCmd("stats [--json]", "instance statistics: counts and per-repository disk usage", "admin", "stats"),
339 hostCmd("runners [--json]", "runner accounts: last poll, scope, the build each holds", "admin", "runners"),
339 hostCmd("audit [--limit n] [--json]", "print the security audit log, newest first", "audit"), 340 hostCmd("audit [--limit n] [--json]", "print the security audit log, newest first", "audit"),
340 backupCmd(), 341 backupCmd(),
341 gcCmd(), 342 gcCmd(),
e2e/reap_test.go +61
@@ -2,6 +2,7 @@ package e2e
2 2
3import ( 3import (
4 "encoding/json" 4 "encoding/json"
5 "fmt"
5 "os" 6 "os"
6 "path/filepath" 7 "path/filepath"
7 "strings" 8 "strings"
@@ -78,3 +79,63 @@ func TestStaleBuildReapedWithoutRunner(t *testing.T) {
78 t.Fatalf("log lacks the abandonment note:\n%s", out) 79 t.Fatalf("log lacks the abandonment note:\n%s", out)
79 } 80 }
80} 81}
82
83func TestAdminRunners(t *testing.T) {
84 inst := startInstance(t)
85 rootKey := inst.newKey(t, "root")
86 aliceKey := inst.newKey(t, "alice")
87 runnerKey := inst.newKey(t, "ci")
88 inst.admin(t, "admin", "user", "create", "root", "--key", rootKey+".pub", "--admin")
89 inst.admin(t, "admin", "user", "create", "alice", "--key", aliceKey+".pub")
90 inst.admin(t, "admin", "user", "create", "ci", "--key", runnerKey+".pub", "--admin")
91 if _, _, code := inst.ssh(t, aliceKey, "", "admin", "runners"); code != 4 {
92 t.Fatal("non-admin listed runners")
93 }
94 if out, _, code := inst.ssh(t, rootKey, "", "admin", "runners", "--json"); code != 0 || strings.TrimSpace(out) != `{"protocol_version":1,"data":[]}` {
95 t.Fatalf("no runners yet: exit %d %s", code, out)
96 }
97 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 {
98 t.Fatal("repo create failed")
99 }
100 // An idle poll registers the runner with its scope.
101 if _, _, code := inst.ssh(t, runnerKey, "", "runner", "next", "alice/app"); code != 0 {
102 t.Fatal("runner next failed")
103 }
104 out, _, _ := inst.ssh(t, rootKey, "", "admin", "runners")
105 if !strings.HasPrefix(out, "ci\t") || !strings.Contains(out, "\talice/app\tidle") {
106 t.Fatalf("idle runner row:\n%s", out)
107 }
108 work := t.TempDir()
109 env := inst.gitEnv(aliceKey)
110 mustGit(t, work, env, "clone", inst.sshURL("alice/app"), "w")
111 dir := filepath.Join(work, "w")
112 os.MkdirAll(filepath.Join(dir, ".gitbay"), 0o755)
113 os.WriteFile(filepath.Join(dir, ".gitbay", "ci.yml"), []byte("jobs:\n ok:\n steps:\n - echo fine\n"), 0o644)
114 mustGit(t, dir, env, "checkout", "-q", "-b", "main")
115 mustGit(t, dir, env, "add", ".")
116 mustGit(t, dir, env, "commit", "-q", "-m", "ci")
117 mustGit(t, dir, env, "push", "-q", "origin", "main")
118 out, _, code := inst.ssh(t, runnerKey, "", "runner", "next", "--json")
119 if code != 0 || !strings.Contains(out, `"job":"ok"`) {
120 t.Fatalf("claim: %s", out)
121 }
122 var claim struct {
123 Data struct {
124 ID int64 `json:"id"`
125 } `json:"data"`
126 }
127 json.Unmarshal([]byte(out), &claim)
128 if out, _, _ := inst.ssh(t, rootKey, "", "admin", "runners"); !strings.Contains(out, "\tany\talice/app #1 ok since ") {
129 t.Fatalf("holding runner row:\n%s", out)
130 }
131 if _, _, code := inst.ssh(t, runnerKey, "", "runner", "done", fmt.Sprint(claim.Data.ID), "success"); code != 0 {
132 t.Fatal("runner done failed")
133 }
134 if out, _, _ := inst.ssh(t, rootKey, "", "admin", "runners"); !strings.Contains(out, "\tany\tidle") {
135 t.Fatalf("runner still holds a build after done:\n%s", out)
136 }
137 // Host-local, the same read.
138 if out := inst.admin(t, "admin", "runners", "--json"); !strings.Contains(out, `"username":"ci"`) {
139 t.Fatalf("host runners:\n%s", out)
140 }
141}
internal/control/admin.go +30
@@ -29,6 +29,10 @@ func init() {
29 Summary: "remove instance admin from an account (never the last one)", 29 Summary: "remove instance admin from an account (never the last one)",
30 Usage: "admin user demote <username>", 30 Usage: "admin user demote <username>",
31 SSHOnly: true, Run: runAdminUserDemote}) 31 SSHOnly: true, Run: runAdminUserDemote})
32 register(Command{Path: []string{"admin", "runners"},
33 Summary: "runner accounts: last poll, scope, the build each holds (instance admins)",
34 Usage: "admin runners",
35 ReadOnly: true, SSHOnly: true, Run: runAdminRunners})
32 register(Command{Path: []string{"admin", "repo", "list"}, 36 register(Command{Path: []string{"admin", "repo", "list"},
33 Summary: "list every repository with size and last push (instance admins)", 37 Summary: "list every repository with size and last push (instance admins)",
34 Usage: "admin repo list [--owner <name>] [--visibility public|private] [--limit <n>] [--cursor <c>]", 38 Usage: "admin repo list [--owner <name>] [--visibility public|private] [--limit <n>] [--cursor <c>]",
@@ -449,3 +453,29 @@ func runAdminRepoDelete(c *Ctx, args []string) int {
449 c.Store.Audit(c.User.ID, "admin repo.delete", map[string]any{"repo": repo.Path()}) 453 c.Store.Audit(c.User.ID, "admin repo.delete", map[string]any{"repo": repo.Path()})
450 return protocol.ExitOK 454 return protocol.ExitOK
451} 455}
456
457func runAdminRunners(c *Ctx, args []string) int {
458 if code := requireInstanceAdmin(c); code >= 0 {
459 return code
460 }
461 if len(args) != 0 {
462 return c.fail(protocol.ExitUsage, "usage: admin runners")
463 }
464 runners, err := c.Store.ListRunners()
465 if err != nil {
466 return c.fail(protocol.ExitFailure, "%v", err)
467 }
468 return c.emit(runners, func(w io.Writer) {
469 for _, r := range runners {
470 scope := r.Scope
471 if scope == "" {
472 scope = "any"
473 }
474 held := "idle"
475 if r.BuildNumber != 0 {
476 held = fmt.Sprintf("%s #%d %s since %s", r.BuildRepo, r.BuildNumber, r.BuildJob, r.StartedAt)
477 }
478 fmt.Fprintf(w, "%s\t%s\t%s\t%s\n", r.Username, r.LastSeen, scope, held)
479 }
480 })
481}
internal/control/build.go +3
@@ -329,6 +329,8 @@ func runRunnerNext(c *Ctx, args []string) int {
329 if err != nil { 329 if err != nil {
330 return c.fail(protocol.ExitFailure, "%v", err) 330 return c.fail(protocol.ExitFailure, "%v", err)
331 } 331 }
332 // The poll itself is the runner's heartbeat: admin runners reads it.
333 c.Store.TouchRunner(c.User.ID, strings.Join(args, ","), b.ID)
332 if !ok { 334 if !ok {
333 return c.emit(map[string]any{}, func(w io.Writer) { fmt.Fprintln(w, "no pending builds") }) 335 return c.emit(map[string]any{}, func(w io.Writer) { fmt.Fprintln(w, "no pending builds") })
334 } 336 }
@@ -412,6 +414,7 @@ func runRunnerDone(c *Ctx, args []string) int {
412 if err := c.Store.FinishBuild(id, args[1]); err != nil { 414 if err := c.Store.FinishBuild(id, args[1]); err != nil {
413 return c.fail(protocol.ExitFailure, "finishing build %d: %v", id, err) 415 return c.fail(protocol.ExitFailure, "finishing build %d: %v", id, err)
414 } 416 }
417 c.Store.RunnerDone(c.User.ID)
415 repo, err := c.Store.RepoByID(b.RepoID) 418 repo, err := c.Store.RepoByID(b.RepoID)
416 if err != nil { 419 if err != nil {
417 return c.fail(protocol.ExitFailure, "%v", err) 420 return c.fail(protocol.ExitFailure, "%v", err)
internal/store/migrations/0030_runner_seen.down.sql added +1
@@ -0,0 +1 @@
1DROP TABLE runner_seen;
internal/store/migrations/0030_runner_seen.up.sql added +8
@@ -0,0 +1,8 @@
1-- One row per runner account: when it last polled, the repositories it
2-- asked to be limited to, and the build it currently holds.
3CREATE TABLE runner_seen (
4 user_id INTEGER PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
5 last_seen TEXT NOT NULL,
6 scope TEXT NOT NULL DEFAULT '',
7 build_id INTEGER REFERENCES builds(id) ON DELETE SET NULL
8);
internal/store/runners.go added +59
@@ -0,0 +1,59 @@
1package store
2
3// Runner is one runner account as the instance admin sees it.
4type Runner struct {
5 Username string `json:"username"`
6 LastSeen string `json:"last_seen"`
7 Scope string `json:"scope,omitempty"` // comma-joined owner/name, "" for any
8 // The build it holds, if any.
9 BuildRepo string `json:"build_repo,omitempty"`
10 BuildNumber int64 `json:"build_number,omitempty"`
11 BuildJob string `json:"build_job,omitempty"`
12 StartedAt string `json:"started_at,omitempty"`
13}
14
15// TouchRunner records a poll: the time, the scope the runner asked for,
16// and the build it just claimed (0 for none).
17func (s *Store) TouchRunner(userID int64, scope string, buildID int64) error {
18 _, err := s.DB.Exec(`INSERT INTO runner_seen (user_id, last_seen, scope, build_id)
19 VALUES (?1, strftime('%Y-%m-%dT%H:%M:%fZ','now'), ?2, NULLIF(?3, 0))
20 ON CONFLICT (user_id) DO UPDATE SET
21 last_seen = excluded.last_seen, scope = excluded.scope,
22 build_id = COALESCE(excluded.build_id, runner_seen.build_id)`,
23 userID, scope, buildID)
24 return err
25}
26
27// RunnerDone records that the runner reported and holds nothing now.
28func (s *Store) RunnerDone(userID int64) error {
29 _, err := s.DB.Exec(`UPDATE runner_seen SET last_seen = strftime('%Y-%m-%dT%H:%M:%fZ','now'),
30 build_id = NULL WHERE user_id = ?`, userID)
31 return err
32}
33
34// ListRunners lists every account that has ever polled as a runner,
35// most recently seen first.
36func (s *Store) ListRunners() ([]Runner, error) {
37 rows, err := s.DB.Query(`SELECT u.username, r.last_seen, r.scope,
38 COALESCE(COALESCE(bu.username, bo.name) || '/' || br.name, ''),
39 COALESCE(b.number, 0), COALESCE(b.job, ''), COALESCE(b.started_at, '')
40 FROM runner_seen r JOIN users u ON u.id = r.user_id
41 LEFT JOIN builds b ON b.id = r.build_id AND b.status = 'running'
42 LEFT JOIN repos br ON br.id = b.repo_id
43 LEFT JOIN users bu ON br.owner_kind = 'user' AND bu.id = br.owner_id
44 LEFT JOIN orgs bo ON br.owner_kind = 'org' AND bo.id = br.owner_id
45 ORDER BY r.last_seen DESC`)
46 if err != nil {
47 return nil, err
48 }
49 defer rows.Close()
50 var out []Runner
51 for rows.Next() {
52 var r Runner
53 if err := rows.Scan(&r.Username, &r.LastSeen, &r.Scope, &r.BuildRepo, &r.BuildNumber, &r.BuildJob, &r.StartedAt); err != nil {
54 return nil, err
55 }
56 out = append(out, r)
57 }
58 return out, rows.Err()
59}