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)
22
23type workflowFinalizer interface {
24 FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error
25}
26
27func 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) {
28 l.Info("starting all workflows in parallel", "pipeline", pipelineId)
29
30 var allSecrets []secrets.UnlockedSecret
31 // never pass secrets to pipelines that run untrusted (e.g. fork) code
32 if pipeline.TrustedSource && pipeline.RepoDid != "" {
33 if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil {
34 allSecrets = res
35 }
36 } else if !pipeline.TrustedSource {
37 l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId)
38 }
39
40 secretValues := make([]string, len(allSecrets))
41 for i, s := range allSecrets {
42 secretValues[i] = s.Value
43 }
44
45 s3, err := NewS3(cfg.S3.LogBucket)
46 if err != nil {
47 l.Error("error creating s3 client", "err", err)
48 }
49
50 // wid.String() is lossy so two different names can map to the same key
51 // eg. "foo bar" and "foo-bar"...
52 wfCounts := make(map[string]int)
53 for _, wfs := range pipeline.Workflows {
54 for _, w := range wfs {
55 wid := models.WorkflowId{
56 PipelineId: pipelineId,
57 Name: w.Name,
58 }
59 wfCounts[wid.String()]++
60 }
61 }
62
63 var wg sync.WaitGroup
64 for eng, wfs := range pipeline.Workflows {
65 workflowTimeout := eng.WorkflowTimeout()
66 l.Info("using workflow timeout", "timeout", workflowTimeout)
67
68 for _, w := range wfs {
69 w := w
70 wid := models.WorkflowId{
71 PipelineId: pipelineId,
72 Name: w.Name,
73 }
74
75 if wfCounts[wid.String()] > 1 {
76 l.Warn("skipping workflow due to name collision", "wid", wid, "key", wid.String())
77 dbErr := db.StatusFailed(wid, fmt.Sprintf("colliding workflow name: %s; rename to something else", wid.String()), -1, n)
78 if dbErr != nil {
79 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
80 }
81 continue
82 }
83
84 wg.Go(func() {
85
86 defer func() {
87 if s3 != nil {
88 logFile := filepath.Join(cfg.Server.LogDir, fmt.Sprintf("%s.log", wid.String()))
89 if err := s3.WriteFile(ctx, logFile); err != nil {
90 l.Error("error uploading logs", "err", err)
91 }
92 }
93 }()
94
95 wfLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues)
96 if err != nil {
97 l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
98 wfLogger = models.NullLogger{}
99 } else {
100 l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
101 defer wfLogger.Close()
102 }
103
104 l.Info("waiting for slot", "wid", wid)
105 slot := WorkflowSlot(NoopSlot{})
106 if s, ok := eng.(WorkflowSlotter); ok {
107 var err error
108 slot, err = s.AcquireWorkflowSlot(ctx, wid, &w)
109 if err != nil {
110 l.Error("failed to acquire slot", "wid", wid, "err", err)
111 dbErr := db.StatusFailed(wid, err.Error(), -1, n)
112 if dbErr != nil {
113 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
114 }
115 return
116 }
117 }
118 defer slot.Release()
119
120 err = db.StatusRunning(wid, n)
121 if err != nil {
122 l.Error("failed to set workflow status to running", "wid", wid, "err", err)
123 return
124 }
125
126 err = eng.SetupWorkflow(ctx, wid, &w, wfLogger)
127 if err != nil {
128 // TODO(winter): Should this always set StatusFailed?
129 // In the original, we only do in a subset of cases.
130 l.Error("setting up workflow", "wid", wid, "err", err)
131
132 destroyErr := eng.DestroyWorkflow(ctx, wid)
133 if destroyErr != nil {
134 l.Error("failed to destroy workflow after setup failure", "error", destroyErr)
135 }
136
137 dbErr := db.StatusFailed(wid, err.Error(), -1, n)
138 if dbErr != nil {
139 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
140 }
141 return
142 }
143 defer eng.DestroyWorkflow(ctx, wid)
144
145 ctx, cancel := context.WithTimeout(ctx, workflowTimeout)
146 defer cancel()
147
148 for stepIdx, step := range w.Steps {
149 // log start of step
150 if wfLogger != nil {
151 wfLogger.
152 ControlWriter(stepIdx, step, models.StepStatusStart).
153 Write([]byte{0})
154 }
155
156 err = eng.RunStep(ctx, wid, &w, stepIdx, allSecrets, wfLogger)
157
158 // log end of step
159 if wfLogger != nil {
160 wfLogger.
161 ControlWriter(stepIdx, step, models.StepStatusEnd).
162 Write([]byte{0})
163 }
164
165 if err != nil {
166 if errors.Is(err, ErrTimedOut) {
167 dbErr := db.StatusTimeout(wid, n)
168 if dbErr != nil {
169 l.Error("failed to set workflow status to timeout", "wid", wid, "err", dbErr)
170 }
171 } else {
172 dbErr := db.StatusFailed(wid, err.Error(), -1, n)
173 if dbErr != nil {
174 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
175 }
176 }
177 return
178 }
179 }
180
181 if finalizer, ok := eng.(workflowFinalizer); ok {
182 if err := finalizer.FinalizeWorkflow(ctx, wid, &w, wfLogger); err != nil {
183 dbErr := db.StatusFailed(wid, err.Error(), -1, n)
184 if dbErr != nil {
185 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
186 }
187 return
188 }
189 }
190
191 err = db.StatusSuccess(wid, n)
192 if err != nil {
193 l.Error("failed to set workflow status to success", "wid", wid, "err", err)
194 }
195 })
196 }
197 }
198
199 wg.Wait()
200 l.Info("all workflows completed")
201}