From 41b14efce46c9f91edc19a4bd5660881967e0de5 Mon Sep 17 00:00:00 2001 From: dawn Date: Wed, 21 Oct 2026 16:14:29 +0300 Subject: [PATCH] spindle: cap a workflow's log file Signed-off-by: dawn --- spindle/config/config.go | 7 ++- spindle/engine/engine.go | 2 +- spindle/models/logger.go | 89 ++++++++++++++++++++++++------ spindle/models/logger_cap_test.go | 92 +++++++++++++++++++++++++++++++ 4 files changed, 168 insertions(+), 22 deletions(-) create mode 100644 spindle/models/logger_cap_test.go diff --git a/spindle/config/config.go b/spindle/config/config.go index 754480596..23866e561 100644 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -31,9 +31,10 @@ type Server struct { KeyPath string `env:"KEY_PATH"` // PEM holding the EC P-256 key behind the spindle's did document Secrets Secrets `env:",prefix=SECRETS_"` LogDir string `env:"LOG_DIR, default=/var/log/spindle"` - MaxJobCount int `env:"MAX_JOB_COUNT, default=2"` // max number of pipelines that run at a time - DockerSocket string `env:"DOCKER_SOCKET"` // path to a docker socket to expose to workflow containers - CorsOrigins []string `env:"CORS_ORIGINS"` // origins browsers may call xrpc from, "*" for any. empty leaves cors off + MaxJobCount int `env:"MAX_JOB_COUNT, default=2"` // max number of pipelines that run at a time + LogMaxBytes int64 `env:"LOG_MAX_BYTES, default=33554432"` // size of one workflow's stored jsonl log, envelope included, not raw output. output past it is dropped, 0 means no cap + DockerSocket string `env:"DOCKER_SOCKET"` // path to a docker socket to expose to workflow containers + CorsOrigins []string `env:"CORS_ORIGINS"` // origins browsers may call xrpc from, "*" for any. empty leaves cors off } type Hydrant struct { diff --git a/spindle/engine/engine.go b/spindle/engine/engine.go index d9ff27987..8b8bdbd5a 100644 --- a/spindle/engine/engine.go +++ b/spindle/engine/engine.go @@ -459,7 +459,7 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q closeLog := func() {} if p, ok := eng.(workflowLoggerProvider); ok { wfLogger = p.WorkflowLogger(wid) - } else if fileLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues); err != nil { + } else if fileLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues, models.WithMaxLogBytes(cfg.Server.LogMaxBytes)); err != nil { wl.WarnContext(wfCtx, "failed to setup step logger; logs will not be persisted", "error", err) wfLogger = models.NullLogger{} } else { diff --git a/spindle/models/logger.go b/spindle/models/logger.go index 8ab7e6a7c..c1194cb15 100644 --- a/spindle/models/logger.go +++ b/spindle/models/logger.go @@ -3,16 +3,17 @@ package models import ( "bytes" "encoding/json" + "errors" "fmt" "io" "os" "path/filepath" "strings" + "sync" "unicode/utf8" ) -// a step that never prints a newline still has to reach the log, so a line -// goes out once this much is waiting. 64KiB keeps a log line readable in the ui +// cut at this size, because a step that never prints a newline would buffer forever const maxLogLine = 64 << 10 type WorkflowLogger interface { @@ -35,24 +36,50 @@ func (l NullLogger) EndWriter(idx int, step Step, outcome StepOutcome) io.Writer type FileWorkflowLogger struct { file *os.File + mu sync.Mutex encoder *json.Encoder + written int64 + maxBytes int64 + truncated bool mask *SecretMask secretValues []string dataWriters []*dataWriter } -func NewFileWorkflowLogger(baseDir string, wid WorkflowId, secretValues []string) (WorkflowLogger, error) { +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) } - return &FileWorkflowLogger{ + l := &FileWorkflowLogger{ file: file, - encoder: json.NewEncoder(file), mask: NewSecretMask(secretValues), secretValues: append([]string(nil), secretValues...), - }, nil + } + 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 @@ -73,12 +100,11 @@ func LegacyLogFilePath(baseDir, knot string, pipelineID PipelineId, workflow str } func (l *FileWorkflowLogger) Close() error { + var errs []error for _, w := range l.dataWriters { - if err := w.flush(); err != nil { - return err - } + errs = append(errs, w.flush()) } - return l.file.Close() + return errors.Join(append(errs, l.file.Close())...) } func (l *FileWorkflowLogger) DataWriter(idx int, stream string) io.Writer { @@ -110,6 +136,12 @@ func (l *FileWorkflowLogger) EndWriter(idx int, step Step, outcome StepOutcome) } } +func (l *FileWorkflowLogger) isTruncated() bool { + l.mu.Lock() + defer l.mu.Unlock() + return l.truncated +} + type dataWriter struct { logger *FileWorkflowLogger idx int @@ -117,12 +149,14 @@ type dataWriter struct { // trailing bytes held back so a secret split across writes still // matches, flushed on Close or once enough data arrives pending []byte - // leading bytes of pending already known to hold no newline, so a write - // only scans what it added 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 @@ -156,9 +190,8 @@ func (w *dataWriter) flushCompleteLines() error { } } -// masks first and cuts after, because a cut in raw bytes can land inside a secret -// and let both halves through. the last window bytes of the masked text stay -// held, they are what a secret still arriving could start with +// 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() @@ -171,7 +204,7 @@ func (w *dataWriter) emitLongRun() error { return nil } w.pending = []byte(masked[cut:]) - return w.emit([]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 @@ -191,8 +224,26 @@ func (w *dataWriter) emit(p []byte) error { if w.logger.mask != nil { line = w.logger.mask.Mask(line) } - entry := NewDataLogLine(w.idx, line, w.stream) - return w.logger.encoder.Encode(entry) + 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 { @@ -206,6 +257,8 @@ type controlWriter struct { 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 } diff --git a/spindle/models/logger_cap_test.go b/spindle/models/logger_cap_test.go new file mode 100644 index 000000000..369a6c7ac --- /dev/null +++ b/spindle/models/logger_cap_test.go @@ -0,0 +1,92 @@ +package models + +import ( + "bufio" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +type capTestStep struct{} + +func (capTestStep) Name() string { return "build" } +func (capTestStep) Command() string { return "make" } +func (capTestStep) Kind() StepKind { return StepKindUser } + +func TestLogPastTheCapIsDroppedWithOneMarker(t *testing.T) { + dir := t.TempDir() + wid := testWorkflowId("capped") + const maxBytes = 100 << 10 + logger, err := NewFileWorkflowLogger(dir, wid, nil, WithMaxLogBytes(maxBytes)) + if err != nil { + t.Fatal(err) + } + out := logger.DataWriter(0, "stdout") + line := []byte(strings.Repeat("x", 1000) + "\n") + for range 1000 { + if _, err := out.Write(line); err != nil { + t.Fatal(err) + } + } + if _, err := logger.EndWriter(0, capTestStep{}, StepOutcomeSuccess).Write(nil); err != nil { + t.Fatal(err) + } + if err := logger.Close(); err != nil { + t.Fatal(err) + } + + path := filepath.Join(dir, wid.String()+".log") + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if info.Size() > maxBytes+4<<10 { + t.Fatalf("log is %d bytes, want at most %d plus a line and a marker", info.Size(), maxBytes) + } + + f, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + defer f.Close() + var markers int + var last LogLine + sc := bufio.NewScanner(f) + for sc.Scan() { + last = LogLine{} + if err := json.Unmarshal(sc.Bytes(), &last); err != nil { + t.Fatal(err) + } + if strings.HasPrefix(last.Content, "log truncated") { + markers++ + } + } + if markers != 1 { + t.Fatalf("%d truncation markers, want 1", markers) + } + if last.Kind != LogKindControl || last.StepStatus != StepStatusEnd { + t.Fatalf("last line = %+v, want the step end line to survive the cap", last) + } +} + +func TestWriterDropsOutputPastTheCapWithoutBufferingIt(t *testing.T) { + logger, err := NewFileWorkflowLogger(t.TempDir(), testWorkflowId("cap-buffer"), nil, WithMaxLogBytes(1<<10)) + if err != nil { + t.Fatal(err) + } + w := logger.DataWriter(0, "stdout").(*dataWriter) + if _, err := w.Write([]byte(strings.Repeat("x", 2<<10) + "\n")); err != nil { + t.Fatal(err) + } + if _, err := w.Write([]byte("no newline yet")); err != nil { + t.Fatal(err) + } + if len(w.pending) != 0 { + t.Fatalf("%d bytes are buffered after the log hit its cap, want none", len(w.pending)) + } + if err := logger.Close(); err != nil { + t.Fatal(err) + } +} -- 2.54.0