boltless.me
Signed-off-by: Seongmin Lee git@boltless.me
+2
-18
spindle/db/events.go
+2
-18
spindle/db/events.go
···
18
EventJson string `json:"event"`
19
}
20
21
-
func (d *DB) InsertEvent(event Event, notifier *notifier.Notifier) error {
22
_, err := d.Exec(
23
`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`,
24
event.Rkey,
···
70
return evts, nil
71
}
72
73
-
func (d *DB) CreateStatusEvent(rkey string, s tangled.PipelineStatus, n *notifier.Notifier) error {
74
-
eventJson, err := json.Marshal(s)
75
-
if err != nil {
76
-
return err
77
-
}
78
-
79
-
event := Event{
80
-
Rkey: rkey,
81
-
Nsid: tangled.PipelineStatusNSID,
82
-
Created: time.Now().UnixNano(),
83
-
EventJson: string(eventJson),
84
-
}
85
-
86
-
return d.InsertEvent(event, n)
87
-
}
88
-
89
func (d *DB) createStatusEvent(
90
workflowId models.WorkflowId,
91
statusKind models.StatusKind,
···
116
EventJson: string(eventJson),
117
}
118
119
-
return d.InsertEvent(event, n)
120
121
}
122
···
18
EventJson string `json:"event"`
19
}
20
21
+
func (d *DB) insertEvent(event Event, notifier *notifier.Notifier) error {
22
_, err := d.Exec(
23
`insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`,
24
event.Rkey,
···
70
return evts, nil
71
}
72
73
func (d *DB) createStatusEvent(
74
workflowId models.WorkflowId,
75
statusKind models.StatusKind,
···
100
EventJson: string(eventJson),
101
}
102
103
+
return d.insertEvent(event, n)
104
105
}
106
+24
-10
spindle/engines/nixery/engine.go
+24
-10
spindle/engines/nixery/engine.go
···
179
return err
180
}
181
e.registerCleanup(wid, func(ctx context.Context) error {
182
-
return e.docker.NetworkRemove(ctx, networkName(wid))
183
})
184
185
addl := wf.Data.(addlFields)
···
229
return fmt.Errorf("creating container: %w", err)
230
}
231
e.registerCleanup(wid, func(ctx context.Context) error {
232
-
err = e.docker.ContainerStop(ctx, resp.ID, container.StopOptions{})
233
if err != nil {
234
-
return err
235
}
236
237
-
return e.docker.ContainerRemove(ctx, resp.ID, container.RemoveOptions{
238
RemoveVolumes: true,
239
RemoveLinks: false,
240
Force: false,
241
})
242
})
243
244
err = e.docker.ContainerStart(ctx, resp.ID, container.StartOptions{})
···
394
}
395
396
func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
397
-
e.cleanupMu.Lock()
398
-
key := wid.String()
399
-
400
-
fns := e.cleanup[key]
401
-
delete(e.cleanup, key)
402
-
e.cleanupMu.Unlock()
403
404
for _, fn := range fns {
405
if err := fn(ctx); err != nil {
···
417
e.cleanup[key] = append(e.cleanup[key], fn)
418
}
419
420
func networkName(wid models.WorkflowId) string {
421
return fmt.Sprintf("workflow-network-%s", wid)
422
}
···
179
return err
180
}
181
e.registerCleanup(wid, func(ctx context.Context) error {
182
+
err := e.docker.NetworkRemove(ctx, networkName(wid))
183
+
if err != nil {
184
+
return fmt.Errorf("removing network: %w", err)
185
+
}
186
+
return nil
187
})
188
189
addl := wf.Data.(addlFields)
···
233
return fmt.Errorf("creating container: %w", err)
234
}
235
e.registerCleanup(wid, func(ctx context.Context) error {
236
+
err := e.docker.ContainerStop(ctx, resp.ID, container.StopOptions{})
237
if err != nil {
238
+
return fmt.Errorf("stopping container: %w", err)
239
}
240
241
+
err = e.docker.ContainerRemove(ctx, resp.ID, container.RemoveOptions{
242
RemoveVolumes: true,
243
RemoveLinks: false,
244
Force: false,
245
})
246
+
if err != nil {
247
+
return fmt.Errorf("removing container: %w", err)
248
+
}
249
+
return nil
250
})
251
252
err = e.docker.ContainerStart(ctx, resp.ID, container.StartOptions{})
···
402
}
403
404
func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
405
+
fns := e.drainCleanups(wid)
406
407
for _, fn := range fns {
408
if err := fn(ctx); err != nil {
···
420
e.cleanup[key] = append(e.cleanup[key], fn)
421
}
422
423
+
func (e *Engine) drainCleanups(wid models.WorkflowId) []cleanupFunc {
424
+
e.cleanupMu.Lock()
425
+
key := wid.String()
426
+
427
+
fns := e.cleanup[key]
428
+
delete(e.cleanup, key)
429
+
e.cleanupMu.Unlock()
430
+
431
+
return fns
432
+
}
433
+
434
func networkName(wid models.WorkflowId) string {
435
return fmt.Sprintf("workflow-network-%s", wid)
436
}
+4
-4
spindle/server.go
+4
-4
spindle/server.go
···
302
tpl.TriggerMetadata.Repo.Repo,
303
)
304
if err != nil {
305
-
return err
306
}
307
308
pipelineId := models.PipelineId{
···
323
Name: w.Name,
324
}, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n)
325
if err != nil {
326
-
return err
327
}
328
329
continue
···
337
338
ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl)
339
if err != nil {
340
-
return err
341
}
342
343
// inject TANGLED_* env vars after InitWorkflow
···
354
Name: w.Name,
355
}, s.n)
356
if err != nil {
357
-
return err
358
}
359
}
360
}
···
302
tpl.TriggerMetadata.Repo.Repo,
303
)
304
if err != nil {
305
+
return fmt.Errorf("failed to get repo: %w", err)
306
}
307
308
pipelineId := models.PipelineId{
···
323
Name: w.Name,
324
}, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n)
325
if err != nil {
326
+
return fmt.Errorf("db.StatusFailed: %w", err)
327
}
328
329
continue
···
337
338
ewf, err := s.engs[w.Engine].InitWorkflow(*w, tpl)
339
if err != nil {
340
+
return fmt.Errorf("init workflow: %w", err)
341
}
342
343
// inject TANGLED_* env vars after InitWorkflow
···
354
Name: w.Name,
355
}, s.n)
356
if err != nil {
357
+
return fmt.Errorf("db.StatusPending: %w", err)
358
}
359
}
360
}