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
70// the mill streams the executor's real log file into place itself
71// a local logger here would only write competing lines
72type workflowLoggerProvider interface {
73 WorkflowLogger(wid models.WorkflowId) models.WorkflowLogger
74}
75
76// for engines that manage status updates outside StartWorkflows
77type RemoteStatusEngine interface {
78 AuthorsRemoteStatus()
79}
80
81func reportWorkflowStatusError(l *slog.Logger, database *db.DB, n *notifier.Notifier, wid models.WorkflowId, err error) {
82 if errors.Is(err, ErrTimedOut) {
83 dbErr := database.StatusTimeout(wid, n)
84 if dbErr != nil {
85 l.Error("failed to set workflow status to timeout", "wid", wid, "err", dbErr)
86 }
87 } else if errors.Is(err, ErrWorkflowCanceled) {
88 dbErr := database.StatusCancelled(wid, err.Error(), -1, n)
89 if dbErr != nil {
90 l.Error("failed to set workflow status to cancelled", "wid", wid, "err", dbErr)
91 }
92 } else {
93 dbErr := database.StatusFailed(wid, err.Error(), -1, n)
94 if dbErr != nil {
95 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
96 }
97 }
98}
99
100func 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) {
101 l.Info("starting all workflows in parallel", "pipeline", pipelineId)
102
103 var allSecrets []secrets.UnlockedSecret
104 // never pass secrets to pipelines that run untrusted (e.g. fork) code
105 if pipeline.TrustedSource && pipeline.RepoDid != "" {
106 if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil {
107 allSecrets = res
108 }
109 } else if !pipeline.TrustedSource {
110 l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId)
111 }
112
113 secretValues := make([]string, len(allSecrets))
114 for i, s := range allSecrets {
115 secretValues[i] = s.Value
116 }
117
118 // wid.String() is lossy so two different names can map to the same key
119 // eg. "foo bar" and "foo-bar"...
120 wfCounts := make(map[string]int)
121 for _, wfs := range pipeline.Workflows {
122 for _, w := range wfs {
123 wid := models.WorkflowId{
124 PipelineId: pipelineId,
125 Name: w.Name,
126 }
127 wfCounts[wid.String()]++
128 }
129 }
130 var wg sync.WaitGroup
131 for eng, wfs := range pipeline.Workflows {
132 workflowTimeout := eng.WorkflowTimeout()
133 l.Info("using workflow timeout", "timeout", workflowTimeout)
134
135 for _, w := range wfs {
136 w := w
137 wid := models.WorkflowId{
138 PipelineId: pipelineId,
139 Name: w.Name,
140 }
141
142 if wfCounts[wid.String()] > 1 {
143 l.Warn("skipping workflow due to name collision", "wid", wid, "key", wid.String())
144 dbErr := db.StatusFailed(wid, fmt.Sprintf("colliding workflow name: %s; rename to something else", wid.String()), -1, n)
145 if dbErr != nil {
146 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
147 }
148 continue
149 }
150
151 wg.Go(func() {
152 if st, err := db.GetStatus(wid); err == nil && models.StatusKind(st.Status).IsFinish() {
153 l.Info("skipping finished workflow", "wid", wid, "status", st.Status)
154 return
155 }
156 var err error
157 var wfLogger models.WorkflowLogger
158 if p, ok := eng.(workflowLoggerProvider); ok {
159 wfLogger = p.WorkflowLogger(wid)
160 } else if fileLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues); err != nil {
161 l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
162 wfLogger = models.NullLogger{}
163 } else {
164 l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
165 wfLogger = fileLogger
166 defer archiveWorkflowLog(l, stores, db, cfg.Server.LogDir, wid)
167 defer fileLogger.Close()
168 }
169
170 timeoutCtx, timeoutCancel := context.WithTimeout(ctx, workflowTimeout)
171 defer timeoutCancel()
172
173 wfCtx, userCancel := context.WithCancelCause(timeoutCtx)
174 defer userCancel(nil)
175
176 // allow wf context to be cancelled properly by manual cancel
177 activeMu.Lock()
178 activeCancels[wid] = userCancel
179 activeMu.Unlock()
180 defer func() {
181 activeMu.Lock()
182 delete(activeCancels, wid)
183 activeMu.Unlock()
184 }()
185
186 l.Info("waiting for slot", "wid", wid)
187 slot := WorkflowSlot(NoopSlot{})
188 _, remoteStatus := eng.(RemoteStatusEngine)
189
190 if s, ok := eng.(WorkflowSlotter); ok {
191 slot, err = s.AcquireWorkflowSlot(wfCtx, wid, &w, Wait)
192 if err != nil {
193 writeWfError(db, n, l, wfCtx, wid, "waiting for slot", err)
194 return
195 }
196 }
197 defer slot.Release()
198
199 if !remoteStatus {
200 err := db.StatusRunning(wid, n)
201 if err != nil {
202 l.Error("failed to set workflow status to running", "wid", wid, "err", err)
203 return
204 }
205 }
206
207 err = eng.SetupWorkflow(wfCtx, wid, &w, wfLogger)
208 if err != nil {
209 if !isCanceled(wfCtx) {
210 if destroyErr := eng.DestroyWorkflow(ctx, wid); destroyErr != nil {
211 l.Error("failed to destroy workflow after setup failure", "error", destroyErr)
212 }
213 }
214 if !remoteStatus {
215 writeWfError(db, n, l, wfCtx, wid, "setting up workflow", err)
216 }
217 return
218 }
219 defer eng.DestroyWorkflow(ctx, wid)
220
221 for stepIdx, step := range w.Steps {
222 if wfLogger != nil {
223 wfLogger.
224 ControlWriter(stepIdx, step, models.StepStatusStart).
225 Write([]byte{0})
226 }
227
228 err = eng.RunStep(wfCtx, wid, &w, stepIdx, allSecrets, wfLogger)
229
230 if wfLogger != nil {
231 wfLogger.
232 ControlWriter(stepIdx, step, models.StepStatusEnd).
233 Write([]byte{0})
234 }
235
236 if err != nil {
237 if !remoteStatus {
238 writeWfError(db, n, l, wfCtx, wid, "running step", err)
239 }
240 return
241 }
242 }
243
244 if isCanceled(wfCtx) {
245 if !remoteStatus {
246 writeWfError(db, n, l, wfCtx, wid, "before success", nil)
247 }
248 return
249 }
250
251 if !remoteStatus {
252 err = db.StatusSuccess(wid, n)
253 if err != nil {
254 l.Error("failed to set workflow status to success", "wid", wid, "err", err)
255 }
256 }
257 })
258 }
259 }
260
261 wg.Wait()
262 l.Info("all workflows completed")
263}
264
265func archiveWorkflowLog(l *slog.Logger, stores *artifactstore.Stores, database *db.DB, logDir string, wid models.WorkflowId) {
266 if stores == nil {
267 return
268 }
269 logPath := models.LogFilePath(logDir, wid)
270 file, err := os.Open(logPath)
271 if err != nil {
272 l.Error("open workflow log for archival", "wid", wid, "err", err)
273 return
274 }
275 hash := sha256.New()
276 if _, err := io.Copy(hash, file); err != nil {
277 _ = file.Close()
278 l.Error("hash workflow log", "wid", wid, "err", err)
279 return
280 }
281 _ = file.Close()
282
283 ref := wid.String() + ".log"
284 uploadCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
285 defer cancel()
286 errs := stores.PutFile(uploadCtx, ref, logPath)
287 for _, err := range errs {
288 l.Error("archive workflow log", "wid", wid, "err", err)
289 }
290 if len(errs) == len(stores.Names()) {
291 return
292 }
293 digest := "sha256:" + hex.EncodeToString(hash.Sum(nil))
294 if err := database.SaveArtifactRef(wid.String(), wid.Name, ref, digest); err != nil {
295 l.Error("save workflow log artifact", "wid", wid, "err", err)
296 }
297}