alice.pds.demo.boltless.dev
This repository has no description
1package executor
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/hex"
7 "fmt"
8 "io"
9 "os"
10 "strings"
11 "sync"
12 "time"
13
14 "github.com/hpcloud/tail"
15 "tangled.org/core/api/tangled"
16 millproto "tangled.org/core/spindle/mill/proto"
17 millv1 "tangled.org/core/spindle/mill/proto/gen"
18 "tangled.org/core/spindle/models"
19 "tangled.org/core/spindle/secrets"
20)
21
22func (e *Executor) observeLoop(ctx context.Context, sub <-chan struct{}, cursor int64) {
23 ticker := time.NewTicker(5 * time.Second)
24 defer ticker.Stop()
25
26 for {
27 select {
28 case <-ctx.Done():
29 return
30 case <-sub:
31 case <-ticker.C:
32 }
33 e.drainEvents(&cursor)
34 }
35}
36
37func (e *Executor) drainEvents(cursor *int64) {
38 events, err := e.db.GetEvents(*cursor, 128)
39 if err != nil {
40 e.l.Error("drain status events failed", "err", err)
41 return
42 }
43 for _, ev := range events {
44 if ev.Created > *cursor {
45 *cursor = ev.Created
46 }
47 st, ok := parseStatus(ev.EventJson)
48 if !ok {
49 continue
50 }
51 if err := e.onStatusRow(st); err != nil {
52 e.l.Error("process status row failed", "err", err)
53 }
54 }
55}
56
57func (e *Executor) onStatusRow(st *tangled.PipelineStatus) error {
58 res := e.reservationFor(st.Pipeline, st.Workflow)
59 if res == nil {
60 return nil
61 }
62
63 if models.StatusKind(st.Status).IsFinish() {
64 return e.finishJob(res, st)
65 }
66
67 return e.appendStatus(res.leaseID, st)
68}
69
70func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) error {
71 e.mu.Lock()
72 if e.active[res.leaseID] != res {
73 e.mu.Unlock()
74 return nil
75 }
76 cancelled := res.cancelled
77 e.mu.Unlock()
78
79 // 1. Finalize log tail first so all log lines precede terminal event
80 if res.stopTail != nil {
81 res.stopTail()
82 }
83
84 terminalStatus := st.Status
85 if cancelled {
86 terminalStatus = string(models.StatusKindCancelled)
87 }
88 var logDir string
89 if e.cfg != nil {
90 logDir = e.cfg.Server.LogDir
91 }
92 logPath := models.LogFilePath(logDir, res.wid)
93
94 var errStr string
95 if st != nil && st.Error != nil {
96 errStr = *st.Error
97 }
98 var exitCode int64
99 if st != nil && st.ExitCode != nil {
100 exitCode = *st.ExitCode
101 }
102
103 // Calculate SHA256 of log file
104 hash := ""
105 if f, err := os.Open(logPath); err == nil {
106 h := sha256.New()
107 if _, err := io.Copy(h, f); err == nil {
108 hash = "sha256:" + hex.EncodeToString(h.Sum(nil))
109 }
110 _ = f.Close()
111 }
112
113 // refs are opaque keys interpreted by the configured artifact store
114 ref := "logs/" + res.leaseID + ".log"
115
116 // 2. Persist restart-retryable pending artifact state
117 if e.db != nil {
118 _ = e.db.SavePendingArtifact(res.leaseID, res.wid.Name, terminalStatus, errStr, exitCode, ref, hash)
119 }
120
121 // 3. Upload artifact with non-cancelled cleanup context
122 cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(context.Background()), 2*time.Minute)
123 defer cancel()
124
125 if e.writer != nil {
126 if f, err := os.Open(logPath); err == nil {
127 defer f.Close()
128 if uploadErr := e.writer.Put(cleanupCtx, ref, f); uploadErr != nil {
129 e.l.Error("artifact upload failed", "lease", res.leaseID, "err", uploadErr)
130 return fmt.Errorf("artifact upload: %w", uploadErr)
131 }
132 }
133 }
134
135 // 4. Append terminal event (with LogArtifact) to outbox
136 if err := e.appendTerminalWithArtifact(res.leaseID, terminalStatus, st, ref, hash); err != nil {
137 return err
138 }
139
140 // 5. Remove pending artifact state after terminal outbox append succeeds
141 if e.db != nil {
142 _ = e.db.RemovePendingArtifact(res.leaseID)
143 }
144 e.mu.Lock()
145 cleanup := e.removeReservationLocked(res, false)
146 e.mu.Unlock()
147
148 cleanup()
149
150 e.pushSnapshot()
151 return nil
152}
153
154func (e *Executor) reservationFor(pipelineAturi, workflow string) *reservation {
155 e.mu.Lock()
156 defer e.mu.Unlock()
157 for _, res := range e.active {
158 if string(res.wid.PipelineId.AtUri()) == pipelineAturi && res.wid.Name == workflow {
159 return res
160 }
161 }
162 return nil
163}
164
165func (e *Executor) maskSecrets(res *reservation, text string) string {
166 if res == nil || res.vault == nil {
167 return text
168 }
169 for _, s := range res.vault.secrets {
170 if s.Value != "" {
171 text = strings.ReplaceAll(text, s.Value, "***")
172 }
173 }
174 return text
175}
176
177func (e *Executor) SendLiveLog(leaseID string, raw []byte) error {
178 e.connMu.Lock()
179 enc := e.enc
180 e.connMu.Unlock()
181 if enc == nil {
182 return nil
183 }
184 return enc.Encode(&millproto.Message{
185 LiveLog: &millv1.LiveLog{
186 LeaseId: leaseID,
187 RawJson: raw,
188 },
189 })
190}
191
192func (e *Executor) startTail(res *reservation) {
193 path := models.LogFilePath(e.cfg.Server.LogDir, res.wid)
194 t, err := tail.TailFile(path, tail.Config{
195 Follow: true,
196 ReOpen: true,
197 MustExist: false,
198 Location: &tail.SeekInfo{Offset: 0, Whence: io.SeekStart},
199 Logger: tail.DiscardingLogger,
200 })
201 if err != nil {
202 e.l.Error("tail log file failed", "wid", res.wid, "err", err)
203 return
204 }
205
206 done := make(chan struct{})
207 go func() {
208 defer close(done)
209 for line := range t.Lines {
210 if line == nil || line.Err != nil {
211 continue
212 }
213 masked := e.maskSecrets(res, line.Text)
214 _ = e.SendLiveLog(res.leaseID, []byte(masked+"\n"))
215 }
216 }()
217
218 var once sync.Once
219 res.stopTail = func() {
220 once.Do(func() {
221 _ = t.StopAtEOF()
222 <-done
223 })
224 }
225}
226
227type memVault struct {
228 secrets []secrets.UnlockedSecret
229}
230
231func newMemVault(pb []*millv1.Secret) *memVault {
232 v := &memVault{secrets: make([]secrets.UnlockedSecret, 0, len(pb))}
233 for _, s := range pb {
234 v.secrets = append(v.secrets, secrets.UnlockedSecret{
235 Key: s.Key,
236 Value: s.Value,
237 })
238 }
239 return v
240}
241
242func (v *memVault) GetSecretsUnlocked(ctx context.Context, repo secrets.RepoIdentifier) ([]secrets.UnlockedSecret, error) {
243 return v.secrets, nil
244}
245func (v *memVault) GetSecretsLocked(ctx context.Context, repo secrets.RepoIdentifier) ([]secrets.LockedSecret, error) {
246 return nil, nil
247}
248func (v *memVault) AddSecret(ctx context.Context, s secrets.UnlockedSecret) error { return nil }
249func (v *memVault) RemoveSecret(ctx context.Context, s secrets.Secret[any]) error { return nil }
250
251var _ secrets.Manager = (*memVault)(nil)