alice.pds.demo.boltless.dev
This repository has no description
1package engine
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/hex"
7 "errors"
8 "fmt"
9 "io"
10 "log/slog"
11 "os"
12 "sync"
13 "time"
14
15 "tangled.org/core/notifier"
16 "tangled.org/core/spindle/artifactstore"
17 "tangled.org/core/spindle/config"
18 "tangled.org/core/spindle/db"
19 "tangled.org/core/spindle/models"
20 "tangled.org/core/spindle/secrets"
21)
22
23var (
24 ErrTimedOut = errors.New("timed out")
25 ErrWorkflowFailed = errors.New("workflow failed")
26 ErrWorkflowCanceled = errors.New("workflow canceled")
27)
28
29var (
30 activeMu sync.Mutex
31 activeCancels = make(map[models.WorkflowId]context.CancelCauseFunc)
32)
33
34func CancelWorkflow(wid models.WorkflowId) {
35 activeMu.Lock()
36 cancel, ok := activeCancels[wid]
37 activeMu.Unlock()
38 if ok {
39 cancel(ErrWorkflowCanceled)
40 }
41}
42
43// user cancel, timeout is DeadlineExceeded
44func isCanceled(wfCtx context.Context) bool {
45 return errors.Is(context.Cause(wfCtx), ErrWorkflowCanceled)
46}
47
48// for when recording early wf cancellations
49func writeWfError(db *db.DB, n *notifier.Notifier, l *slog.Logger, wfCtx context.Context, wid models.WorkflowId, phase string, err error) {
50 l = l.With("wid", wid, "phase", phase)
51 switch {
52 case isCanceled(wfCtx):
53 l.Info("workflow canceled")
54 if dbErr := db.StatusCancelled(wid, "User canceled the workflow", -1, n); dbErr != nil {
55 l.Error("failed to set workflow status to cancelled", "err", dbErr)
56 }
57 case errors.Is(err, ErrTimedOut) || errors.Is(wfCtx.Err(), context.DeadlineExceeded):
58 l.Info("workflow timed out")
59 if dbErr := db.StatusTimeout(wid, n); dbErr != nil {
60 l.Error("failed to set workflow status to timeout", "err", dbErr)
61 }
62 default:
63 l.Error("workflow failed", "err", err)
64 if dbErr := db.StatusFailed(wid, err.Error(), -1, n); dbErr != nil {
65 l.Error("failed to set workflow status to failed", "err", dbErr)
66 }
67 }
68}
69
70type workflowFinalizer interface {
71 FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error
72}
73
74func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, stores *artifactstore.Stores, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
75 l.Info("starting all workflows in parallel", "pipeline", pipelineId)
76
77 var allSecrets []secrets.UnlockedSecret
78 // never pass secrets to pipelines that run untrusted (e.g. fork) code
79 if pipeline.TrustedSource && pipeline.RepoDid != "" {
80 if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil {
81 allSecrets = res
82 }
83 } else if !pipeline.TrustedSource {
84 l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId)
85 }
86
87 secretValues := make([]string, len(allSecrets))
88 for i, s := range allSecrets {
89 secretValues[i] = s.Value
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 wfLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues)
132 if err != nil {
133 l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
134 wfLogger = models.NullLogger{}
135 } else {
136 l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
137 defer archiveWorkflowLog(l, stores, db, cfg.Server.LogDir, wid)
138 defer wfLogger.Close()
139 }
140
141 timeoutCtx, timeoutCancel := context.WithTimeout(ctx, workflowTimeout)
142 defer timeoutCancel()
143
144 wfCtx, userCancel := context.WithCancelCause(timeoutCtx)
145 defer userCancel(nil)
146
147 // allow wf context to be cancelled properly by manual cancel
148 activeMu.Lock()
149 activeCancels[wid] = userCancel
150 activeMu.Unlock()
151 defer func() {
152 activeMu.Lock()
153 delete(activeCancels, wid)
154 activeMu.Unlock()
155 }()
156
157 l.Info("waiting for slot", "wid", wid)
158 slot := WorkflowSlot(NoopSlot{})
159 if s, ok := eng.(WorkflowSlotter); ok {
160 slot, err = s.AcquireWorkflowSlot(wfCtx, wid, &w)
161 if err != nil {
162 writeWfError(db, n, l, wfCtx, wid, "waiting for slot", err)
163 return
164 }
165 }
166 defer slot.Release()
167
168 err = db.StatusRunning(wid, n)
169 if err != nil {
170 l.Error("failed to set workflow status to running", "wid", wid, "err", err)
171 return
172 }
173
174 err = eng.SetupWorkflow(wfCtx, wid, &w, wfLogger)
175 if err != nil {
176 if !isCanceled(wfCtx) {
177 if destroyErr := eng.DestroyWorkflow(ctx, wid); destroyErr != nil {
178 l.Error("failed to destroy workflow after setup failure", "error", destroyErr)
179 }
180 }
181 writeWfError(db, n, l, wfCtx, wid, "setting up workflow", err)
182 return
183 }
184 defer eng.DestroyWorkflow(ctx, wid)
185
186 for stepIdx, step := range w.Steps {
187 if wfLogger != nil {
188 wfLogger.
189 ControlWriter(stepIdx, step, models.StepStatusStart).
190 Write([]byte{0})
191 }
192
193 err = eng.RunStep(wfCtx, wid, &w, stepIdx, allSecrets, wfLogger)
194
195 if wfLogger != nil {
196 wfLogger.
197 ControlWriter(stepIdx, step, models.StepStatusEnd).
198 Write([]byte{0})
199 }
200
201 if err != nil {
202 writeWfError(db, n, l, wfCtx, wid, "running step", err)
203 return
204 }
205 }
206
207 if finalizer, ok := eng.(workflowFinalizer); ok {
208 if err := finalizer.FinalizeWorkflow(wfCtx, wid, &w, wfLogger); err != nil {
209 writeWfError(db, n, l, wfCtx, wid, "finalizing", err)
210 return
211 }
212 }
213
214 if isCanceled(wfCtx) {
215 writeWfError(db, n, l, wfCtx, wid, "before success", nil)
216 return
217 }
218
219 err = db.StatusSuccess(wid, n)
220 if err != nil {
221 l.Error("failed to set workflow status to success", "wid", wid, "err", err)
222 }
223 })
224 }
225 }
226
227 wg.Wait()
228 l.Info("all workflows completed")
229}
230
231func archiveWorkflowLog(l *slog.Logger, stores *artifactstore.Stores, database *db.DB, logDir string, wid models.WorkflowId) {
232 if stores == nil {
233 return
234 }
235 logPath := models.LogFilePath(logDir, wid)
236 file, err := os.Open(logPath)
237 if err != nil {
238 l.Error("open workflow log for archival", "wid", wid, "err", err)
239 return
240 }
241 hash := sha256.New()
242 if _, err := io.Copy(hash, file); err != nil {
243 _ = file.Close()
244 l.Error("hash workflow log", "wid", wid, "err", err)
245 return
246 }
247 _ = file.Close()
248
249 ref := wid.String() + ".log"
250 uploadCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
251 defer cancel()
252 errs := stores.PutFile(uploadCtx, ref, logPath)
253 for _, err := range errs {
254 l.Error("archive workflow log", "wid", wid, "err", err)
255 }
256 if len(errs) == len(stores.Names()) {
257 return
258 }
259 digest := "sha256:" + hex.EncodeToString(hash.Sum(nil))
260 if err := database.SaveArtifactRef(wid.String(), wid.Name, ref, digest); err != nil {
261 l.Error("save workflow log artifact", "wid", wid, "err", err)
262 }
263}