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 "tangled.org/core/spindle/storage"
17)
18
19var (
20 ErrTimedOut = errors.New("timed out")
21 ErrWorkflowFailed = errors.New("workflow failed")
22 ErrWorkflowCanceled = errors.New("workflow canceled")
23)
24
25var (
26 activeMu sync.Mutex
27 activeCancels = make(map[models.WorkflowId]context.CancelCauseFunc)
28)
29
30func CancelWorkflow(wid models.WorkflowId) {
31 activeMu.Lock()
32 cancel, ok := activeCancels[wid]
33 activeMu.Unlock()
34 if ok {
35 cancel(ErrWorkflowCanceled)
36 }
37}
38
39// user cancel, timeout is DeadlineExceeded
40func isCanceled(wfCtx context.Context) bool {
41 return errors.Is(context.Cause(wfCtx), ErrWorkflowCanceled)
42}
43
44// for when recording early wf cancellations
45func writeWfError(db *db.DB, n *notifier.Notifier, l *slog.Logger, wfCtx context.Context, wid models.WorkflowId, phase string, err error) {
46 l = l.With("wid", wid, "phase", phase)
47 switch {
48 case isCanceled(wfCtx):
49 l.Info("workflow canceled")
50 if dbErr := db.StatusCancelled(wid, "User canceled the workflow", -1, n); dbErr != nil {
51 l.Error("failed to set workflow status to cancelled", "err", dbErr)
52 }
53 case errors.Is(err, ErrTimedOut) || errors.Is(wfCtx.Err(), context.DeadlineExceeded):
54 l.Info("workflow timed out")
55 if dbErr := db.StatusTimeout(wid, n); dbErr != nil {
56 l.Error("failed to set workflow status to timeout", "err", dbErr)
57 }
58 default:
59 l.Error("workflow failed", "err", err)
60 if dbErr := db.StatusFailed(wid, err.Error(), -1, n); dbErr != nil {
61 l.Error("failed to set workflow status to failed", "err", dbErr)
62 }
63 }
64}
65
66type workflowFinalizer interface {
67 FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error
68}
69
70func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, cacheStore storage.Storage, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
71 l.Info("starting all workflows in parallel", "pipeline", pipelineId)
72
73 isTrustedRepo := pipeline.TrustedSource && pipeline.RepoDid != ""
74 var allSecrets []secrets.UnlockedSecret
75 // never pass secrets to pipelines that run untrusted (e.g. fork) code
76 if isTrustedRepo {
77 if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil {
78 allSecrets = res
79 }
80 } else if !pipeline.TrustedSource {
81 l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId)
82 }
83 // untrusted runs cant read or write shared caches
84 cacheOwnerDID := ""
85 cacheEnabled := cacheStore != nil && isTrustedRepo
86 if cacheEnabled {
87 repo, err := db.GetRepoByDid(pipeline.RepoDid)
88 if err != nil {
89 l.Warn("cache owner lookup failed; caching disabled", "repo", pipeline.RepoDid, "err", err)
90 cacheEnabled = false
91 } else {
92 cacheOwnerDID = repo.Owner.String()
93 }
94 } else if cacheStore != nil && !pipeline.TrustedSource {
95 l.Info("skipping caches for untrusted pipeline source", "pipeline", pipelineId)
96 }
97
98 // hash the checked commit, not the working tree
99 cacheRepoPath, cacheRev := "", ""
100 if tm := pipeline.TriggerMetadata; tm != nil {
101 if rev, err := models.ExtractCommitSHA(*tm); err == nil {
102 did := pipeline.RepoDid.String()
103 if tm.SourceRepo != nil && *tm.SourceRepo != "" {
104 did = *tm.SourceRepo
105 }
106 cacheRepoPath, cacheRev = filepath.Join(cfg.Server.RepoDir, did), rev
107 } else {
108 l.Warn("cannot resolve pipeline commit; cache hashing disabled", "err", err)
109 }
110 }
111
112 secretValues := make([]string, len(allSecrets))
113 for i, s := range allSecrets {
114 secretValues[i] = s.Value
115 }
116
117 s3, err := NewS3(cfg.S3.LogBucket)
118 if err != nil {
119 l.Error("error creating s3 client", "err", err)
120 }
121
122 // wid.String() is lossy so two different names can map to the same key
123 // eg. "foo bar" and "foo-bar"...
124 wfCounts := make(map[string]int)
125 for _, wfs := range pipeline.Workflows {
126 for _, w := range wfs {
127 wid := models.WorkflowId{
128 PipelineId: pipelineId,
129 Name: w.Name,
130 }
131 wfCounts[wid.String()]++
132 }
133 }
134
135 var wg sync.WaitGroup
136 for eng, wfs := range pipeline.Workflows {
137 workflowTimeout := eng.WorkflowTimeout()
138 cacheRunner, cachesSupported := eng.(CacheRunner)
139 l.Info("using workflow timeout", "timeout", workflowTimeout)
140
141 for _, w := range wfs {
142 w := w
143 wid := models.WorkflowId{
144 PipelineId: pipelineId,
145 Name: w.Name,
146 }
147
148 if wfCounts[wid.String()] > 1 {
149 l.Warn("skipping workflow due to name collision", "wid", wid, "key", wid.String())
150 dbErr := db.StatusFailed(wid, fmt.Sprintf("colliding workflow name: %s; rename to something else", wid.String()), -1, n)
151 if dbErr != nil {
152 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
153 }
154 continue
155 }
156
157 wg.Go(func() {
158 if st, err := db.GetStatus(wid); err == nil && models.StatusKind(st.Status).IsFinish() {
159 l.Info("skipping finished workflow", "wid", wid, "status", st.Status)
160 return
161 }
162 defer func() {
163 if s3 != nil {
164 logFile := filepath.Join(cfg.Server.LogDir, fmt.Sprintf("%s.log", wid.String()))
165 if err := s3.WriteFile(ctx, logFile); err != nil {
166 l.Error("error uploading logs", "err", err)
167 }
168 }
169 }()
170
171 wfLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues)
172 if err != nil {
173 l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
174 wfLogger = models.NullLogger{}
175 } else {
176 l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
177 defer wfLogger.Close()
178 }
179
180 timeoutCtx, timeoutCancel := context.WithTimeout(ctx, workflowTimeout)
181 defer timeoutCancel()
182
183 wfCtx, userCancel := context.WithCancelCause(timeoutCtx)
184 defer userCancel(nil)
185
186 // allow wf context to be cancelled properly by manual cancel
187 activeMu.Lock()
188 activeCancels[wid] = userCancel
189 activeMu.Unlock()
190 defer func() {
191 activeMu.Lock()
192 delete(activeCancels, wid)
193 activeMu.Unlock()
194 }()
195
196 l.Info("waiting for slot", "wid", wid)
197 slot := WorkflowSlot(NoopSlot{})
198 if s, ok := eng.(WorkflowSlotter); ok {
199 slot, err = s.AcquireWorkflowSlot(wfCtx, wid, &w)
200 if err != nil {
201 writeWfError(db, n, l, wfCtx, wid, "waiting for slot", err)
202 return
203 }
204 }
205 defer slot.Release()
206
207 err = db.StatusRunning(wid, n)
208 if err != nil {
209 l.Error("failed to set workflow status to running", "wid", wid, "err", err)
210 return
211 }
212
213 err = eng.SetupWorkflow(wfCtx, wid, &w, wfLogger)
214 if err != nil {
215 if !isCanceled(wfCtx) {
216 if destroyErr := eng.DestroyWorkflow(ctx, wid); destroyErr != nil {
217 l.Error("failed to destroy workflow after setup failure", "error", destroyErr)
218 }
219 }
220 writeWfError(db, n, l, wfCtx, wid, "setting up workflow", err)
221 return
222 }
223 defer eng.DestroyWorkflow(ctx, wid)
224
225 ctx, cancel := context.WithTimeout(ctx, workflowTimeout)
226 defer cancel()
227
228 var resolvedCaches []ResolvedCache
229 if cacheEnabled && len(w.Caches) > 0 {
230 if cachesSupported {
231 resolvedCaches = ResolveCaches(ctx, l, db, pipeline.RepoDid.String(), w.Engine, cacheRepoPath, cacheRev, w.Caches)
232 wfLogger.ControlWriter(CacheRestoreStepIdx, CacheRestoreStep, models.StepStatusStart).Write([]byte{0})
233 // caches are an optimization, never a reason to fail the workflow
234 restoreStore := cacheStoreForRestore(cacheStore, db, l, resolvedCaches)
235 if err := cacheRunner.RestoreCache(ctx, wid, &w, restoreStore, resolvedCaches, wfLogger); err != nil {
236 l.Warn("cache restore failed", "wid", wid, "err", err)
237 }
238 wfLogger.ControlWriter(CacheRestoreStepIdx, CacheRestoreStep, models.StepStatusEnd).Write([]byte{0})
239 } else {
240 l.Warn("engine does not support caches, skipping restore", "wid", wid)
241 }
242 }
243
244 // dont save on timeouts, their context is already dead
245 saveCaches := func(failed bool) {
246 toSave := resolvedCaches[:0]
247 for _, rc := range resolvedCaches {
248 if rc.saveOn(failed) {
249 toSave = append(toSave, rc)
250 }
251 }
252 if len(toSave) == 0 {
253 return
254 }
255 wfLogger.ControlWriter(CacheSaveStepIdx, CacheSaveStep, models.StepStatusStart).Write([]byte{0})
256 saveStore, err := prepareCacheSaves(ctx, cacheStore, db, l, cacheOwnerDID, pipeline.RepoDid.String(), w.Engine, toSave)
257 if err != nil {
258 l.Warn("cache metadata setup failed", "wid", wid, "err", err)
259 } else {
260 if err := cacheRunner.SaveCache(ctx, wid, &w, saveStore, toSave, wfLogger); err != nil {
261 l.Warn("cache save failed", "wid", wid, "err", err)
262 }
263 saveStore.cleanup(context.WithoutCancel(ctx))
264 }
265 wfLogger.ControlWriter(CacheSaveStepIdx, CacheSaveStep, models.StepStatusEnd).Write([]byte{0})
266 }
267 for stepIdx, step := range w.Steps {
268 if wfLogger != nil {
269 wfLogger.
270 ControlWriter(stepIdx, step, models.StepStatusStart).
271 Write([]byte{0})
272 }
273
274 err = eng.RunStep(wfCtx, wid, &w, stepIdx, allSecrets, wfLogger)
275
276 if wfLogger != nil {
277 wfLogger.
278 ControlWriter(stepIdx, step, models.StepStatusEnd).
279 Write([]byte{0})
280 }
281
282 if err != nil {
283 if !errors.Is(err, ErrTimedOut) && !errors.Is(wfCtx.Err(), context.DeadlineExceeded) && !isCanceled(wfCtx) {
284 saveCaches(true)
285 }
286 writeWfError(db, n, l, wfCtx, wid, "running step", err)
287 return
288 }
289 }
290
291 saveCaches(false)
292
293 if finalizer, ok := eng.(workflowFinalizer); ok {
294 if err := finalizer.FinalizeWorkflow(wfCtx, wid, &w, wfLogger); err != nil {
295 writeWfError(db, n, l, wfCtx, wid, "finalizing", err)
296 return
297 }
298 }
299
300 if isCanceled(wfCtx) {
301 writeWfError(db, n, l, wfCtx, wid, "before success", nil)
302 return
303 }
304
305 err = db.StatusSuccess(wid, n)
306 if err != nil {
307 l.Error("failed to set workflow status to success", "wid", wid, "err", err)
308 }
309 })
310 }
311 }
312
313 wg.Wait()
314 l.Info("all workflows completed")
315}