diff --git a/cmd/internal/agent.go b/cmd/internal/agent.go index c2a7a9108..acbdbf27e 100644 --- a/cmd/internal/agent.go +++ b/cmd/internal/agent.go @@ -33,10 +33,6 @@ func NewAgentCmd(globalFlags *flags.GlobalFlags) *cobra.Command { return agentCmd } -// agentPreRunE builds the PersistentPreRunE shared by the agent command and the -// utility plumbing commands. Logging is forced to JSON because the agent -// subprocess uses stdout as a binary protocol channel, so log output must stay -// on stderr as single-line JSON captured by TunnelLogStreamer.lastLines. func agentPreRunE(globalFlags *flags.GlobalFlags) func(*cobra.Command, []string) error { return func(cobraCmd *cobra.Command, _ []string) error { root := cobraCmd diff --git a/cmd/machine/ssh.go b/cmd/machine/ssh.go index 4fb8ccfe6..7396bee63 100644 --- a/cmd/machine/ssh.go +++ b/cmd/machine/ssh.go @@ -130,8 +130,11 @@ func (cmd *SSHCmd) Run(ctx context.Context, args []string) error { return err } - writer := log.Writer(log.LevelInfo) - defer func() { _ = writer.Close() }() + streamer := log.NewJSONLogStreamer(log.StreamerOptions{ + FallbackLevel: log.LevelInfo, + DetectLevelPrefixes: true, + }) + defer func() { _ = streamer.Close() }() // Get the timeout from the context options timeout := config.ParseTimeOption(devsyConfig, config.ContextOptionAgentInjectTimeout) @@ -170,7 +173,7 @@ func (cmd *SSHCmd) Run(ctx context.Context, args []string) error { Timeout: timeout, }) }, - Stderr: writer, + Stderr: streamer, }) } diff --git a/cmd/workspace/logs.go b/cmd/workspace/logs.go index 6943e2102..2aa2f1aca 100644 --- a/cmd/workspace/logs.go +++ b/cmd/workspace/logs.go @@ -118,8 +118,11 @@ func (cmd *LogsCmd) getWorkspaceClient( // injectLogsAgent injects the devsy agent binary over stdin/stdout and runs the // remote ssh-server that runLogsSession then connects to. func injectLogsAgent(ctx context.Context, params injectLogsAgentParams) error { - stderr := log.Writer(log.LevelDebug) - defer func() { _ = stderr.Close() }() + streamer := log.NewJSONLogStreamer(log.StreamerOptions{ + FallbackLevel: log.LevelDebug, + DetectLevelPrefixes: true, + }) + defer func() { _ = streamer.Close() }() return agent.InjectAgent(ctx, &agent.InjectOptions{ Exec: func(ctx context.Context, command string, stdinR io.Reader, stdoutW io.Writer, stderrW io.Writer) error { @@ -136,7 +139,7 @@ func injectLogsAgent(ctx context.Context, params injectLogsAgentParams) error { Command: params.sshServerCmd, Stdin: params.stdin, Stdout: params.stdout, - Stderr: stderr, + Stderr: streamer, Timeout: params.timeout, }) } diff --git a/pkg/client/clientimplementation/workspace_client.go b/pkg/client/clientimplementation/workspace_client.go index 0859088c3..79f72d2c1 100644 --- a/pkg/client/clientimplementation/workspace_client.go +++ b/pkg/client/clientimplementation/workspace_client.go @@ -1154,11 +1154,14 @@ type workspaceInjectionConfig struct { func runAgentInjection(ctx context.Context, cfg workspaceInjectionConfig) chan error { errChan := make(chan error, 1) go func() { - writer := log.Writer(log.LevelInfo) + streamer := log.NewJSONLogStreamer(log.StreamerOptions{ + FallbackLevel: log.LevelInfo, + DetectLevelPrefixes: true, + }) defer func() { log.Debugf("up command completed") cfg.cancel() - _ = writer.Close() + _ = streamer.Close() }() errChan <- agent.InjectAgent(ctx, &agent.InjectOptions{ @@ -1176,7 +1179,7 @@ func runAgentInjection(ctx context.Context, cfg workspaceInjectionConfig) chan e Command: cfg.command, Stdin: cfg.stdin, Stdout: cfg.stdout, - Stderr: writer, + Stderr: streamer, Timeout: cfg.timeout, }) }() diff --git a/pkg/devcontainer/sshtunnel/sshtunnel.go b/pkg/devcontainer/sshtunnel/sshtunnel.go index 32e37a2b9..5fe9dde40 100644 --- a/pkg/devcontainer/sshtunnel/sshtunnel.go +++ b/pkg/devcontainer/sshtunnel/sshtunnel.go @@ -1,16 +1,12 @@ package sshtunnel import ( - "bufio" "context" - "encoding/json" "errors" "fmt" "io" "os" "os/exec" - "strings" - "sync" "time" client2 "github.com/devsy-org/devsy/pkg/client" @@ -98,13 +94,10 @@ func executeSSHServerHelper( ) error { defer log.Debug("done executing SSH server helper command") - // AgentInject's stderr always carries the ssh-server command's structured + // AgentInject's stderr carries the ssh-server command's structured // JSON logs, and for some callers (see newAgentInjectFunc) also the - // injection script's plain-text preamble first. TunnelLogStreamer handles - // both: JSON lines are re-emitted at their own level, plain-text lines - // fall back to level-prefix extraction, so neither is silently dropped or - // double-wrapped as a single log line. - streamer := NewTunnelLogStreamer() + // injection script's plain-text preamble. + streamer := newSSHTunnelJSONLogStreamer() defer func() { _ = streamer.Close() }() log.Debugf("injecting and running SSH server command: %q", opts.SSHCommand) @@ -246,7 +239,7 @@ type sshCommandParams struct { } func runCommandInSSHTunnel(ctx context.Context, p sshCommandParams, stdin, stdout *os.File) error { - streamer := NewTunnelLogStreamer() + streamer := newSSHTunnelJSONLogStreamer() defer func() { _ = streamer.Close() }() log.Debugf("running agent command in SSH tunnel: %q", p.command) @@ -269,157 +262,11 @@ func runCommandInSSHTunnel(ctx context.Context, p sshCommandParams, stdin, stdou const maxLogLines = 1 -type TunnelLogStreamer struct { - pw *io.PipeWriter - done chan struct{} - - mu sync.Mutex - lastLines []string -} - -func NewTunnelLogStreamer() *TunnelLogStreamer { - pr, pw := io.Pipe() - l := &TunnelLogStreamer{ - pw: pw, - done: make(chan struct{}), - lastLines: make([]string, 0, maxLogLines), - } - - go l.process(pr) - return l -} - -func (l *TunnelLogStreamer) Write(p []byte) (int, error) { - return l.pw.Write(p) -} - -func (l *TunnelLogStreamer) Close() error { - err := l.pw.Close() - <-l.done - return err -} - -func (l *TunnelLogStreamer) ErrorOutput() string { - l.mu.Lock() - defer l.mu.Unlock() - - if len(l.lastLines) == 0 { - return "" - } - - return strings.Join(l.lastLines, "\n") -} - -func (l *TunnelLogStreamer) process(r io.Reader) { - defer close(l.done) - scanner := bufio.NewScanner(r) - - buf := make([]byte, 0, 64*1024) - scanner.Buffer(buf, 1024*1024) - - for scanner.Scan() { - line := scanner.Text() - l.logLine(line) - - l.mu.Lock() - if len(l.lastLines) >= maxLogLines { - l.lastLines = l.lastLines[1:] - } - l.lastLines = append(l.lastLines, line) - l.mu.Unlock() - } - - if err := scanner.Err(); err != nil { - log.Debugf("error reading tunnel output: %v", err) - } -} - -type jsonLogLine struct { - Message string `json:"message,omitempty"` - Msg string `json:"msg,omitempty"` - Level string `json:"level,omitempty"` -} - -func (j *jsonLogLine) text() string { - if j.Message != "" { - return j.Message - } - return j.Msg -} - -func (l *TunnelLogStreamer) logLine(line string) { - line = strings.TrimSpace(line) - // Remove carriage returns to prevent terminal overwriting (e.g. git progress) - line = strings.ReplaceAll(line, "\r", "") - if line == "" { - return - } - - var obj jsonLogLine - if json.Unmarshal([]byte(line), &obj) == nil && obj.text() != "" { - level := normalizeLevel(obj.Level) - logAtLevel(level, obj.text()) - return - } - - if matched, level := extractLogLevel(line); matched { - logAtLevel(level, line) - } else { - log.Debug(line) - } -} - -const ( - levelDebug = "debug" - levelInfo = "info" - levelWarn = "warn" - levelError = "error" - levelFatal = "fatal" -) - -func normalizeLevel(raw string) string { - switch strings.ToLower(raw) { - case "trace", levelDebug: - return levelDebug - case levelInfo: - return levelInfo - case "warning", levelWarn: - return levelWarn - case levelError, "panic", levelFatal: - return levelError - default: - return levelDebug - } -} - -func extractLogLevel(line string) (bool, string) { - parts := strings.SplitN(line, " ", 3) - if len(parts) < 2 || !strings.Contains(parts[0], ":") { - return false, "" - } - - level := strings.ToLower(parts[1]) - switch level { - case levelDebug, levelInfo, levelWarn, levelError, levelFatal: - return true, level - default: - return false, "" - } -} - -func logAtLevel(level, msg string) { - switch level { - case levelDebug: - log.Debug(msg) - case levelInfo: - log.Info(msg) - case levelWarn: - log.Warn(msg) - case levelError: - log.Error(msg) - case levelFatal: - log.Error(msg) - default: - log.Debug(msg) - } +func newSSHTunnelJSONLogStreamer() *log.JSONLogStreamer { + return log.NewJSONLogStreamer(log.StreamerOptions{ + FallbackLevel: log.LevelDebug, + CaptureLines: maxLogLines, + DetectLevelPrefixes: true, + TreatUnknownJSONAsDebug: true, + }) } diff --git a/pkg/devcontainer/sshtunnel/sshtunnel_test.go b/pkg/devcontainer/sshtunnel/sshtunnel_test.go index 7461fe46a..15160d3d9 100644 --- a/pkg/devcontainer/sshtunnel/sshtunnel_test.go +++ b/pkg/devcontainer/sshtunnel/sshtunnel_test.go @@ -5,7 +5,6 @@ import ( "fmt" "io" "os" - "strings" "testing" config2 "github.com/devsy-org/devsy/pkg/devcontainer/config" @@ -41,6 +40,12 @@ func TestLogLine_JSONPassthrough(t *testing.T) { wantMsg: "connection lost", wantLevel: zapcore.ErrorLevel, }, + { + name: "unknown JSON level keeps tunnel debug fallback", + input: `{"level":"notice","msg":"legacy tunnel output"}`, + wantMsg: "legacy tunnel output", + wantLevel: zapcore.DebugLevel, + }, { name: "json debug level", input: `{"level":"debug","message":"heartbeat sent"}`, @@ -48,12 +53,12 @@ func TestLogLine_JSONPassthrough(t *testing.T) { wantLevel: zapcore.DebugLevel, }, } - for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { logs := log.InitTestObserved(t, zapcore.DebugLevel) - streamer := &TunnelLogStreamer{} - streamer.logLine(tt.input) + streamer := newSSHTunnelJSONLogStreamer() + defer func() { _ = streamer.Close() }() + streamer.LogLine(tt.input) entries := logs.All() require.Len(t, entries, 1) @@ -93,8 +98,9 @@ func TestLogLine_JSONLevelNormalization(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { logs := log.InitTestObserved(t, zapcore.DebugLevel) - streamer := &TunnelLogStreamer{} - streamer.logLine(tt.input) + streamer := newSSHTunnelJSONLogStreamer() + defer func() { _ = streamer.Close() }() + streamer.LogLine(tt.input) entries := logs.All() require.Len(t, entries, 1) @@ -128,8 +134,9 @@ func TestLogLine_PlainText(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { logs := log.InitTestObserved(t, zapcore.DebugLevel) - streamer := &TunnelLogStreamer{} - streamer.logLine(tt.input) + streamer := newSSHTunnelJSONLogStreamer() + defer func() { _ = streamer.Close() }() + streamer.LogLine(tt.input) entries := logs.All() require.Len(t, entries, 1) @@ -141,114 +148,28 @@ func TestLogLine_PlainText(t *testing.T) { func TestLogLine_EmptyAndWhitespace(t *testing.T) { logs := log.InitTestObserved(t, zapcore.DebugLevel) - streamer := &TunnelLogStreamer{} + streamer := newSSHTunnelJSONLogStreamer() + defer func() { _ = streamer.Close() }() - streamer.logLine("") - streamer.logLine(" ") - streamer.logLine("\r\n") + streamer.LogLine("") + streamer.LogLine(" ") + streamer.LogLine("\r\n") assert.Empty(t, logs.All()) } func TestLogLine_JSONWithoutMessage(t *testing.T) { logs := log.InitTestObserved(t, zapcore.DebugLevel) - streamer := &TunnelLogStreamer{} + streamer := newSSHTunnelJSONLogStreamer() + defer func() { _ = streamer.Close() }() - streamer.logLine(`{"level":"info","key":"value"}`) + streamer.LogLine(`{"level":"info","key":"value"}`) entries := logs.All() require.Len(t, entries, 1) assert.Equal(t, zapcore.DebugLevel, entries[0].Level) } -func TestExtractLogLevel(t *testing.T) { - tests := []struct { - input string - wantMatch bool - wantLevel string - }{ - {"2024-01-01T00:00:00Z debug foo", true, "debug"}, - {"2024-01-01T00:00:00Z info bar", true, "info"}, - {"2024-01-01T00:00:00Z warn baz", true, "warn"}, - {"2024-01-01T00:00:00Z error qux", true, "error"}, - {"2024-01-01T00:00:00Z fatal crash", true, "fatal"}, - {"no-colon info msg", false, ""}, - {"plain text", false, ""}, - {"ts: unknown msg", false, ""}, - } - - for _, tt := range tests { - t.Run(tt.input, func(t *testing.T) { - matched, level := extractLogLevel(tt.input) - assert.Equal(t, tt.wantMatch, matched) - assert.Equal(t, tt.wantLevel, level) - }) - } -} - -func TestRunSSHTunnel_TimingLogs(t *testing.T) { - logs := log.InitTestObserved(t, zapcore.DebugLevel) - - pb, err := tunnel.NewPipeBridge() - require.NoError(t, err) - defer pb.Close() - - // Close the read side so StdioClient fails immediately. - _ = pb.StdoutReader.Close() - - grpcBridge, err := tunnel.NewPipeBridge() - require.NoError(t, err) - defer grpcBridge.Close() - - _, err = runSSHTunnel(t.Context(), sshTunnelParams{ - stdout: pb.StdoutReader, - stdin: pb.StdinWriter, - grpcBridge: grpcBridge, - }) - require.Error(t, err) - - messages := make([]string, 0, len(logs.All())) - for _, entry := range logs.All() { - messages = append(messages, entry.Message) - } - assert.Contains(t, messages, "tunnel: setup start") - - foundComplete := false - for _, msg := range messages { - if strings.HasPrefix(msg, "tunnel: setup complete elapsed=") { - foundComplete = true - break - } - } - assert.True(t, foundComplete, "missing 'tunnel: setup complete' log: %v", messages) -} - -func TestNormalizeLevel(t *testing.T) { - tests := []struct { - input string - want string - }{ - {"trace", "debug"}, - {"DEBUG", "debug"}, - {"info", "info"}, - {"INFO", "info"}, - {"warning", "warn"}, - {"warn", "warn"}, - {"WARN", "warn"}, - {"error", "error"}, - {"panic", "error"}, - {"fatal", "error"}, - {"unknown", "debug"}, - {"", "debug"}, - } - - for _, tt := range tests { - t.Run(tt.input, func(t *testing.T) { - assert.Equal(t, tt.want, normalizeLevel(tt.input)) - }) - } -} - func TestExecuteCommand_PipeBridgeIntegration(t *testing.T) { t.Run("helper error propagates through PipeBridge", func(t *testing.T) { wantErr := fmt.Errorf("inject failed") diff --git a/pkg/driver/docker/build.go b/pkg/driver/docker/build.go index 34366a2b6..a02257943 100644 --- a/pkg/driver/docker/build.go +++ b/pkg/driver/docker/build.go @@ -406,7 +406,9 @@ func (d *dockerDriver) executeBuild( buildOptions *build.BuildOptions, ) error { log.Infof("build with %s", strategy.name()) - writer := log.Writer(log.LevelInfo) + writer := log.NewJSONLogStreamer(log.StreamerOptions{ + FallbackLevel: log.LevelInfo, + }) defer func() { _ = writer.Close() }() if err := strategy.build(ctx, writer, req.Options.Platform, buildOptions); err != nil { diff --git a/pkg/driver/docker/build_test.go b/pkg/driver/docker/build_test.go index 8fff94d2a..bdbd9aa03 100644 --- a/pkg/driver/docker/build_test.go +++ b/pkg/driver/docker/build_test.go @@ -1,6 +1,8 @@ package docker import ( + "context" + "io" "os" "os/exec" "path/filepath" @@ -9,9 +11,12 @@ import ( "github.com/devsy-org/devsy/pkg/devcontainer/build" "github.com/devsy-org/devsy/pkg/docker" + "github.com/devsy-org/devsy/pkg/driver" + "github.com/devsy-org/devsy/pkg/log" "github.com/devsy-org/devsy/pkg/provider" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.uber.org/zap/zapcore" ) func writeHelperScript(t *testing.T, dir, name, output string) string { @@ -177,9 +182,53 @@ func TestTailBuffer(t *testing.T) { assert.Equal(t, "bcde", b.String()) // A single oversized write keeps its tail (buildx emits the real error last). + b2 := &tailBuffer{limit: 4} n, err = b2.Write([]byte("0123456789")) require.NoError(t, err) assert.Equal(t, 10, n) assert.Equal(t, "6789", b2.String()) } + +type testBuildStrategy struct { + output string +} + +func (s testBuildStrategy) build( + _ context.Context, + writer io.Writer, + _ string, + _ *build.BuildOptions, +) error { + _, err := io.WriteString(writer, s.output) + return err +} + +func (testBuildStrategy) name() string { return "test build" } + +func TestExecuteBuildDecodesStructuredOutput(t *testing.T) { + logs := log.InitTestObserved(t, zapcore.DebugLevel) + driverUnderTest := &dockerDriver{} + strategy := testBuildStrategy{ + output: "build starting\n" + + `{"level":"debug","msg":"credential request received"}` + "\n" + + "build complete\n", + } + + err := driverUnderTest.executeBuild( + t.Context(), + strategy, + driver.BuildRequest{}, + &build.BuildOptions{}, + ) + require.NoError(t, err) + + entries := logs.All() + require.Len(t, entries, 4) + assert.Equal(t, "build with test build", entries[0].Message) + assert.Equal(t, "build starting", entries[1].Message) + assert.Equal(t, zapcore.DebugLevel, entries[2].Level) + assert.Equal(t, "credential request received", entries[2].Message) + assert.Equal(t, "build complete", entries[3].Message) + assert.NotContains(t, entries[2].Message, `{"level":"debug"`) +} diff --git a/pkg/log/jsonstream.go b/pkg/log/jsonstream.go index 5d497816b..a7d6b20da 100644 --- a/pkg/log/jsonstream.go +++ b/pkg/log/jsonstream.go @@ -1,9 +1,7 @@ package log import ( - "encoding/json" "io" - "strings" "github.com/devsy-org/devsy/pkg/scanner" ) @@ -44,30 +42,6 @@ func PipeJSONStreamWithFallback(fallback io.Writer) (io.WriteCloser, chan struct return writer, done } -type jsonLine struct { - Message string `json:"message,omitempty"` - Msg string `json:"msg,omitempty"` - Level string `json:"level,omitempty"` -} - -func (l *jsonLine) text() string { - if l.Message != "" { - return l.Message - } - return l.Msg -} - -var levelFuncs = map[string]func(...any){ - "trace": Debug, - "debug": Debug, - "info": Info, - "warning": Warn, - "warn": Warn, - "error": Error, - "panic": Error, - "fatal": Error, -} - func ReadJSONStream(reader io.Reader) { readJSONStreamWithFallback(reader, nil) } @@ -82,22 +56,13 @@ func readJSONStreamWithFallback(reader io.Reader, fallback io.Writer) { if len(line) == 0 { continue } - obj := &jsonLine{} - if err := json.Unmarshal(line, obj); err != nil { - writeFallbackLine(fallback, line) - continue - } - msg := obj.text() - if msg == "" { - writeFallbackLine(fallback, line) - continue - } - fn, ok := levelFuncs[strings.ToLower(obj.Level)] - if !ok { + + decoded, ok := decodeJSONLogLine(line) + if !ok || !decoded.recognized { writeFallbackLine(fallback, line) continue } - fn(msg) + logAtZapLevel(decoded.level, decoded.text) } } diff --git a/pkg/log/streamer.go b/pkg/log/streamer.go new file mode 100644 index 000000000..32f605609 --- /dev/null +++ b/pkg/log/streamer.go @@ -0,0 +1,222 @@ +package log + +import ( + "bufio" + "encoding/json" + "io" + "strings" + "sync" + + "go.uber.org/zap/zapcore" +) + +// StreamerOptions configures a JSON-aware subprocess log streamer. +type StreamerOptions struct { + // FallbackLevel is used for non-structured subprocess output. + FallbackLevel int + // CaptureLines retains this many raw lines for ErrorOutput. + CaptureLines int + // DetectLevelPrefixes preserves a level from timestamp-prefixed plain text. + DetectLevelPrefixes bool + // TreatUnknownJSONAsDebug preserves the historical tunnel behavior for + // JSON messages whose level is absent or unrecognized. + TreatUnknownJSONAsDebug bool +} + +// JSONLogStreamer consumes line-oriented subprocess output. Devsy structured +// log envelopes are decoded and emitted at their original level; all other +// lines are emitted at FallbackLevel. +type JSONLogStreamer struct { + pw *io.PipeWriter + done chan struct{} + + fallbackLevel zapcore.Level + detectLevelPrefixes bool + treatUnknownJSONAsDebug bool + captureLines int + + mu sync.Mutex + lastLines []string + closeOnce sync.Once + closeErr error +} + +// NewJSONLogStreamer returns a writer that decodes Devsy JSON log lines while +// preserving ordinary subprocess output at the configured fallback level. +func NewJSONLogStreamer(options StreamerOptions) *JSONLogStreamer { + pr, pw := io.Pipe() + streamer := &JSONLogStreamer{ + pw: pw, + done: make(chan struct{}), + fallbackLevel: verbosityConstToZapLevel(options.FallbackLevel), + detectLevelPrefixes: options.DetectLevelPrefixes, + treatUnknownJSONAsDebug: options.TreatUnknownJSONAsDebug, + captureLines: options.CaptureLines, + } + if options.CaptureLines > 0 { + streamer.lastLines = make([]string, 0, options.CaptureLines) + } + + go streamer.process(pr) + return streamer +} + +func (s *JSONLogStreamer) Write(p []byte) (int, error) { + return s.pw.Write(p) +} + +// Close stops the reader after draining all complete and unterminated lines. +func (s *JSONLogStreamer) Close() error { + if s.pw == nil { + return nil + } + s.closeOnce.Do(func() { s.closeErr = s.pw.Close() }) + <-s.done + return s.closeErr +} + +// ErrorOutput returns the most recent captured raw lines, joined by newlines. +func (s *JSONLogStreamer) ErrorOutput() string { + s.mu.Lock() + defer s.mu.Unlock() + + return strings.Join(s.lastLines, "\n") +} + +// LogLine processes one line. It is exported for callers that need to feed a +// line directly, while Write remains the normal subprocess boundary API. +func (s *JSONLogStreamer) LogLine(line string) { + line = strings.TrimSpace(line) + line = strings.ReplaceAll(line, "\r", "") + if line == "" { + return + } + + if decoded, ok := decodeJSONLogLine([]byte(line)); ok && + (decoded.recognized || s.treatUnknownJSONAsDebug) { + logAtZapLevel(decoded.level, decoded.text) + } else if s.detectLevelPrefixes { + if matched, level := extractLevelPrefix(line); matched { + logAtZapLevel(level, line) + } else { + logAtZapLevel(s.fallbackLevel, line) + } + } else { + logAtZapLevel(s.fallbackLevel, line) + } +} + +func (s *JSONLogStreamer) process(reader io.Reader) { + defer close(s.done) + scanner := bufio.NewScanner(reader) + buf := make([]byte, 0, 64*1024) + scanner.Buffer(buf, 1024*1024) + + for scanner.Scan() { + line := scanner.Text() + s.LogLine(line) + s.capture(line) + } + if err := scanner.Err(); err != nil { + Debugf("error reading subprocess output: %v", err) + } + if closer, ok := reader.(io.Closer); ok { + _ = closer.Close() + } +} + +func (s *JSONLogStreamer) capture(line string) { + if s.captureLines <= 0 { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if len(s.lastLines) >= s.captureLines { + s.lastLines = s.lastLines[1:] + } + s.lastLines = append(s.lastLines, line) +} + +type jsonLine struct { + Message string `json:"message,omitempty"` + Msg string `json:"msg,omitempty"` + Level string `json:"level,omitempty"` +} + +func (l *jsonLine) text() string { + if l.Message != "" { + return l.Message + } + return l.Msg +} + +type decodedJSONLogLine struct { + text string + level zapcore.Level + recognized bool +} + +func decodeJSONLogLine(line []byte) (decodedJSONLogLine, bool) { + var obj jsonLine + if err := json.Unmarshal(line, &obj); err != nil || obj.text() == "" { + return decodedJSONLogLine{}, false + } + level, recognized := normalizeZapLevel(obj.Level) + return decodedJSONLogLine{ + text: obj.text(), + level: level, + recognized: recognized, + }, true +} + +const infoLevelName = "info" + +func normalizeZapLevel(raw string) (zapcore.Level, bool) { + switch strings.ToLower(raw) { + case "trace", "debug": + return zapcore.DebugLevel, true + case infoLevelName: + return zapcore.InfoLevel, true + case "warning", "warn": + return zapcore.WarnLevel, true + case "error", "panic", "fatal": + return zapcore.ErrorLevel, true + default: + return zapcore.DebugLevel, false + } +} + +func extractLevelPrefix(line string) (bool, zapcore.Level) { + parts := strings.Fields(line) + if len(parts) < 2 || !strings.Contains(parts[0], ":") { + return false, 0 + } + + switch strings.ToLower(parts[1]) { + case "trace", "debug": + return true, zapcore.DebugLevel + case infoLevelName: + return true, zapcore.InfoLevel + case "warning", "warn": + return true, zapcore.WarnLevel + case "error", "panic", "fatal": + return true, zapcore.ErrorLevel + default: + return false, 0 + } +} + +func logAtZapLevel(level zapcore.Level, message string) { + switch level { + case zapcore.DebugLevel: + Debug(message) + case zapcore.InfoLevel: + Info(message) + case zapcore.WarnLevel: + Warn(message) + case zapcore.ErrorLevel: + Error(message) + default: + Debug(message) + } +} diff --git a/pkg/log/streamer_test.go b/pkg/log/streamer_test.go new file mode 100644 index 000000000..dfaf268a6 --- /dev/null +++ b/pkg/log/streamer_test.go @@ -0,0 +1,188 @@ +package log + +import ( + "bytes" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +func streamTestOutput(t *testing.T, input string, options StreamerOptions) []observerEntry { + t.Helper() + logs := InitTestObserved(t, zapcore.DebugLevel) + streamer := NewJSONLogStreamer(options) + _, err := streamer.Write([]byte(input)) + require.NoError(t, err) + require.NoError(t, streamer.Close()) + + entries := logs.All() + output := make([]observerEntry, len(entries)) + for i, entry := range entries { + output[i] = observerEntry{Level: entry.Level, Message: entry.Message} + } + return output +} + +type observerEntry struct { + Level zapcore.Level + Message string +} + +func TestJSONLogStreamerStructuredLogPreservesLevel(t *testing.T) { + entries := streamTestOutput( + t, + `{"level":"debug","ts":"2026-09-01T08:27:30.002Z","msg":"received docker credentials post data: bytes=23"}`+"\n", + StreamerOptions{FallbackLevel: LevelInfo}, + ) + require.Len(t, entries, 1) + assert.Equal(t, zapcore.DebugLevel, entries[0].Level) + assert.Equal(t, "received docker credentials post data: bytes=23", entries[0].Message) + assert.NotContains(t, entries[0].Message, `{"level":"debug"`) +} + +func TestJSONLogStreamerSupportsMessageField(t *testing.T) { + entries := streamTestOutput( + t, + `{"level":"warn","message":"example warning"}`+"\n", + StreamerOptions{ + FallbackLevel: LevelInfo, + }, + ) + + require.Len(t, entries, 1) + assert.Equal(t, zapcore.WarnLevel, entries[0].Level) + assert.Equal(t, "example warning", entries[0].Message) +} + +func TestJSONLogStreamerUnknownLevelUsesFallback(t *testing.T) { + line := `{"msg":"application JSON output"}` + entries := streamTestOutput( + t, + line+"\n", + StreamerOptions{FallbackLevel: LevelInfo}, + ) + + require.Len(t, entries, 1) + assert.Equal(t, zapcore.InfoLevel, entries[0].Level) + assert.Equal(t, line, entries[0].Message) +} + +func TestJSONLogStreamerPlainTextUsesFallbackLevel(t *testing.T) { + entries := streamTestOutput( + t, + "#12 [4/8] RUN apt-get update\n", + StreamerOptions{ + FallbackLevel: LevelInfo, + }, + ) + require.Len(t, entries, 1) + assert.Equal(t, zapcore.InfoLevel, entries[0].Level) + assert.Equal(t, "#12 [4/8] RUN apt-get update", entries[0].Message) +} + +func TestJSONLogStreamerMalformedJSONUsesFallbackLevel(t *testing.T) { + line := `{"level":"debug","msg":` + entries := streamTestOutput( + t, + line+"\n", + StreamerOptions{FallbackLevel: LevelInfo}, + ) + + require.Len(t, entries, 1) + assert.Equal(t, zapcore.InfoLevel, entries[0].Level) + assert.Equal(t, line, entries[0].Message) +} + +func TestJSONLogStreamerMixedStream(t *testing.T) { + entries := streamTestOutput( + t, + strings.Join([]string{ + "#1 loading build definition", + `{"level":"debug","msg":"received docker credentials post data: bytes=23"}`, + "#2 building image", + "", + }, "\n"), + StreamerOptions{FallbackLevel: LevelInfo}, + ) + + require.Len(t, entries, 3) + assert.Equal(t, zapcore.InfoLevel, entries[0].Level) + assert.Equal(t, "#1 loading build definition", entries[0].Message) + assert.Equal(t, zapcore.DebugLevel, entries[1].Level) + assert.Equal(t, "received docker credentials post data: bytes=23", entries[1].Message) + assert.Equal(t, zapcore.InfoLevel, entries[2].Level) + assert.Equal(t, "#2 building image", entries[2].Message) +} + +func TestJSONLogStreamerNoDoubleWrappedStructuredOutput(t *testing.T) { + entries := streamTestOutput( + t, + `{"level":"debug","msg":"credential request received"}`+"\n", + StreamerOptions{ + FallbackLevel: LevelInfo, + }, + ) + + require.Len(t, entries, 1) + assert.Equal(t, zapcore.DebugLevel, entries[0].Level) + assert.NotContains(t, entries[0].Message, `{"level":"debug"`) +} + +func TestJSONLogStreamerTabDelimitedLevelPrefix(t *testing.T) { + entries := streamTestOutput( + t, + "2026-09-01T08:27:30.002Z\tDEBUG\tcredential request received\n", + StreamerOptions{ + FallbackLevel: LevelInfo, + DetectLevelPrefixes: true, + }, + ) + + require.Len(t, entries, 1) + assert.Equal(t, zapcore.DebugLevel, entries[0].Level) + assert.Equal( + t, + "2026-09-01T08:27:30.002Z\tDEBUG\tcredential request received", + entries[0].Message, + ) +} + +func TestJSONLogStreamerCaptureLinesIsBounded(t *testing.T) { + logs := InitTestObserved(t, zapcore.DebugLevel) + streamer := NewJSONLogStreamer(StreamerOptions{ + FallbackLevel: LevelInfo, + CaptureLines: 1, + }) + _, err := streamer.Write([]byte("first\nsecond\n")) + require.NoError(t, err) + require.NoError(t, streamer.Close()) + + assert.Equal(t, "second", streamer.ErrorOutput()) + assert.Len(t, logs.All(), 2) +} + +func TestJSONLogStreamerFormattedOutputHasNoNestedEnvelope(t *testing.T) { + var output bytes.Buffer + previous := sugar.Load() + encoderConfig := zap.NewDevelopmentEncoderConfig() + encoderConfig.EncodeLevel = zapcore.CapitalLevelEncoder + core := zapcore.NewCore( + zapcore.NewConsoleEncoder(encoderConfig), + zapcore.AddSync(&output), + zapcore.DebugLevel, + ) + sugar.Store(zap.New(core).Sugar()) + t.Cleanup(func() { sugar.Store(previous) }) + + streamer := NewJSONLogStreamer(StreamerOptions{FallbackLevel: LevelInfo}) + _, err := streamer.Write([]byte(`{"level":"debug","msg":"credential request received"}` + "\n")) + require.NoError(t, err) + require.NoError(t, streamer.Close()) + + assert.NotContains(t, output.String(), `INFO {"level":"debug"`) + assert.Contains(t, output.String(), "DEBUG\tcredential request received") +}