launch.go
1package runner
2
3import (
4 "context"
5 "fmt"
6 "io/fs"
7 "maps"
8 "os"
9 "os/exec"
10 "os/user"
11 "path/filepath"
12 "slices"
13 "strconv"
14 "strings"
15 "syscall"
16
17 "git.kilimanjaro.io/ci/internal/engine"
18 "git.kilimanjaro.io/ci/internal/lex"
19)
20
21// Segment is one engine process: a fresh run, a manual run starting at an
22// action, or a gate answer.
23type Segment struct {
24 From string // ci runner-exec --from
25 Choose int // 1-based option, 0 for none
26 Dismiss bool
27 Expire bool
28 Actor string
29}
30
31func (s Segment) args() []string {
32 var a []string
33 if s.From != "" {
34 a = append(a, "--from", s.From)
35 }
36 if s.Choose > 0 {
37 a = append(a, "--choose", strconv.Itoa(s.Choose))
38 }
39 if s.Dismiss {
40 a = append(a, "--dismiss")
41 }
42 if s.Expire {
43 a = append(a, "--expire")
44 }
45 if s.Actor != "" {
46 a = append(a, "--actor", s.Actor)
47 }
48 return a
49}
50
51// Job is what a Launcher starts.
52type Job struct {
53 Run string // run id
54 Dir string // run directory
55 Repo string
56 Segment int
57 Args []string // ci runner-exec arguments, without the run dir
58}
59
60// Launcher starts and inspects engine processes.
61type Launcher interface {
62 // Start starts the job and returns a handle for Alive and Stop.
63 Start(ctx context.Context, j Job) (string, error)
64 // Alive reports whether the process behind handle is still running.
65 Alive(handle string) bool
66 // Stop terminates it.
67 Stop(handle string) error
68}
69
70// Local runs `ci runner-exec` as a child process: for development and tests.
71type Local struct {
72 Engine string // path of the ci binary
73 Shell string // devenv or bash
74 LocksDir string
75 Env []string // extra environment
76}
77
78func (l *Local) Start(ctx context.Context, j Job) (string, error) {
79 args := append([]string{"runner-exec", "--shell", orDefault(l.Shell, "devenv"), "--locks-dir", l.LocksDir}, j.Args...)
80 cmd := exec.Command(l.Engine, append(args, j.Dir)...)
81 cmd.Env = append(os.Environ(), l.Env...)
82 logf, err := os.OpenFile(filepath.Join(j.Dir, fmt.Sprintf("segment-%d.log", j.Segment)), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644)
83 if err != nil {
84 return "", err
85 }
86 cmd.Stdout, cmd.Stderr = logf, logf
87 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
88 if err := cmd.Start(); err != nil {
89 logf.Close()
90 return "", err
91 }
92 go func() { _ = cmd.Wait(); logf.Close() }()
93 return "pid:" + strconv.Itoa(cmd.Process.Pid), nil
94}
95
96func (l *Local) Alive(handle string) bool {
97 pid, err := strconv.Atoi(strings.TrimPrefix(handle, "pid:"))
98 if err != nil {
99 return false
100 }
101 // a zombie (exited, not yet reaped) still answers signal 0
102 if b, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)); err == nil {
103 if f := strings.Fields(string(b)); len(f) > 2 && f[2] == "Z" {
104 return false
105 }
106 }
107 return syscall.Kill(pid, 0) == nil
108}
109
110func (l *Local) Stop(handle string) error {
111 pid, err := strconv.Atoi(strings.TrimPrefix(handle, "pid:"))
112 if err != nil {
113 return err
114 }
115 return syscall.Kill(-pid, syscall.SIGTERM)
116}
117
118// Systemd runs each segment as a transient unit in a sandbox (DESIGN.md
119// §10): its own slice, an unprivileged user, a read-only system, a private
120// /tmp, and optionally its own network namespace through pasta. Transient
121// units outlive the runner, so a runner restart (or a self-deploy) doesn't
122// kill a run.
123type Systemd struct {
124 Engine string // path of the ci binary
125 Shell string
126 LocksDir string
127 Slice string // default ci.slice
128 User string // default ci
129 // CacheDir holds per-repo HOMEs (shared state between runs of a repo).
130 CacheDir string
131 // Credentials are loaded into each segment with LoadCredential=
132 // (name -> file, read by systemd as root): age-identity becomes
133 // SECRETS_IDENTITY, ssh-key the segment's SSH identity (runner-exec).
134 Credentials map[string]string
135 // Pasta, if set, is the pasta binary: each segment gets its own network
136 // namespace with egress, so dev services can't clash with the host's.
137 Pasta string
138 // PastaHostPorts are TCP ports on the segment's localhost forwarded to
139 // the host's (e.g. the forge's SSH port, so a localhost clone URL works).
140 PastaHostPorts []int
141 // Properties are extra unit properties (e.g. MemoryMax=8G).
142 Properties []string
143 // Env is extra environment, e.g. CI_CLONE_URL.
144 Env []string
145}
146
147// UnitName is the transient unit of a run's segment.
148func UnitName(run string, segment int) string {
149 return fmt.Sprintf("ci-%s-%d", run, segment)
150}
151
152// Command returns the systemd-run command line for j.
153func (s *Systemd) Command(j Job) []string {
154 home := filepath.Join(s.CacheDir, "home", j.Repo)
155 argv := []string{"systemd-run",
156 "--unit=" + UnitName(j.Run, j.Segment),
157 "--slice=" + orDefault(s.Slice, "ci.slice"),
158 "--uid=" + orDefault(s.User, "ci"),
159 "--gid=" + orDefault(s.User, "ci"),
160 "--collect", "--quiet", "--no-block",
161 "--working-directory=" + j.Dir,
162 "--setenv=HOME=" + home,
163 "--setenv=XDG_CACHE_HOME=" + filepath.Join(home, ".cache"),
164 "-p", "ProtectSystem=strict",
165 "-p", "ProtectHome=yes",
166 "-p", "PrivateTmp=yes",
167 "-p", "NoNewPrivileges=yes",
168 "-p", "ReadWritePaths=" + strings.Join([]string{j.Dir, home, s.LocksDir, "/nix/var/nix/daemon-socket"}, " "),
169 "-p", "StandardOutput=append:" + filepath.Join(j.Dir, fmt.Sprintf("segment-%d.log", j.Segment)),
170 "-p", "StandardError=inherit",
171 }
172 for _, name := range slices.Sorted(maps.Keys(s.Credentials)) {
173 argv = append(argv, "-p", "LoadCredential="+name+":"+s.Credentials[name])
174 }
175 for _, e := range s.Env {
176 argv = append(argv, "--setenv="+e)
177 }
178 for _, p := range s.Properties {
179 argv = append(argv, "-p", p)
180 }
181 argv = append(argv, "--")
182 if s.Pasta != "" {
183 // pasta spawns the engine in new user and network namespaces:
184 // egress through user-mode NAT, nothing forwarded in, the listed
185 // ports of its localhost forwarded to the host's. Caveats: the
186 // engine runs as uid 0 of that user namespace (tools that refuse
187 // root, like postgres' initdb, won't run) and the hostname reads
188 // pasta-<host>. Opt-in; see DESIGN.md §10.
189 argv = append(argv, s.PastaCommand()...)
190 argv = append(argv, "--")
191 }
192 argv = append(argv, s.Engine, "runner-exec", "--shell", orDefault(s.Shell, "devenv"), "--locks-dir", s.LocksDir)
193 argv = append(argv, j.Args...)
194 return append(argv, j.Dir)
195}
196
197func (s *Systemd) Start(ctx context.Context, j Job) (string, error) {
198 home := filepath.Join(s.CacheDir, "home", j.Repo)
199 if err := os.MkdirAll(home, 0o755); err != nil {
200 return "", err
201 }
202 // the runner prepares as root; the segment runs as User and writes the
203 // run dir, its repo's HOME and the locks
204 if err := chownTree(orDefault(s.User, "ci"), j.Dir, home, s.LocksDir); err != nil {
205 return "", err
206 }
207
208 argv := s.Command(j)
209 out, err := exec.CommandContext(ctx, argv[0], argv[1:]...).CombinedOutput()
210 if err != nil {
211 return "", fmt.Errorf("systemd-run: %v: %s", err, strings.TrimSpace(string(out)))
212 }
213 return "unit:" + UnitName(j.Run, j.Segment), nil
214}
215
216// PastaCommand is pasta's part of a segment's command line.
217func (s *Systemd) PastaCommand() []string {
218 hostPorts := "none"
219 if len(s.PastaHostPorts) > 0 {
220 var ps []string
221 for _, p := range s.PastaHostPorts {
222 ps = append(ps, strconv.Itoa(p))
223 }
224 hostPorts = strings.Join(ps, ",")
225 }
226 return []string{s.Pasta, "--config-net", "--quiet", "-t", "none", "-u", "none", "-T", hostPorts, "-U", "none"}
227}
228
229func (s *Systemd) Alive(handle string) bool {
230 unit := strings.TrimPrefix(handle, "unit:")
231 out, _ := exec.Command("systemctl", "show", "-p", "ActiveState", "--value", unit+".service").Output()
232 st := strings.TrimSpace(string(out))
233 return st == "active" || st == "activating" || st == "deactivating" || st == "reloading"
234}
235
236func (s *Systemd) Stop(handle string) error {
237 return exec.Command("systemctl", "stop", "--no-block", strings.TrimPrefix(handle, "unit:")+".service").Run()
238}
239
240// chownTree hands paths (recursively) to user, unless we aren't root.
241func chownTree(name string, paths ...string) error {
242 if os.Geteuid() != 0 {
243 return nil
244 }
245 u, err := user.Lookup(name)
246 if err != nil {
247 return err
248 }
249 uid, _ := strconv.Atoi(u.Uid)
250 gid, _ := strconv.Atoi(u.Gid)
251 for _, p := range paths {
252 err := filepath.WalkDir(p, func(path string, d fs.DirEntry, err error) error {
253 if err != nil {
254 return err
255 }
256 return os.Lchown(path, uid, gid)
257 })
258 if err != nil {
259 return err
260 }
261 }
262 return nil
263}
264
265func orDefault(s, d string) string {
266 if s == "" {
267 return d
268 }
269 return s
270}
271
272// launch queues a segment of run: it starts as soon as a slot is free, and
273// its events are followed from then on. It returns at once; a segment that
274// can't be started ends the run as lost.
275func (r *Runner) launch(ctx context.Context, run string, seg Segment) error {
276 if _, err := r.GetRun(run); err != nil {
277 return err
278 }
279 ctx = context.WithoutCancel(ctx)
280 go func() {
281 r.slots <- struct{}{}
282 x, err := r.GetRun(run)
283 if err == nil {
284 n := x.segment + 1
285 var handle string
286 handle, err = r.Launcher.Start(ctx, Job{Run: run, Dir: x.Dir, Repo: x.Repo, Segment: n, Args: seg.args()})
287 if err == nil {
288 if err := r.updateRun(run, `segment = ?, handle = ?, status = ?`, n, handle, lex.StatusRunning); err != nil {
289 r.Logger.Error("update run", "run", run, "err", err)
290 }
291 r.Logger.Info("segment started", "run", run, "segment", n, "handle", handle)
292 if n == 1 && x.Parent == "" {
293 r.status(x.Pipeline, x.Workflow, lex.StatusRunning, "")
294 }
295 r.follow(ctx, run, handle) // releases the slot when the segment ends
296 return
297 }
298 }
299 r.Logger.Error("start segment", "run", run, "err", err)
300 if x.Dir != "" {
301 _ = engine.MarkLost(engine.Dir(x.Dir), "could not start the run: "+err.Error())
302 r.follow(ctx, run, "") // settles it as failed and releases the slot
303 return
304 }
305 <-r.slots
306 }()
307 return nil
308}
