Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457package executor
import ( "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "os" "time"
"google.golang.org/protobuf/proto" "tangled.org/core/api/tangled" "tangled.org/core/spindle/db" "tangled.org/core/spindle/engine" millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" "tangled.org/core/spindle/models")
const ( maxBatchBytes = 4 * 1024 * 1024 maxBatchEvents = 128)
func generateEpoch() string { var b [8]byte _, _ = rand.Read(b[:]) return hex.EncodeToString(b[:])}
func (e *Executor) initOutbox() error { return e.loadOutbox()}
func (e *Executor) loadOutbox() error { e.eventMu.Lock() defer e.eventMu.Unlock()
epoch, _, err := e.db.GetOutboxState() if err != nil { return err } if epoch != "" { e.epoch = epoch } else { e.epoch = generateEpoch() if err := e.db.SetOutboxEpoch(e.epoch); err != nil { return fmt.Errorf("set outbox epoch: %w", err) } }
rows, err := e.db.ListOutboxRows() if err != nil { return fmt.Errorf("list outbox rows: %w", err) }
e.terminalSeqnos = make(map[string]uint64) for _, r := range rows { e.outboxBytes += r.ByteSize entry, err := decodeEvent(r.Payload, r.Seqno) if err == nil && entry.GetAttemptResult() != nil { e.terminalSeqnos[entry.GetLeaseId()] = r.Seqno } } if e.outboxBytes > 0 { e.outboxIdleCh = make(chan struct{}) } return nil}
func (e *Executor) appendAndSend(leaseID string, payload any, control bool) error { entry := &millv1.Event{LeaseId: leaseID} switch payload := payload.(type) { case *millv1.Event_StatusEvent: entry.Payload = payload case *millv1.Event_AttemptResult: entry.Payload = payload case *millv1.Event_CacheUpdate: entry.Payload = payload default: return fmt.Errorf("unsupported stream payload %T", payload) } durable := false isTerminal := false switch payload.(type) { case *millv1.Event_AttemptResult: durable = true isTerminal = true case *millv1.Event_CacheUpdate: durable = true } entry.Seqno = ^uint64(0) wireSize := proto.Size(&millproto.Message{EventBatch: &millv1.EventBatch{ Epoch: e.epoch, Events: []*millv1.Event{entry}, }}) entry.Seqno = 0 if wireSize > maxBatchBytes { return fmt.Errorf("control stream entry exceeds wire limit: %d > %d", wireSize, maxBatchBytes) }
encoded, err := proto.Marshal(entry) if err != nil { return fmt.Errorf("marshal stream entry: %w", err) }
e.eventMu.Lock() if _, terminal := e.terminalSeqnos[leaseID]; terminal { e.l.Warn("dropping stream entry after terminal", "lease", leaseID) e.eventMu.Unlock() return nil } if control && !durable && e.maxOutboxBytes > 0 && e.outboxBytes+int64(len(encoded)) > e.maxOutboxBytes { e.l.Warn("outbox reserve exhausted; dropping nonterminal status", "cap", e.maxOutboxBytes) e.eventMu.Unlock() return nil } seqno, err := e.db.AppendOutboxRow(encoded, control) if err != nil { e.eventMu.Unlock() return fmt.Errorf("append outbox row: %w", err) } if isTerminal { e.terminalSeqnos[leaseID] = seqno } if e.outboxBytes == 0 { e.outboxIdleCh = make(chan struct{}) } e.outboxBytes += int64(len(encoded)) e.eventMu.Unlock()
e.sendPending() return nil}
func (e *Executor) appendStatus(leaseID string, st *tangled.PipelineStatus) error { if st.Status != string(models.StatusKindRunning) { return fmt.Errorf("unsupported nonterminal status %q", st.Status) } exit, errStr := parseStatusExitAndError(st) payload := &millv1.Event_StatusEvent{StatusEvent: &millv1.StatusEvent{ Status: millv1.NonterminalStatus_NONTERMINAL_STATUS_RUNNING, Error: errStr, ExitCode: exit, }} return e.appendAndSend(leaseID, payload, true)}
func (e *Executor) appendCacheUpdate(leaseID string, update engine.CacheUpdate) error { var action millv1.CacheUpdateAction switch update.Action { case engine.CacheUsed: action = millv1.CacheUpdateAction_CACHE_UPDATE_ACTION_USED case engine.CacheStored: action = millv1.CacheUpdateAction_CACHE_UPDATE_ACTION_STORED case engine.CacheDiscarded: action = millv1.CacheUpdateAction_CACHE_UPDATE_ACTION_DISCARDED case engine.CacheMissing: action = millv1.CacheUpdateAction_CACHE_UPDATE_ACTION_MISSING default: return fmt.Errorf("unsupported cache update action %d", update.Action) } payload := &millv1.Event_CacheUpdate{CacheUpdate: &millv1.CacheUpdate{ Action: action, Id: update.ID, SizeBytes: update.SizeBytes, Checksum: update.Checksum, }} return e.appendAndSend(leaseID, payload, true)}
func (e *Executor) appendTerminal(leaseID, status string, st *tangled.PipelineStatus) error { failureClass, failureReason := defaultTerminalAttribution(status) return e.appendTerminalWithArtifact(leaseID, status, st, "", "", failureClass, failureReason, true)}
func defaultTerminalAttribution(status string) (string, string) { switch status { case string(models.StatusKindSuccess): return string(engine.FailureClassNone), string(engine.FailureReasonSuccess) case string(models.StatusKindTimeout): return string(engine.FailureClassUser), string(engine.FailureReasonTimeout) case string(models.StatusKindCancelled): return string(engine.FailureClassUser), string(engine.FailureReasonCancelled) default: return string(engine.FailureClassInfrastructure), string(engine.FailureReasonRuntimeFailed) }}
func (e *Executor) appendTerminalWithArtifact( leaseID, status string, st *tangled.PipelineStatus, ref, hash, failureClass, failureReason string, millRecordsTerminalMetrics bool,) error { var terminalStatus millv1.TerminalStatus switch status { case string(models.StatusKindSuccess): terminalStatus = millv1.TerminalStatus_TERMINAL_STATUS_SUCCESS case string(models.StatusKindFailed): terminalStatus = millv1.TerminalStatus_TERMINAL_STATUS_FAILED case string(models.StatusKindTimeout): terminalStatus = millv1.TerminalStatus_TERMINAL_STATUS_TIMEOUT case string(models.StatusKindCancelled): terminalStatus = millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED default: return fmt.Errorf("unsupported terminal status %q", status) }
exit, errStr := parseStatusExitAndError(st) var logArtifact *millv1.LogArtifact if ref != "" { logArtifact = &millv1.LogArtifact{ Ref: ref, Hash: hash, } } payload := &millv1.Event_AttemptResult{AttemptResult: &millv1.AttemptResult{ Status: terminalStatus, Error: errStr, ExitCode: exit, LogArtifact: logArtifact, FailureClass: failureClass, FailureReason: failureReason, MillRecordsTerminalMetrics: millRecordsTerminalMetrics, }} return e.appendAndSend(leaseID, payload, true)}
func (e *Executor) recoverPendingArtifacts(parent context.Context) error { if e.db == nil { return nil } pending, err := e.db.ListPendingArtifacts() if err != nil || len(pending) == 0 { return err } ctx, cancel := context.WithTimeout(parent, 2*time.Minute) defer cancel() for _, p := range pending { if p.Ref != "" { if e.writer == nil { continue } var logDir string if e.cfg != nil { logDir = e.cfg.Server.LogDir } wid := models.WorkflowId{ PipelineId: models.PipelineId{Knot: p.Knot, Rkey: p.Rkey}, Name: p.Workflow, } logPath := models.LogFilePath(logDir, wid) f, openErr := os.Open(logPath) if openErr != nil { continue } uploadErr := e.writer.Put(ctx, p.Ref, f) _ = f.Close() if uploadErr != nil { continue } }
st := &tangled.PipelineStatus{ Status: p.Status, Error: &p.Error, ExitCode: &p.ExitCode, } if err := e.appendTerminalWithArtifact( p.LeaseID, p.Status, st, p.Ref, p.Hash, p.FailureClass, p.FailureReason, p.MillRecordsTerminalMetrics, ); err == nil { _ = e.db.RemovePendingArtifact(p.LeaseID) } } return nil}
func truncateEventString(value string) string { if len(value) > 65536 { return value[:65536] } return value}
func (e *Executor) sendPending() { e.connMu.Lock() enc := e.enc e.connMu.Unlock()
if enc == nil { return }
e.flushMu.Lock() defer e.flushMu.Unlock()
if err := e.sendPendingLocked(enc); err != nil { e.l.Error("send pending events failed", "err", err) }}
func (e *Executor) sendPendingLocked(enc messageEncoder) error { for { rows, err := e.db.ListOutboxRowsAfter(e.sentSeqno, maxBatchEvents) if err != nil { return fmt.Errorf("list outbox rows after %d: %w", e.sentSeqno, err) } if len(rows) == 0 { return nil } if err := e.sendRows(enc, rows); err != nil { return err } }}
func (e *Executor) sendRows(enc messageEncoder, rows []db.OutboxRow) error { events := make([]*millv1.Event, 0, len(rows)) var lastSeqno uint64
for _, r := range rows { ev, err := decodeEvent(r.Payload, r.Seqno) if err != nil { return fmt.Errorf("decode event at seqno %d: %w", r.Seqno, err) } events = append(events, ev) lastSeqno = r.Seqno }
batch := &millv1.EventBatch{ Epoch: e.epoch, Events: events, }
e.sendMu.Lock() defer e.sendMu.Unlock()
if err := enc.Encode(&millproto.Message{EventBatch: batch}); err != nil { return fmt.Errorf("encode event batch (seqno %d..%d): %w", rows[0].Seqno, lastSeqno, err) } e.sentSeqno = lastSeqno return nil}
func (e *Executor) replay(ackSeqno uint64) error { e.connMu.Lock() enc := e.enc e.connMu.Unlock() if enc == nil { return nil }
e.flushMu.Lock() defer e.flushMu.Unlock()
if ackSeqno > 0 { if err := e.deleteOutboxPrefix(ackSeqno); err != nil { return fmt.Errorf("delete resumed outbox prefix through %d: %w", ackSeqno, err) } } e.sendMu.Lock() e.sentSeqno = ackSeqno e.sendMu.Unlock()
return e.sendPendingLocked(enc)}
func (e *Executor) handleAck(ack *millv1.Ack) { if ack == nil || ack.GetEpoch() != e.epoch { return }
upTo := ack.GetUpToSeqno() if upTo == 0 { return }
if err := e.deleteOutboxPrefix(upTo); err != nil { e.l.Error("delete outbox prefix failed", "upTo", upTo, "err", err) }}
func (e *Executor) subtractOutboxBytes(deleted db.OutboxDeletion) { e.outboxBytes = max(0, e.outboxBytes-deleted.Bytes)}
func parseStatus(raw json.RawMessage) (*tangled.PipelineStatus, bool) { var st tangled.PipelineStatus if err := json.Unmarshal(raw, &st); err != nil { return nil, false } return &st, true}
func (e *Executor) deleteOutboxPrefix(upTo uint64) error { e.eventMu.Lock() defer e.eventMu.Unlock()
deleted, err := e.db.DeleteOutboxPrefix(upTo) if err == nil { e.subtractOutboxBytes(deleted) for leaseID, seqno := range e.terminalSeqnos { if seqno <= upTo { delete(e.terminalSeqnos, leaseID) } } if e.outboxBytes == 0 && e.outboxIdleCh != nil { close(e.outboxIdleCh) e.outboxIdleCh = nil } } return err}
func (e *Executor) outboxIdle() <-chan struct{} { e.eventMu.Lock() defer e.eventMu.Unlock()
if e.outboxBytes > 0 { if e.outboxIdleCh == nil { e.outboxIdleCh = make(chan struct{}) } return e.outboxIdleCh }
idle := make(chan struct{}) close(idle) return idle}
func parseStatusExitAndError(st *tangled.PipelineStatus) (int64, string) { if st == nil { return 0, "" } var exit int64 if st.ExitCode != nil { exit = *st.ExitCode } var errStr string if st.Error != nil { errStr = truncateEventString(*st.Error) } return exit, errStr}
func decodeEvent(payload []byte, seqno uint64) (*millv1.Event, error) { var ev millv1.Event if err := proto.Unmarshal(payload, &ev); err != nil { return nil, err } ev.Seqno = seqno return &ev, nil}