alice.pds.demo.boltless.dev
This repository has no description
1package nixery
2
3import (
4 "bufio"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "log/slog"
10 "path"
11 "runtime"
12 "sync"
13 "time"
14
15 "github.com/docker/docker/api/types/container"
16 "github.com/docker/docker/api/types/image"
17 "github.com/docker/docker/api/types/mount"
18 "github.com/docker/docker/api/types/network"
19 "github.com/docker/docker/client"
20 "github.com/docker/docker/pkg/stdcopy"
21 "gopkg.in/yaml.v3"
22 "tangled.org/core/api/tangled"
23 "tangled.org/core/log"
24 "tangled.org/core/spindle/config"
25 "tangled.org/core/spindle/engine"
26 "tangled.org/core/spindle/models"
27 "tangled.org/core/spindle/secrets"
28)
29
30const (
31 workspaceDir = "/tangled/workspace"
32 homeDir = "/tangled/home"
33)
34
35type cleanupFunc func(context.Context) error
36
37type Engine struct {
38 docker client.APIClient
39 l *slog.Logger
40 cfg *config.Config
41
42 cleanupMu sync.Mutex
43 cleanup map[string][]cleanupFunc
44}
45
46type Step struct {
47 name string
48 kind models.StepKind
49 command string
50 environment map[string]string
51}
52
53func (s Step) Name() string {
54 return s.name
55}
56
57func (s Step) Command() string {
58 return s.command
59}
60
61func (s Step) Kind() models.StepKind {
62 return s.kind
63}
64
65// setupSteps get added to start of Steps
66type setupSteps []models.Step
67
68// addStep adds a step to the beginning of the workflow's steps.
69func (ss *setupSteps) addStep(step models.Step) {
70 *ss = append(*ss, step)
71}
72
73type addlFields struct {
74 image string
75 container string
76}
77
78func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
79 swf := &models.Workflow{}
80 addl := addlFields{}
81
82 dwf := &struct {
83 Steps []struct {
84 Command string `yaml:"command"`
85 Name string `yaml:"name"`
86 Environment map[string]string `yaml:"environment"`
87 } `yaml:"steps"`
88 Dependencies map[string][]string `yaml:"dependencies"`
89 Environment map[string]string `yaml:"environment"`
90 }{}
91 err := yaml.Unmarshal([]byte(twf.Raw), &dwf)
92 if err != nil {
93 return nil, err
94 }
95
96 for _, dstep := range dwf.Steps {
97 sstep := Step{}
98 sstep.environment = dstep.Environment
99 sstep.command = dstep.Command
100 sstep.name = dstep.Name
101 sstep.kind = models.StepKindUser
102 swf.Steps = append(swf.Steps, sstep)
103 }
104 swf.Name = twf.Name
105 swf.Environment = dwf.Environment
106 addl.image = workflowImage(dwf.Dependencies, e.cfg.NixeryPipelines.Nixery)
107
108 setup := &setupSteps{}
109
110 setup.addStep(nixConfStep())
111 setup.addStep(models.BuildCloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev))
112 // this step could be empty
113 if s := dependencyStep(dwf.Dependencies); s != nil {
114 setup.addStep(*s)
115 }
116
117 // append setup steps in order to the start of workflow steps
118 swf.Steps = append(*setup, swf.Steps...)
119 swf.Data = addl
120
121 return swf, nil
122}
123
124func (e *Engine) WorkflowTimeout() time.Duration {
125 workflowTimeoutStr := e.cfg.NixeryPipelines.WorkflowTimeout
126 workflowTimeout, err := time.ParseDuration(workflowTimeoutStr)
127 if err != nil {
128 e.l.Error("failed to parse workflow timeout", "error", err, "timeout", workflowTimeoutStr)
129 workflowTimeout = 5 * time.Minute
130 }
131
132 return workflowTimeout
133}
134
135func workflowImage(deps map[string][]string, nixery string) string {
136 var dependencies string
137 for reg, ds := range deps {
138 if reg == "nixpkgs" {
139 dependencies = path.Join(ds...)
140 }
141 }
142
143 // load defaults from somewhere else
144 dependencies = path.Join(dependencies, "bash", "git", "coreutils", "nix")
145
146 if runtime.GOARCH == "arm64" {
147 dependencies = path.Join("arm64", dependencies)
148 }
149
150 return path.Join(nixery, dependencies)
151}
152
153func New(ctx context.Context, cfg *config.Config) (*Engine, error) {
154 dcli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
155 if err != nil {
156 return nil, err
157 }
158
159 l := log.FromContext(ctx).With("component", "spindle")
160
161 e := &Engine{
162 docker: dcli,
163 l: l,
164 cfg: cfg,
165 }
166
167 e.cleanup = make(map[string][]cleanupFunc)
168
169 return e, nil
170}
171
172func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error {
173 /// -------------------------INITIAL SETUP------------------------------------------
174 l := e.l.With("workflow", wid)
175 l.Info("setting up workflow")
176
177 setupStep := Step{
178 name: "nixery image pull",
179 kind: models.StepKindSystem,
180 }
181 setupStepIdx := -1
182
183 wfLogger.ControlWriter(setupStepIdx, setupStep, models.StepStatusStart).Write([]byte{0})
184 defer wfLogger.ControlWriter(setupStepIdx, setupStep, models.StepStatusEnd).Write([]byte{0})
185
186 /// -------------------------NETWORK CREATION---------------------------------------
187 _, err := e.docker.NetworkCreate(ctx, networkName(wid), network.CreateOptions{
188 Driver: "bridge",
189 })
190 if err != nil {
191 return err
192 }
193
194 e.registerCleanup(wid, func(ctx context.Context) error {
195 if err := e.docker.NetworkRemove(ctx, networkName(wid)); err != nil {
196 return fmt.Errorf("removing network: %w", err)
197 }
198 return nil
199 })
200
201 /// -------------------------IMAGE PULL---------------------------------------------
202 addl := wf.Data.(addlFields)
203 l.Info("pulling image", "image", addl.image)
204 fmt.Fprintf(
205 wfLogger.DataWriter(setupStepIdx, "stdout"),
206 "pulling image: %s",
207 addl.image,
208 )
209
210 reader, err := e.docker.ImagePull(ctx, addl.image, image.PullOptions{})
211 if err != nil {
212 l.Error("pipeline image pull failed!", "error", err.Error())
213 fmt.Fprintf(wfLogger.DataWriter(setupStepIdx, "stderr"), "image pull failed: %s", err)
214 return fmt.Errorf("pulling image: %w", err)
215 }
216 defer reader.Close()
217
218 scanner := bufio.NewScanner(reader)
219 for scanner.Scan() {
220 line := scanner.Text()
221 wfLogger.DataWriter(setupStepIdx, "stdout").Write([]byte(line))
222 l.Info("image pull progress", "stdout", line)
223 }
224
225 /// -------------------------CONTAINER CREATION-------------------------------------
226 l.Info("creating container")
227 wfLogger.DataWriter(setupStepIdx, "stdout").Write([]byte("creating container..."))
228
229 resp, err := e.docker.ContainerCreate(ctx, &container.Config{
230 Image: addl.image,
231 Cmd: []string{"cat"},
232 OpenStdin: true, // so cat stays alive :3
233 Tty: false,
234 Hostname: "spindle",
235 WorkingDir: workspaceDir,
236 Labels: map[string]string{
237 "sh.tangled.pipeline/workflow_id": wid.String(),
238 },
239 // TODO(winter): investigate whether environment variables passed here
240 // get propagated to ContainerExec processes
241 }, &container.HostConfig{
242 Mounts: []mount.Mount{
243 {
244 Type: mount.TypeTmpfs,
245 Target: "/tmp",
246 ReadOnly: false,
247 TmpfsOptions: &mount.TmpfsOptions{
248 Mode: 0o1777, // world-writable sticky bit
249 Options: [][]string{
250 {"exec"},
251 },
252 },
253 },
254 },
255 ReadonlyRootfs: false,
256 CapDrop: []string{"ALL"},
257 CapAdd: []string{"CAP_DAC_OVERRIDE", "CAP_CHOWN", "CAP_FOWNER", "CAP_SETUID", "CAP_SETGID"},
258 SecurityOpt: []string{"no-new-privileges"},
259 ExtraHosts: []string{"host.docker.internal:host-gateway"},
260 Resources: container.Resources{
261 Memory: e.cfg.NixeryPipelines.MaxJobMemoryMB * 1024 * 1024,
262 },
263 }, nil, nil, "")
264 if err != nil {
265 fmt.Fprintf(
266 wfLogger.DataWriter(setupStepIdx, "stderr"),
267 "container creation failed: %s",
268 err,
269 )
270 return fmt.Errorf("creating container: %w", err)
271 }
272
273 e.registerCleanup(wid, func(ctx context.Context) error {
274 if err := e.docker.ContainerStop(ctx, resp.ID, container.StopOptions{}); err != nil {
275 return fmt.Errorf("stopping container: %w", err)
276 }
277
278 err := e.docker.ContainerRemove(ctx, resp.ID, container.RemoveOptions{
279 RemoveVolumes: true,
280 RemoveLinks: false,
281 Force: false,
282 })
283 if err != nil {
284 return fmt.Errorf("removing container: %w", err)
285 }
286
287 return nil
288 })
289
290 /// -------------------------CONTAINER START----------------------------------------
291 wfLogger.DataWriter(setupStepIdx, "stdout").Write([]byte("starting container..."))
292 if err := e.docker.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil {
293 return fmt.Errorf("starting container: %w", err)
294 }
295
296 mkExecResp, err := e.docker.ContainerExecCreate(ctx, resp.ID, container.ExecOptions{
297 Cmd: []string{"mkdir", "-p", workspaceDir, homeDir},
298 AttachStdout: true, // NOTE(winter): pretty sure this will make it so that when stdout read is done below, mkdir is done. maybe??
299 AttachStderr: true, // for good measure, backed up by docker/cli ("If -d is not set, attach to everything by default")
300 })
301 if err != nil {
302 return err
303 }
304
305 // This actually *starts* the command. Thanks, Docker!
306 execResp, err := e.docker.ContainerExecAttach(ctx, mkExecResp.ID, container.ExecAttachOptions{})
307 if err != nil {
308 return err
309 }
310 defer execResp.Close()
311
312 // This is apparently best way to wait for the command to complete.
313 _, err = io.ReadAll(execResp.Reader)
314 if err != nil {
315 return err
316 }
317
318 /// -----------------------------------FINISH---------------------------------------
319 execInspectResp, err := e.docker.ContainerExecInspect(ctx, mkExecResp.ID)
320 if err != nil {
321 return err
322 }
323
324 if execInspectResp.ExitCode != 0 {
325 return fmt.Errorf("mkdir exited with exit code %d", execInspectResp.ExitCode)
326 } else if execInspectResp.Running {
327 return errors.New("mkdir is somehow still running??")
328 }
329
330 addl.container = resp.ID
331 wf.Data = addl
332
333 return nil
334}
335
336func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
337 addl := w.Data.(addlFields)
338 workflowEnvs := ConstructEnvs(w.Environment)
339 // TODO(winter): should SetupWorkflow also have secret access?
340 // IMO yes, but probably worth thinking on.
341 for _, s := range secrets {
342 workflowEnvs.AddEnv(s.Key, s.Value)
343 }
344
345 step := w.Steps[idx]
346
347 select {
348 case <-ctx.Done():
349 return ctx.Err()
350 default:
351 }
352
353 envs := append(EnvVars(nil), workflowEnvs...)
354 if nixStep, ok := step.(Step); ok {
355 for k, v := range nixStep.environment {
356 envs.AddEnv(k, v)
357 }
358 }
359
360 envs.AddEnv("HOME", homeDir)
361 existingPath := "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
362 envs.AddEnv("PATH", fmt.Sprintf("%s/.nix-profile/bin:/nix/var/nix/profiles/default/bin:%s", homeDir, existingPath))
363
364 mkExecResp, err := e.docker.ContainerExecCreate(ctx, addl.container, container.ExecOptions{
365 Cmd: []string{"bash", "-c", step.Command()},
366 AttachStdout: true,
367 AttachStderr: true,
368 Env: envs,
369 })
370 if err != nil {
371 return fmt.Errorf("creating exec: %w", err)
372 }
373
374 // start tailing logs in background
375 tailDone := make(chan error, 1)
376 go func() {
377 tailDone <- e.tailStep(ctx, wfLogger, mkExecResp.ID, idx)
378 }()
379
380 select {
381 case <-tailDone:
382
383 case <-ctx.Done():
384 // cleanup will be handled by DestroyWorkflow, since
385 // Docker doesn't provide an API to kill an exec run
386 // (sure, we could grab the PID and kill it ourselves,
387 // but that's wasted effort)
388 e.l.Warn("step timed out", "step", step.Name())
389
390 <-tailDone
391
392 return engine.ErrTimedOut
393 }
394
395 select {
396 case <-ctx.Done():
397 return ctx.Err()
398 default:
399 }
400
401 execInspectResp, err := e.docker.ContainerExecInspect(ctx, mkExecResp.ID)
402 if err != nil {
403 return err
404 }
405
406 if execInspectResp.ExitCode != 0 {
407 inspectResp, err := e.docker.ContainerInspect(ctx, addl.container)
408 if err != nil {
409 return err
410 }
411
412 e.l.Error("workflow failed!", "workflow_id", wid.String(), "exit_code", execInspectResp.ExitCode, "oom_killed", inspectResp.State.OOMKilled)
413
414 if inspectResp.State.OOMKilled {
415 return ErrOOMKilled
416 }
417 return engine.ErrWorkflowFailed
418 }
419
420 return nil
421}
422
423func (e *Engine) tailStep(ctx context.Context, wfLogger models.WorkflowLogger, execID string, stepIdx int) error {
424 if wfLogger == nil {
425 return nil
426 }
427
428 // This actually *starts* the command. Thanks, Docker!
429 logs, err := e.docker.ContainerExecAttach(ctx, execID, container.ExecAttachOptions{})
430 if err != nil {
431 return err
432 }
433 defer logs.Close()
434
435 _, err = stdcopy.StdCopy(
436 wfLogger.DataWriter(stepIdx, "stdout"),
437 wfLogger.DataWriter(stepIdx, "stderr"),
438 logs.Reader,
439 )
440 if err != nil && err != io.EOF && !errors.Is(err, context.DeadlineExceeded) {
441 return fmt.Errorf("failed to copy logs: %w", err)
442 }
443
444 return nil
445}
446
447func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
448 fns := e.drainCleanups(wid)
449
450 for _, fn := range fns {
451 if err := fn(ctx); err != nil {
452 e.l.Error("failed to cleanup workflow resource", "workflowId", wid, "error", err)
453 }
454 }
455 return nil
456}
457
458func (e *Engine) registerCleanup(wid models.WorkflowId, fn cleanupFunc) {
459 e.cleanupMu.Lock()
460 defer e.cleanupMu.Unlock()
461
462 key := wid.String()
463 e.cleanup[key] = append(e.cleanup[key], fn)
464}
465
466func (e *Engine) drainCleanups(wid models.WorkflowId) []cleanupFunc {
467 e.cleanupMu.Lock()
468 key := wid.String()
469
470 fns := e.cleanup[key]
471 delete(e.cleanup, key)
472 e.cleanupMu.Unlock()
473
474 return fns
475}
476
477func networkName(wid models.WorkflowId) string {
478 return fmt.Sprintf("workflow-network-%s", wid)
479}