Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267package models
import ( "bytes" "encoding/json" "errors" "fmt" "io" "os" "path/filepath" "strings" "sync" "unicode/utf8")
// cut at this size, because a step that never prints a newline would buffer foreverconst maxLogLine = 64 << 10
type WorkflowLogger interface { Close() error DataWriter(idx int, stream string) io.Writer ControlWriter(idx int, step Step, stepStatus StepStatus) io.Writer EndWriter(idx int, step Step, outcome StepOutcome) io.Writer}
type NullLogger struct{}
func (l NullLogger) Close() error { return nil }func (l NullLogger) DataWriter(idx int, stream string) io.Writer { return io.Discard }func (l NullLogger) ControlWriter(idx int, step Step, stepStatus StepStatus) io.Writer { return io.Discard}func (l NullLogger) EndWriter(idx int, step Step, outcome StepOutcome) io.Writer { return io.Discard}
type FileWorkflowLogger struct { file *os.File mu sync.Mutex encoder *json.Encoder written int64 maxBytes int64 truncated bool mask *SecretMask secretValues []string dataWriters []*dataWriter}
type LogOption func(*FileWorkflowLogger)
// data past n bytes is dropped, control lines still get through, 0 means no capfunc WithMaxLogBytes(n int64) LogOption { return func(l *FileWorkflowLogger) { l.maxBytes = n }}
type countingWriter struct { w io.Writer n *int64}
func (c countingWriter) Write(p []byte) (int, error) { n, err := c.w.Write(p) *c.n += int64(n) return n, err}
func NewFileWorkflowLogger(baseDir string, wid WorkflowId, secretValues []string, opts ...LogOption) (WorkflowLogger, error) { path := LogFilePath(baseDir, wid) file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) if err != nil { return nil, fmt.Errorf("creating log file: %w", err) } l := &FileWorkflowLogger{ file: file, mask: NewSecretMask(secretValues), secretValues: append([]string(nil), secretValues...), } l.encoder = json.NewEncoder(countingWriter{w: file, n: &l.written}) for _, opt := range opts { opt(l) } return l, nil}
// AddSecretValues registers values that only became known after the log was// opened, e.g. a credential minted once the workflow held its slot. Call it// before the first write so nothing reaches the file unmasked.func (l *FileWorkflowLogger) AddSecretValues(values ...string) { l.secretValues = append(l.secretValues, values...) l.mask = NewSecretMask(l.secretValues)}
func LogFilePath(baseDir string, workflowID WorkflowId) string { logFilePath := filepath.Join(baseDir, fmt.Sprintf("%s.log", workflowID.String())) return logFilePath}
func LegacyLogFilePath(baseDir, knot string, pipelineID PipelineId, workflow string) string { return filepath.Join(baseDir, fmt.Sprintf("%s-%s-%s.log", normalize(knot), pipelineID, normalize(workflow)))}
func (l *FileWorkflowLogger) Close() error { var errs []error for _, w := range l.dataWriters { errs = append(errs, w.flush()) } return errors.Join(append(errs, l.file.Close())...)}
func (l *FileWorkflowLogger) DataWriter(idx int, stream string) io.Writer { w := &dataWriter{ logger: l, idx: idx, stream: stream, } l.dataWriters = append(l.dataWriters, w) return w}
func (l *FileWorkflowLogger) ControlWriter(idx int, step Step, stepStatus StepStatus) io.Writer { return &controlWriter{ logger: l, idx: idx, step: step, stepStatus: stepStatus, }}
func (l *FileWorkflowLogger) EndWriter(idx int, step Step, outcome StepOutcome) io.Writer { return &controlWriter{ logger: l, idx: idx, step: step, stepStatus: StepStatusEnd, outcome: outcome, }}
func (l *FileWorkflowLogger) isTruncated() bool { l.mu.Lock() defer l.mu.Unlock() return l.truncated}
type dataWriter struct { logger *FileWorkflowLogger idx int stream string // trailing bytes held back so a secret split across writes still // matches, flushed on Close or once enough data arrives pending []byte scanned int}
func (w *dataWriter) Write(p []byte) (int, error) { if w.logger.isTruncated() { w.pending, w.scanned = nil, 0 return len(p), nil } w.pending = append(w.pending, p...) if err := w.flushCompleteLines(); err != nil { return 0, err } return len(p), nil}
func (w *dataWriter) flushCompleteLines() error { limit := len(w.pending) - w.logger.mask.Window() if limit <= 0 { return nil }
for { i := bytes.IndexByte(w.pending[w.scanned:limit], '\n') if i < 0 { if limit < maxLogLine { w.scanned = limit return nil } return w.emitLongRun() } lineEnd := w.scanned + i + 1 line := append([]byte(nil), w.pending[:lineEnd]...) w.pending = w.pending[lineEnd:] limit -= lineEnd w.scanned = 0 if err := w.emit(line); err != nil { return err } }}
// mask before cutting, a raw cut can split a secret and let both halves through// the last window bytes stay held, a secret still arriving could start therefunc (w *dataWriter) emitLongRun() error { masked := w.logger.mask.Mask(string(w.pending)) cut := len(masked) - w.logger.mask.Window() for cut > 0 && cut < len(masked) && !utf8.RuneStart(masked[cut]) { cut-- } w.scanned = 0 if cut <= 0 { w.pending = []byte(masked) return nil } w.pending = []byte(masked[cut:]) return w.write(strings.TrimRight(masked[:cut], "\r\n"))}
// the writer is done, so a buffered tail can no longer grow into a full// secret and goes out as-isfunc (w *dataWriter) flush() error { if len(w.pending) == 0 { return nil } pending := w.pending w.pending = nil w.scanned = 0 return w.emit(pending)}
func (w *dataWriter) emit(p []byte) error { line := strings.TrimRight(string(p), "\r\n") if w.logger.mask != nil { line = w.logger.mask.Mask(line) } return w.write(line)}
// callers mask first, write does notfunc (w *dataWriter) write(line string) error { l := w.logger l.mu.Lock() defer l.mu.Unlock() if l.truncated { return nil } if err := l.encoder.Encode(NewDataLogLine(w.idx, line, w.stream)); err != nil { return err } if l.maxBytes > 0 && l.written >= l.maxBytes { l.truncated = true marker := fmt.Sprintf("log truncated after %d bytes, the rest of the output was dropped", l.maxBytes) return l.encoder.Encode(NewDataLogLine(w.idx, marker, "stderr")) } return nil}
type controlWriter struct { logger *FileWorkflowLogger idx int step Step stepStatus StepStatus outcome StepOutcome}
func (w *controlWriter) Write(_ []byte) (int, error) { entry := NewControlLogLine(w.idx, w.step, w.stepStatus) entry.StepOutcome = w.outcome w.logger.mu.Lock() defer w.logger.mu.Unlock() if err := w.logger.encoder.Encode(entry); err != nil { return 0, err } return len(w.step.Name()), nil}