package 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 forever
const 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 cap
func 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
}
// emit masks each line alone, so a secret can't span a newline and complete lines needn't wait
func (w *dataWriter) flushCompleteLines() error {
for {
i := bytes.IndexByte(w.pending[w.scanned:], '\n')
if i < 0 {
w.scanned = len(w.pending)
if len(w.pending)-w.logger.mask.Window() < maxLogLine {
return nil
}
return w.emitLongRun()
}
lineEnd := w.scanned + i + 1
line := append([]byte(nil), w.pending[:lineEnd]...)
w.pending = w.pending[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 there
func (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-is
func (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 not
func (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
}