alice.pds.demo.boltless.dev
This repository has no description
1package engine
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "log/slog"
8 "path/filepath"
9 "sync"
10
11 "tangled.org/core/notifier"
12 "tangled.org/core/spindle/config"
13 "tangled.org/core/spindle/db"
14 "tangled.org/core/spindle/models"
15 "tangled.org/core/spindle/secrets"
16)
17
18var (
19 ErrTimedOut = errors.New("timed out")
20 ErrWorkflowFailed = errors.New("workflow failed")
21 ErrWorkflowCanceled = errors.New("workflow canceled")
22)
23
24var (
25 activeMu sync.Mutex
26 activeCancels = make(map[models.WorkflowId]context.CancelCauseFunc)
27)
28
29func CancelWorkflow(wid models.WorkflowId) {
30 activeMu.Lock()
31 cancel, ok := activeCancels[wid]
32 activeMu.Unlock()
33 if ok {
34 cancel(ErrWorkflowCanceled)
35 }
36}
37
38// user cancel, timeout is DeadlineExceeded
39func isCanceled(wfCtx context.Context) bool {
40 return errors.Is(context.Cause(wfCtx), ErrWorkflowCanceled)
41}
42
43// for when recording early wf cancellations
44func writeWfError(db *db.DB, n *notifier.Notifier, l *slog.Logger, wfCtx context.Context, wid models.WorkflowId, phase string, err error) {
45 l = l.With("wid", wid, "phase", phase)
46 switch {
47 case isCanceled(wfCtx):
48 l.Info("workflow canceled")
49 if dbErr := db.StatusCancelled(wid, "User canceled the workflow", -1, n); dbErr != nil {
50 l.Error("failed to set workflow status to cancelled", "err", dbErr)
51 }
52 case errors.Is(err, ErrTimedOut) || errors.Is(wfCtx.Err(), context.DeadlineExceeded):
53 l.Info("workflow timed out")
54 if dbErr := db.StatusTimeout(wid, n); dbErr != nil {
55 l.Error("failed to set workflow status to timeout", "err", dbErr)
56 }
57 default:
58 l.Error("workflow failed", "err", err)
59 if dbErr := db.StatusFailed(wid, err.Error(), -1, n); dbErr != nil {
60 l.Error("failed to set workflow status to failed", "err", dbErr)
61 }
62 }
63}
64
65type workflowFinalizer interface {
66 FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error
67}
68
69func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
70 l.Info("starting all workflows in parallel", "pipeline", pipelineId)
71
72 var allSecrets []secrets.UnlockedSecret
73 // never pass secrets to pipelines that run untrusted (e.g. fork) code
74 if pipeline.TrustedSource && pipeline.RepoDid != "" {
75 if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil {
76 allSecrets = res
77 }
78 } else if !pipeline.TrustedSource {
79 l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId)
80 }
81
82 secretValues := make([]string, len(allSecrets))
83 for i, s := range allSecrets {
84 secretValues[i] = s.Value
85 }
86
87 s3, err := NewS3(cfg.S3.LogBucket)
88 if err != nil {
89 l.Error("error creating s3 client", "err", err)
90 }
91
92 // wid.String() is lossy so two different names can map to the same key
93 // eg. "foo bar" and "foo-bar"...
94 wfCounts := make(map[string]int)
95 for _, wfs := range pipeline.Workflows {
96 for _, w := range wfs {
97 wid := models.WorkflowId{
98 PipelineId: pipelineId,
99 Name: w.Name,
100 }
101 wfCounts[wid.String()]++
102 }
103 }
104
105 var wg sync.WaitGroup
106 for eng, wfs := range pipeline.Workflows {
107 workflowTimeout := eng.WorkflowTimeout()
108 l.Info("using workflow timeout", "timeout", workflowTimeout)
109
110 for _, w := range wfs {
111 w := w
112 wid := models.WorkflowId{
113 PipelineId: pipelineId,
114 Name: w.Name,
115 }
116
117 if wfCounts[wid.String()] > 1 {
118 l.Warn("skipping workflow due to name collision", "wid", wid, "key", wid.String())
119 dbErr := db.StatusFailed(wid, fmt.Sprintf("colliding workflow name: %s; rename to something else", wid.String()), -1, n)
120 if dbErr != nil {
121 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
122 }
123 continue
124 }
125
126 wg.Go(func() {
127 if st, err := db.GetStatus(wid); err == nil && models.StatusKind(st.Status).IsFinish() {
128 l.Info("skipping finished workflow", "wid", wid, "status", st.Status)
129 return
130 }
131 defer func() {
132 if s3 != nil {
133 logFile := filepath.Join(cfg.Server.LogDir, fmt.Sprintf("%s.log", wid.String()))
134 if err := s3.WriteFile(ctx, logFile); err != nil {
135 l.Error("error uploading logs", "err", err)
136 }
137 }
138 }()
139
140 wfLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues)
141 if err != nil {
142 l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
143 wfLogger = models.NullLogger{}
144 } else {
145 l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
146 defer wfLogger.Close()
147 }
148
149 timeoutCtx, timeoutCancel := context.WithTimeout(ctx, workflowTimeout)
150 defer timeoutCancel()
151
152 wfCtx, userCancel := context.WithCancelCause(timeoutCtx)
153 defer userCancel(nil)
154
155 // allow wf context to be cancelled properly by manual cancel
156 activeMu.Lock()
157 activeCancels[wid] = userCancel
158 activeMu.Unlock()
159 defer func() {
160 activeMu.Lock()
161 delete(activeCancels, wid)
162 activeMu.Unlock()
163 }()
164
165 l.Info("waiting for slot", "wid", wid)
166 slot := WorkflowSlot(NoopSlot{})
167 if s, ok := eng.(WorkflowSlotter); ok {
168 slot, err = s.AcquireWorkflowSlot(wfCtx, wid, &w)
169 if err != nil {
170 writeWfError(db, n, l, wfCtx, wid, "waiting for slot", err)
171 return
172 }
173 }
174 defer slot.Release()
175
176 err = db.StatusRunning(wid, n)
177 if err != nil {
178 l.Error("failed to set workflow status to running", "wid", wid, "err", err)
179 return
180 }
181
182 err = eng.SetupWorkflow(wfCtx, wid, &w, wfLogger)
183 if err != nil {
184 if !isCanceled(wfCtx) {
185 if destroyErr := eng.DestroyWorkflow(ctx, wid); destroyErr != nil {
186 l.Error("failed to destroy workflow after setup failure", "error", destroyErr)
187 }
188 }
189 writeWfError(db, n, l, wfCtx, wid, "setting up workflow", err)
190 return
191 }
192 defer eng.DestroyWorkflow(ctx, wid)
193
194 for stepIdx, step := range w.Steps {
195 if wfLogger != nil {
196 wfLogger.
197 ControlWriter(stepIdx, step, models.StepStatusStart).
198 Write([]byte{0})
199 }
200
201 err = eng.RunStep(wfCtx, wid, &w, stepIdx, allSecrets, wfLogger)
202
203 if wfLogger != nil {
204 wfLogger.
205 ControlWriter(stepIdx, step, models.StepStatusEnd).
206 Write([]byte{0})
207 }
208
209 if err != nil {
210 writeWfError(db, n, l, wfCtx, wid, "running step", err)
211 return
212 }
213 }
214
215 if finalizer, ok := eng.(workflowFinalizer); ok {
216 if err := finalizer.FinalizeWorkflow(wfCtx, wid, &w, wfLogger); err != nil {
217 writeWfError(db, n, l, wfCtx, wid, "finalizing", err)
218 return
219 }
220 }
221
222 if isCanceled(wfCtx) {
223 writeWfError(db, n, l, wfCtx, wid, "before success", nil)
224 return
225 }
226
227 err = db.StatusSuccess(wid, n)
228 if err != nil {
229 l.Error("failed to set workflow status to success", "wid", wid, "err", err)
230 }
231 })
232 }
233 }
234
235 wg.Wait()
236 l.Info("all workflows completed")
237}