alice.pds.demo.boltless.dev
This repository has no description
1package nixery
2
3import (
4 "bufio"
5 "context"
6 "fmt"
7 "io"
8
9 "github.com/docker/docker/api/types"
10 "github.com/docker/docker/api/types/container"
11 "github.com/docker/docker/pkg/stdcopy"
12
13 "tangled.org/core/spindle/engine"
14 "tangled.org/core/spindle/models"
15 "tangled.org/core/spindle/storage"
16)
17
18func (e *Engine) baseEnv() EnvVars {
19 envs := EnvVars{}
20 envs.AddEnv("HOME", homeDir)
21 envs.AddEnv("PATH", fmt.Sprintf("%s/.nix-profile/bin:/nix/var/nix/profiles/default/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", homeDir))
22 return envs
23}
24
25func (e *Engine) execAttached(ctx context.Context, containerID string, opts container.ExecOptions) (string, types.HijackedResponse, error) {
26 execResp, err := e.docker.ContainerExecCreate(ctx, containerID, opts)
27 if err != nil {
28 return "", types.HijackedResponse{}, fmt.Errorf("create exec: %w", err)
29 }
30 attach, err := e.docker.ContainerExecAttach(ctx, execResp.ID, container.ExecAttachOptions{})
31 if err != nil {
32 return "", types.HijackedResponse{}, fmt.Errorf("attach exec: %w", err)
33 }
34 return execResp.ID, attach, nil
35}
36
37func (e *Engine) containerID(wf *models.Workflow) (string, error) {
38 addl, ok := wf.Data.(addlFields)
39 if !ok || addl.container == "" {
40 return "", fmt.Errorf("nixery workflow has no container")
41 }
42 return addl.container, nil
43}
44
45func (e *Engine) RestoreCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []engine.ResolvedCache, wfLogger models.WorkflowLogger) error {
46 containerID, err := e.containerID(wf)
47 if err != nil {
48 return err
49 }
50
51 out := wfLogger.DataWriter(engine.CacheRestoreStepIdx, "stdout")
52 for _, entry := range caches {
53 if err := ctx.Err(); err != nil {
54 return err
55 }
56 if entry.RestoreKey == "" {
57 fmt.Fprintf(out, "cache %q: miss\n", entry.Key)
58 continue
59 }
60 if entry.RestoreName != "" {
61 fmt.Fprintf(out, "cache %q: restoring from %q\n", entry.Key, entry.RestoreName)
62 }
63
64 rc, err := store.Get(ctx, entry.RestoreKey)
65 if err != nil {
66 fmt.Fprintf(out, "cache %q: fetch failed: %v\n", entry.Key, err)
67 continue
68 }
69
70 br := bufio.NewReader(rc)
71 execID, attach, err := e.execAttached(ctx, containerID, container.ExecOptions{
72 Cmd: []string{"bash", "-c", engine.CacheDecompressCmd(br) + " | tar -x -C /"},
73 Env: e.baseEnv(),
74 AttachStdin: true,
75 AttachStdout: true,
76 AttachStderr: true,
77 })
78 if err != nil {
79 rc.Close()
80 return fmt.Errorf("restore cache %q: %w", entry.Key, err)
81 }
82
83 // drain this now or tar can block on stderr before reading stdin
84 copyDone := make(chan error, 1)
85 go func() {
86 _, err := io.Copy(attach.Conn, br)
87 _ = attach.CloseWrite()
88 copyDone <- err
89 }()
90 _, _ = stdcopy.StdCopy(out, out, attach.Reader)
91 copyErr := <-copyDone
92 rc.Close()
93 attach.Close()
94 if copyErr != nil {
95 return fmt.Errorf("restore cache %q: stream archive: %w", entry.Key, copyErr)
96 }
97
98 inspect, err := e.docker.ContainerExecInspect(ctx, execID)
99 if err != nil {
100 return fmt.Errorf("restore cache %q: %w", entry.Key, err)
101 }
102 if inspect.ExitCode != 0 {
103 fmt.Fprintf(out, "cache %q: extract failed (exit %d)\n", entry.Key, inspect.ExitCode)
104 continue
105 }
106 fmt.Fprintf(out, "cache %q: restored\n", entry.Key)
107 }
108 return nil
109}
110
111func (e *Engine) SaveCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []engine.ResolvedCache, wfLogger models.WorkflowLogger) error {
112 containerID, err := e.containerID(wf)
113 if err != nil {
114 return err
115 }
116
117 out := wfLogger.DataWriter(engine.CacheSaveStepIdx, "stdout")
118 for _, entry := range caches {
119 if err := ctx.Err(); err != nil {
120 return err
121 }
122
123 script := engine.CacheSaveScript(entry.Paths, workspaceDir, entry.CompressionLevel)
124
125 execID, attach, err := e.execAttached(ctx, containerID, container.ExecOptions{
126 Cmd: []string{"bash", "-c", script},
127 Env: e.baseEnv(),
128 AttachStdout: true,
129 AttachStderr: true,
130 })
131 if err != nil {
132 return fmt.Errorf("save cache %q: %w", entry.Key, err)
133 }
134
135 up := engine.NewCacheUpload(ctx, store, entry.SaveKey)
136
137 // StdCopy only returns once the archive is fully written
138 _, copyErr := stdcopy.StdCopy(up.Writer, out, attach.Reader)
139 attach.Close()
140 inspect, inspectErr := e.docker.ContainerExecInspect(ctx, execID)
141
142 switch {
143 case inspectErr == nil && inspect.ExitCode == engine.CacheExitNoPaths:
144 up.Abort(fmt.Errorf("no cache paths"))
145 fmt.Fprintf(out, "cache %q: nothing to save\n", entry.Key)
146 continue
147 case inspectErr == nil && inspect.ExitCode == engine.CacheExitNoCompressor:
148 up.Abort(fmt.Errorf("zstd not available"))
149 fmt.Fprintf(out, "cache %q: zstd not available in image; skipping\n", entry.Key)
150 continue
151 case copyErr != nil:
152 up.Abort(copyErr)
153 return fmt.Errorf("save cache %q: stream archive: %w", entry.Key, copyErr)
154 case inspectErr != nil:
155 up.Abort(inspectErr)
156 return fmt.Errorf("save cache %q: %w", entry.Key, inspectErr)
157 case inspect.ExitCode != 0:
158 up.Abort(fmt.Errorf("exited %d", inspect.ExitCode))
159 return fmt.Errorf("save cache %q: tar exited %d", entry.Key, inspect.ExitCode)
160 }
161 if err := up.Finish(); err != nil {
162 return fmt.Errorf("save cache %q: %w", entry.Key, err)
163 }
164 fmt.Fprintf(out, "cache %q: saved\n", entry.Key)
165 }
166 return nil
167}