lewis.moe
Monorepo for Tangled
spindle/models: stream nixery image pull errors to clients
Signed-off-by: oppiliappan <me@oppi.li>
authored by
oppiliappan
and committed by
+16
-14
spindle/engine/engine.go
+16
-14
spindle/engine/engine.go
···
30
}
31
}
32
33
var wg sync.WaitGroup
34
for eng, wfs := range pipeline.Workflows {
35
workflowTimeout := eng.WorkflowTimeout()
···
45
Name: w.Name,
46
}
47
48
-
err := db.StatusRunning(wid, n)
49
if err != nil {
50
l.Error("failed to set workflow status to running", "wid", wid, "err", err)
51
return
52
}
53
54
-
err = eng.SetupWorkflow(ctx, wid, &w)
55
if err != nil {
56
// TODO(winter): Should this always set StatusFailed?
57
// In the original, we only do in a subset of cases.
···
69
return
70
}
71
defer eng.DestroyWorkflow(ctx, wid)
72
-
73
-
secretValues := make([]string, len(allSecrets))
74
-
for i, s := range allSecrets {
75
-
secretValues[i] = s.Value
76
-
}
77
-
wfLogger, err := models.NewWorkflowLogger(cfg.Server.LogDir, wid, secretValues)
78
-
if err != nil {
79
-
l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
80
-
wfLogger = nil
81
-
} else {
82
-
defer wfLogger.Close()
83
-
}
84
85
ctx, cancel := context.WithTimeout(ctx, workflowTimeout)
86
defer cancel()
···
30
}
31
}
32
33
+
secretValues := make([]string, len(allSecrets))
34
+
for i, s := range allSecrets {
35
+
secretValues[i] = s.Value
36
+
}
37
+
38
var wg sync.WaitGroup
39
for eng, wfs := range pipeline.Workflows {
40
workflowTimeout := eng.WorkflowTimeout()
···
50
Name: w.Name,
51
}
52
53
+
wfLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues)
54
+
if err != nil {
55
+
l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
56
+
wfLogger = models.NullLogger{}
57
+
} else {
58
+
l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
59
+
defer wfLogger.Close()
60
+
}
61
+
62
+
err = db.StatusRunning(wid, n)
63
if err != nil {
64
l.Error("failed to set workflow status to running", "wid", wid, "err", err)
65
return
66
}
67
68
+
err = eng.SetupWorkflow(ctx, wid, &w, wfLogger)
69
if err != nil {
70
// TODO(winter): Should this always set StatusFailed?
71
// In the original, we only do in a subset of cases.
···
83
return
84
}
85
defer eng.DestroyWorkflow(ctx, wid)
86
87
ctx, cancel := context.WithTimeout(ctx, workflowTimeout)
88
defer cancel()
+49
-9
spindle/engines/nixery/engine.go
+49
-9
spindle/engines/nixery/engine.go
···
1
package nixery
2
3
import (
4
"context"
5
"errors"
6
"fmt"
7
"io"
8
"log/slog"
9
-
"os"
10
"path"
11
"runtime"
12
"sync"
···
169
return e, nil
170
}
171
172
-
func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) error {
173
-
e.l.Info("setting up workflow", "workflow", wid)
174
175
_, err := e.docker.NetworkCreate(ctx, networkName(wid), network.CreateOptions{
176
Driver: "bridge",
177
})
178
if err != nil {
179
return err
180
}
181
e.registerCleanup(wid, func(ctx context.Context) error {
182
if err := e.docker.NetworkRemove(ctx, networkName(wid)); err != nil {
183
return fmt.Errorf("removing network: %w", err)
···
185
return nil
186
})
187
188
addl := wf.Data.(addlFields)
189
190
reader, err := e.docker.ImagePull(ctx, addl.image, image.PullOptions{})
191
if err != nil {
192
-
e.l.Error("pipeline image pull failed!", "image", addl.image, "workflowId", wid, "error", err.Error())
193
-
194
return fmt.Errorf("pulling image: %w", err)
195
}
196
defer reader.Close()
197
-
io.Copy(os.Stdout, reader)
198
199
resp, err := e.docker.ContainerCreate(ctx, &container.Config{
200
Image: addl.image,
···
229
ExtraHosts: []string{"host.docker.internal:host-gateway"},
230
}, nil, nil, "")
231
if err != nil {
232
return fmt.Errorf("creating container: %w", err)
233
}
234
e.registerCleanup(wid, func(ctx context.Context) error {
235
if err := e.docker.ContainerStop(ctx, resp.ID, container.StopOptions{}); err != nil {
236
return fmt.Errorf("stopping container: %w", err)
···
244
if err != nil {
245
return fmt.Errorf("removing container: %w", err)
246
}
247
return nil
248
})
249
250
if err := e.docker.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil {
251
return fmt.Errorf("starting container: %w", err)
252
}
···
273
return err
274
}
275
276
execInspectResp, err := e.docker.ContainerExecInspect(ctx, mkExecResp.ID)
277
if err != nil {
278
return err
···
290
return nil
291
}
292
293
-
func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger *models.WorkflowLogger) error {
294
addl := w.Data.(addlFields)
295
workflowEnvs := ConstructEnvs(w.Environment)
296
// TODO(winter): should SetupWorkflow also have secret access?
···
331
// start tailing logs in background
332
tailDone := make(chan error, 1)
333
go func() {
334
-
tailDone <- e.tailStep(ctx, wfLogger, mkExecResp.ID, wid, idx, step)
335
}()
336
337
select {
···
377
return nil
378
}
379
380
-
func (e *Engine) tailStep(ctx context.Context, wfLogger *models.WorkflowLogger, execID string, wid models.WorkflowId, stepIdx int, step models.Step) error {
381
if wfLogger == nil {
382
return nil
383
}
···
1
package nixery
2
3
import (
4
+
"bufio"
5
"context"
6
"errors"
7
"fmt"
8
"io"
9
"log/slog"
10
"path"
11
"runtime"
12
"sync"
···
169
return e, nil
170
}
171
172
+
func (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)
···
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,
···
259
ExtraHosts: []string{"host.docker.internal:host-gateway"},
260
}, nil, nil, "")
261
if err != nil {
262
+
fmt.Fprintf(
263
+
wfLogger.DataWriter(setupStepIdx, "stderr"),
264
+
"container creation failed: %s",
265
+
err,
266
+
)
267
return fmt.Errorf("creating container: %w", err)
268
}
269
+
270
e.registerCleanup(wid, func(ctx context.Context) error {
271
if err := e.docker.ContainerStop(ctx, resp.ID, container.StopOptions{}); err != nil {
272
return fmt.Errorf("stopping container: %w", err)
···
280
if err != nil {
281
return fmt.Errorf("removing container: %w", err)
282
}
283
+
284
return nil
285
})
286
287
+
/// -------------------------CONTAINER START----------------------------------------
288
+
wfLogger.DataWriter(setupStepIdx, "stdout").Write([]byte("starting container..."))
289
if err := e.docker.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil {
290
return fmt.Errorf("starting container: %w", err)
291
}
···
312
return err
313
}
314
315
+
/// -----------------------------------FINISH---------------------------------------
316
execInspectResp, err := e.docker.ContainerExecInspect(ctx, mkExecResp.ID)
317
if err != nil {
318
return err
···
330
return nil
331
}
332
333
+
func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
334
addl := w.Data.(addlFields)
335
workflowEnvs := ConstructEnvs(w.Environment)
336
// TODO(winter): should SetupWorkflow also have secret access?
···
371
// start tailing logs in background
372
tailDone := make(chan error, 1)
373
go func() {
374
+
tailDone <- e.tailStep(ctx, wfLogger, mkExecResp.ID, idx)
375
}()
376
377
select {
···
417
return nil
418
}
419
420
+
func (e *Engine) tailStep(ctx context.Context, wfLogger models.WorkflowLogger, execID string, stepIdx int) error {
421
if wfLogger == nil {
422
return nil
423
}
+2
-2
spindle/models/engine.go
+2
-2
spindle/models/engine.go
···
10
11
type Engine interface {
12
InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*Workflow, error)
13
-
SetupWorkflow(ctx context.Context, wid WorkflowId, wf *Workflow) error
14
WorkflowTimeout() time.Duration
15
DestroyWorkflow(ctx context.Context, wid WorkflowId) error
16
-
RunStep(ctx context.Context, wid WorkflowId, w *Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger *WorkflowLogger) error
17
}
···
10
11
type Engine interface {
12
InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*Workflow, error)
13
+
SetupWorkflow(ctx context.Context, wid WorkflowId, wf *Workflow, wfLogger WorkflowLogger) error
14
WorkflowTimeout() time.Duration
15
DestroyWorkflow(ctx context.Context, wid WorkflowId) error
16
+
RunStep(ctx context.Context, wid WorkflowId, w *Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger WorkflowLogger) error
17
}