Support JSON output formatting for apps logs command (#4087)

## Changes
- Support JSON output formatting for `apps logs` command

## Why
To be consistent with the global flags we have.

## Tests
Unit tests are in place

## Screenshot

<img width="1164" height="252" alt="image"
src="https://github.com/user-attachments/assets/57df83dc-0e9d-4841-a20f-98a13d1422b3"
/>

## Caveat
While we disable coloring on CLI side for JSON output, you might still
see sth like:
<img width="590" height="499" alt="image"
src="https://github.com/user-attachments/assets/6dc9848c-14c0-49ad-9a25-fe3a186d15e9"
/>

Those ANSI codes come from server and I decided not to alter them and
print as they are. In theory we could add a regex pattern and strip them
out but I believe it's better to print exactly what comes from the
server.

Author: pkosiec

cmd/workspace/apps/logs.go

@@ -19,6 +19,7 @@ import (
 	"github.com/databricks/cli/libs/cmdctx"
 	"github.com/databricks/cli/libs/cmdgroup"
 	"github.com/databricks/cli/libs/cmdio"
+	"github.com/databricks/cli/libs/flags"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/databricks-sdk-go/config"
 	"github.com/databricks/databricks-sdk-go/service/apps"
@@ -142,7 +143,9 @@ via --source APP|SYSTEM. Use --output-file to mirror the stream to a local file
 				defer file.Close()
 				writer = io.MultiWriter(writer, file)
 			}
-			colorizeLogs := outputPath == "" && cmdio.IsTTY(cmd.OutOrStdout())
+
+			outputFormat := root.OutputType(cmd)
+			colorizeLogs := outputPath == "" && outputFormat == flags.OutputText && cmdio.IsTTY(cmd.OutOrStdout())
 
 			sourceMap, err := buildSourceFilter(sourceFilters)
 			if err != nil {
@@ -165,6 +168,7 @@ via --source APP|SYSTEM. Use --output-file to mirror the stream to a local file
 				Writer:           writer,
 				UserAgent:        "databricks-cli apps logs",
 				Colorize:         colorizeLogs,
+				OutputFormat:     outputFormat,
 			})
 		},
 	}

libs/apps/logstream/formatter.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/databricks/cli/libs/flags"
 	"github.com/fatih/color"
 )
 
@@ -27,16 +28,25 @@ func parseLogEntry(raw []byte) (*wsEntry, error) {
 
 // logFormatter formats log entries for output.
 type logFormatter struct {
-	colorize bool
+	colorize     bool
+	outputFormat flags.Output
 }
 
 // newLogFormatter creates a new log formatter.
-func newLogFormatter(colorize bool) *logFormatter {
-	return &logFormatter{colorize: colorize}
+func newLogFormatter(colorize bool, outputFormat flags.Output) *logFormatter {
+	return &logFormatter{colorize: colorize, outputFormat: outputFormat}
 }
 
 // FormatEntry formats a structured log entry for output.
 func (f *logFormatter) FormatEntry(entry *wsEntry) string {
+	if f.outputFormat == flags.OutputJSON {
+		return f.formatEntryJSON(entry)
+	}
+	return f.formatEntryText(entry)
+}
+
+// formatEntryText formats a structured log entry as human-readable text.
+func (f *logFormatter) formatEntryText(entry *wsEntry) string {
 	timestamp := formatTimestamp(entry.Timestamp)
 	source := strings.ToUpper(entry.Source)
 	message := strings.TrimRight(entry.Message, "\r\n")
@@ -49,6 +59,20 @@ func (f *logFormatter) FormatEntry(entry *wsEntry) string {
 	return fmt.Sprintf("%s [%s] %s", timestamp, source, message)
 }
 
+// formatEntryJSON formats a structured log entry as JSON (NDJSON line).
+func (f *logFormatter) formatEntryJSON(entry *wsEntry) string {
+	normalized := wsEntry{
+		Source:    strings.ToUpper(entry.Source),
+		Timestamp: entry.Timestamp,
+		Message:   strings.TrimRight(entry.Message, "\r\n"),
+	}
+	data, err := json.Marshal(normalized)
+	if err != nil {
+		return f.formatEntryText(entry)
+	}
+	return string(data)
+}
+
 // FormatPlain formats a plain text message by trimming trailing newlines.
 func (f *logFormatter) FormatPlain(raw []byte) string {
 	return strings.TrimRight(string(raw), "\r\n")

libs/apps/logstream/formatter_test.go

@@ -0,0 +1,37 @@
+package logstream
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/databricks/cli/libs/flags"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFormatter_FormatEntry(t *testing.T) {
+	entry := &wsEntry{Source: "app", Timestamp: 1705315800.0, Message: "hello world\n"}
+
+	t.Run("json output", func(t *testing.T) {
+		jsonFormatter := newLogFormatter(false, flags.OutputJSON)
+		output := jsonFormatter.FormatEntry(entry)
+
+		var parsed wsEntry
+		require.NoError(t, json.Unmarshal([]byte(output), &parsed))
+
+		assert.Equal(t, "APP", parsed.Source)
+		assert.Greater(t, parsed.Timestamp, 0.0)
+		assert.Equal(t, "hello world", parsed.Message)
+		assert.NotContains(t, output, "\x1b[")
+	})
+
+	t.Run("text output", func(t *testing.T) {
+		textFormatter := newLogFormatter(false, flags.OutputText)
+		output := textFormatter.FormatEntry(entry)
+
+		assert.Contains(t, output, "[APP]")
+		assert.Contains(t, output, "hello world")
+		assert.Contains(t, output, "2024-01-15")
+		assert.NotContains(t, output, "\x1b[")
+	})
+}

libs/apps/logstream/streamer.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/databricks/cli/libs/flags"
 	"github.com/gorilla/websocket"
 )
 
@@ -50,6 +51,7 @@ type Config struct {
 	Writer           io.Writer
 	UserAgent        string
 	Colorize         bool
+	OutputFormat     flags.Output
 }
 
 // Run connects to the log stream described by cfg and copies frames to the writer.
@@ -72,7 +74,7 @@ func Run(ctx context.Context, cfg Config) error {
 		prefetch:         cfg.Prefetch,
 		writer:           cfg.Writer,
 		userAgent:        cfg.UserAgent,
-		formatter:        newLogFormatter(cfg.Colorize),
+		formatter:        newLogFormatter(cfg.Colorize, cfg.OutputFormat),
 	}
 	if streamer.userAgent == "" {
 		streamer.userAgent = defaultUserAgent
@@ -267,7 +269,7 @@ func (s *logStreamer) formatMessage(message []byte) string {
 	source := strings.ToUpper(entry.Source)
 	if len(s.sources) > 0 {
 		if _, ok := s.sources[source]; !ok {
-			return "" // Filtered out
+			return ""
 		}
 	}
 	return s.formatter.FormatEntry(entry)

libs/apps/logstream/streamer_test.go

@@ -14,6 +14,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/databricks/cli/libs/flags"
 	"github.com/fatih/color"
 	"github.com/gorilla/websocket"
 	"github.com/stretchr/testify/assert"
@@ -44,7 +45,7 @@ func TestLogStreamerTailBufferFlushes(t *testing.T) {
 		follow:    false,
 		prefetch:  25 * time.Millisecond,
 		writer:    buf,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	require.NoError(t, streamer.Run(context.Background()))
@@ -77,7 +78,7 @@ func TestLogStreamerTailFlushErrorPropagates(t *testing.T) {
 		follow:    false,
 		prefetch:  0,
 		writer:    &failWriter{err: writerErr},
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	err := streamer.Run(context.Background())
@@ -102,7 +103,7 @@ func TestLogStreamerTrimsCRLFInStructuredEntries(t *testing.T) {
 		url:       toWebSocketURL(server.URL),
 		token:     "token",
 		writer:    buf,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	require.NoError(t, streamer.Run(context.Background()))
@@ -125,7 +126,7 @@ func TestLogStreamerDialErrorIncludesResponseBody(t *testing.T) {
 		url:       toWebSocketURL(server.URL),
 		token:     "test",
 		writer:    &bytes.Buffer{},
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	err := streamer.Run(context.Background())
@@ -152,7 +153,7 @@ func TestLogStreamerRetriesOnDialFailure(t *testing.T) {
 		follow:    true,
 		prefetch:  0,
 		writer:    buf,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
@@ -182,7 +183,7 @@ func TestLogStreamerSendsSearchTerm(t *testing.T) {
 		token:     "test",
 		search:    "ERROR",
 		writer:    buf,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	require.NoError(t, streamer.Run(context.Background()))
@@ -196,7 +197,7 @@ func TestLogStreamerFiltersSources(t *testing.T) {
 		defer conn.Close()
 		_, _, _ = conn.ReadMessage()
 		require.NoError(t, sendEntry(conn, 1, "app"))
-		require.NoError(t, conn.WriteMessage(websocket.TextMessage, mustJSON(wsEntry{Source: "SYSTEM", Timestamp: 2, Message: "sys"})))
+		require.NoError(t, conn.WriteMessage(websocket.TextMessage, mustJSON(t, wsEntry{Source: "SYSTEM", Timestamp: 2, Message: "sys"})))
 		_ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add(time.Second))
 	})
 	defer server.Close()
@@ -210,7 +211,7 @@ func TestLogStreamerFiltersSources(t *testing.T) {
 		token:     "test",
 		sources:   sources,
 		writer:    buf,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	require.NoError(t, streamer.Run(context.Background()))
@@ -226,22 +227,58 @@ func TestFormatLogEntryColorizesWhenEnabled(t *testing.T) {
 
 	entry := &wsEntry{Source: "app", Timestamp: 1, Message: "hello\n"}
 
-	colorFormatter := newLogFormatter(true)
+	colorFormatter := newLogFormatter(true, flags.OutputText)
 	colored := colorFormatter.FormatEntry(entry)
 	assert.Contains(t, colored, "\x1b[")
 	assert.Contains(t, colored, fmt.Sprintf("[%s]", color.HiBlueString("APP")))
 
-	plainFormatter := newLogFormatter(false)
+	plainFormatter := newLogFormatter(false, flags.OutputText)
 	plain := plainFormatter.FormatEntry(entry)
 	assert.NotContains(t, plain, "\x1b[")
 	assert.Contains(t, plain, "[APP]")
 }
 
-func mustJSON(entry wsEntry) []byte {
-	raw, err := json.Marshal(entry)
-	if err != nil {
-		panic(err)
+func TestLogStreamerOutputsNDJSON(t *testing.T) {
+	t.Parallel()
+
+	server := newTestLogServer(t, func(id int, conn *websocket.Conn) {
+		defer conn.Close()
+		_, _, _ = conn.ReadMessage()
+		require.NoError(t, sendEntry(conn, 1.0, "first message"))
+		require.NoError(t, sendEntry(conn, 2.0, "second message"))
+		_ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), time.Now().Add(time.Second))
+	})
+	defer server.Close()
+
+	buf := &bytes.Buffer{}
+	streamer := &logStreamer{
+		dialer:    &websocket.Dialer{},
+		url:       toWebSocketURL(server.URL),
+		token:     "token",
+		writer:    buf,
+		formatter: newLogFormatter(false, flags.OutputJSON),
 	}
+
+	require.NoError(t, streamer.Run(context.Background()))
+
+	lines := strings.Split(strings.TrimSpace(buf.String()), "\n")
+	require.Len(t, lines, 2, "expected two NDJSON lines")
+
+	var entry1 wsEntry
+	require.NoError(t, json.Unmarshal([]byte(lines[0]), &entry1))
+	assert.Equal(t, "APP", entry1.Source)
+	assert.Equal(t, "first message", entry1.Message)
+
+	var entry2 wsEntry
+	require.NoError(t, json.Unmarshal([]byte(lines[1]), &entry2))
+	assert.Equal(t, "APP", entry2.Source)
+	assert.Equal(t, "second message", entry2.Message)
+}
+
+func mustJSON(t *testing.T, entry wsEntry) []byte {
+	raw, err := json.Marshal(entry)
+	require.NoError(t, err)
+
 	return raw
 }
 
@@ -267,7 +304,7 @@ func TestTailWithoutPrefetchRespectsTailSize(t *testing.T) {
 		tail:      2,
 		prefetch:  0,
 		writer:    buf,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	require.NoError(t, streamer.Run(context.Background()))
@@ -291,7 +328,7 @@ func TestCloseErrorPropagatesWhenAbnormal(t *testing.T) {
 		dialer:    &websocket.Dialer{},
 		url:       toWebSocketURL(server.URL),
 		token:     "token",
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	err := streamer.Run(context.Background())
@@ -381,7 +418,7 @@ func TestLogStreamerTailFlushesWithoutFollow(t *testing.T) {
 		follow:    false,
 		prefetch:  50 * time.Millisecond,
 		writer:    writer,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
@@ -422,7 +459,7 @@ func TestLogStreamerFollowTailWithoutPrefetchEmitsRequestedLines(t *testing.T) {
 		follow:    true,
 		prefetch:  0,
 		writer:    writer,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -480,7 +517,7 @@ func TestLogStreamerFollowTailDoesNotReplayAfterReconnect(t *testing.T) {
 		follow:    true,
 		prefetch:  0,
 		writer:    writer,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -561,7 +598,7 @@ func TestLogStreamerRefreshesTokenAfterAuthClose(t *testing.T) {
 		tokenProvider: tokenProvider,
 		follow:        true,
 		writer:        buf,
-		formatter:     newLogFormatter(false),
+		formatter:     newLogFormatter(false, flags.OutputText),
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -600,7 +637,7 @@ func TestLogStreamerEmitsPlainTextFrames(t *testing.T) {
 		url:       toWebSocketURL(server.URL),
 		token:     "token",
 		writer:    buf,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	require.NoError(t, streamer.Run(context.Background()))
@@ -625,7 +662,7 @@ func TestLogStreamerTimeoutStopsQuietFollowStream(t *testing.T) {
 		url:       toWebSocketURL(server.URL),
 		token:     "token",
 		follow:    true,
-		formatter: newLogFormatter(false),
+		formatter: newLogFormatter(false, flags.OutputText),
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
@@ -718,7 +755,7 @@ func TestAppStatusCheckerStopsFollowing(t *testing.T) {
 		follow:           true,
 		writer:           buf,
 		appStatusChecker: appStatusChecker,
-		formatter:        newLogFormatter(false),
+		formatter:        newLogFormatter(false, flags.OutputText),
 	}
 
 	err := streamer.Run(context.Background())