fix(logs): strip injected timestamps from >16KB Docker TTY container log lines (#51024)

## Summary

- Fixes AGNTLOG-81: Docker `--tty` containers with log lines >16KB had a timestamp marker injected into the log payload every 16KB by the Docker daemon (upstream moby/moby#19696).
- The root cause: Docker injects a `TIMESTAMP SPACE` prefix before each 16KB buffer chunk in TTY mode. For multi-chunk log lines the intermediate timestamps appeared verbatim in the delivered message.
- Fix: added `removeTTYPartialTimestamps` in `pkg/logs/internal/parsers/dockerstream/docker_stream.go`, mirroring the existing `removePartialDockerMetadata` used for the non-TTY path but without the 8-byte stream header.
- Added two unit tests (`TestDockerStandaloneParserShouldHandleLargeTtyMessage`, `TestDockerStandaloneParserShouldRemoveTTYPartialTimestamps`) that cover single-chunk (exactly 16KB+) and multi-chunk (16KB+50, 3×16KB+50) cases.

## Manual QA

**Environment**: macOS arm64, Docker Engine 29.3.1, colima VM backend.

**Reproduced the bug before the fix:**

```bash
# Start a TTY container generating 20,000-char log lines
docker run --rm -d --name tty-test --tty alpine sh -c \
  'while true; do dd if=/dev/zero bs=20000 count=1 2>/dev/null | tr "\0" "A"; echo ""; sleep 3; done'

# Capture raw Docker API stream with timestamps
curl -sN --unix-socket ~/.colima/default/docker.sock \
  "http://./containers/tty-test/logs?stdout=1&stderr=1&timestamps=1" > /tmp/all_logs.bin
```

Python verification of raw data showed:
```
Raw first line length: 20062 bytes
Embedded timestamp at offset 16384 in content (BEFORE fix)
```
Content was 20031 bytes (20000 'A's + 31-byte embedded timestamp `2026-05-19T14:12:36.172399096Z `) — confirming the bug.

**Verified the fix on the same raw data:**

Running `removeTTYPartialTimestamps` on the captured data:
```
After fix:
  Timestamp: 2026-05-19T14:12:36.172399096Z
  Content length: 20000
  Content is clean (all 'A' chars): YES
  Embedded timestamp removed: YES
```

The 31-byte embedded timestamp is stripped; content is exactly 20000 clean bytes.

**Unit tests:** `go test -tags docker ./pkg/logs/internal/parsers/dockerstream/...` — all pass, including 2 new TTY-specific tests.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: brian.floersch <brian.floersch@datadoghq.com>

Author: gh123man

pkg/logs/internal/parsers/dockerstream/docker_stream.go

@@ -12,6 +12,7 @@ package dockerstream
 import (
 	"bytes"
 	"fmt"
+	"time"
 
 	pkgconfigsetup "github.com/DataDog/datadog-agent/pkg/config/setup"
 	"github.com/DataDog/datadog-agent/pkg/logs/internal/parsers"
@@ -66,19 +67,32 @@ func parseDockerStream(msg *message.Message, containerID string) (*message.Messa
 		return msg, fmt.Errorf("cannot parse docker message for container %v: expected a 8 bytes header", containerID)
 	}
 
-	// Read the first byte to get the status
+	// Read the first byte to get the status. Non-TTY containers prefix every
+	// chunk with an 8-byte header where byte 0 is 1 (stdout) or 2 (stderr);
+	// TTY containers omit this header entirely (the daemon emits raw
+	// "RFC3339Nano SPACE content" instead). getDockerSeverity returns "" for
+	// the TTY case, which is the gate for everything below.
 	status := getDockerSeverity(content)
 	if status == "" {
-
-		// When tailing logs coming from a container running with a tty, docker
-		// does not add the header. In that case, the message only contains
-		// the timestamp followed by whatever comes from what is running in the
-		// container (and maybe stdin). As a fallback, set the status to info.
+		// TTY path — no 8-byte header. The body looks like:
+		//   RFC3339Nano SPACE content...
+		// and for log lines spanning more than one 16KB Docker buffer:
+		//   TS1 SPACE content1(16KB) TS2 SPACE content2(16KB) ... TSn SPACE contentn
+		// (see https://github.com/moby/moby/issues/19696). When this happens
+		// we need to strip the intermediate TSi SPACE markers so the caller
+		// only sees the concatenated raw content. Spaces inside the user's
+		// log content are preserved — the stripping operates on fixed 16KB
+		// chunk boundaries, not on space lookups inside content.
 		status = message.StatusInfo
+		if len(content) > dockerBufferSize {
+			content = removeTTYPartialTimestamps(content)
+		}
 
 	} else {
-
-		// remove partial headers that are added by docker when the message gets too long
+		// Non-TTY path — 8-byte stream header present. Each 16KB partial chunk
+		// includes its own header + RFC3339Nano timestamp + space, handled by
+		// removePartialDockerMetadata. The new TTY helper above is never
+		// reached on this branch.
 		if len(content) > dockerBufferSize {
 			content = removePartialDockerMetadata(content)
 		}
@@ -176,6 +190,76 @@ func getDockerMetadataLength(msg []byte) int {
 	return dockerHeaderLength + idx + 1
 }
 
+// removeTTYPartialTimestamps removes the timestamp and space that Docker injects
+// before each 16KB chunk of content when a container runs in TTY mode.
+//
+// In TTY mode the 8-byte stream header is absent, so the wire format for a
+// message that spans multiple 16KB buffers is:
+//
+//	TS1 SPACE CONTENT1(16KB) TS2 SPACE CONTENT2(16KB) ... TSn SPACE CONTENTn
+//
+// This function keeps the very first "TS1 SPACE" and concatenates each
+// subsequent content chunk without its leading timestamp prefix:
+//
+//	Input:  TS1 SPACE CONTENT1 TS2 SPACE CONTENT2 TS3 SPACE CONTENT3
+//	Output: TS1 SPACE CONTENT1 CONTENT2 CONTENT3
+//
+// If the tail after a chunk does not begin with a valid RFC3339Nano timestamp
+// followed by a space (for example a single non-Docker frame longer than 16KB,
+// or a truncated frame), the remainder is appended verbatim instead of being
+// reinterpreted as another chunk boundary.
+func removeTTYPartialTimestamps(msgToClean []byte) []byte {
+	metadataLen, ok := getTTYMetadataLength(msgToClean)
+	if !ok {
+		return msgToClean
+	}
+
+	msg := []byte{}
+	start := 0
+	end := min(len(msgToClean), dockerBufferSize+metadataLen)
+
+	for end > 0 {
+		msg = append(msg, msgToClean[start:end]...)
+		msgToClean = msgToClean[end:]
+		if len(msgToClean) == 0 {
+			break
+		}
+		metadataLen, ok = getTTYMetadataLength(msgToClean)
+		if !ok {
+			// Remainder doesn't look like another Docker chunk boundary.
+			// Append the unmatched tail unchanged rather than stripping
+			// bytes that might be part of the user's log content.
+			msg = append(msg, msgToClean...)
+			break
+		}
+		start = metadataLen
+		end = min(len(msgToClean), dockerBufferSize+metadataLen)
+	}
+
+	return msg
+}
+
+// getTTYMetadataLength returns the length of the timestamp and trailing space
+// that Docker prepends to each TTY-mode buffer chunk. In TTY mode there is no
+// 8-byte stream header, so the metadata is just the RFC3339Nano timestamp
+// followed by a single space character.
+//
+// Returns (metadataLen, true) when the message begins with a valid
+// RFC3339Nano timestamp + space, and (0, false) otherwise. The second
+// return guards against treating ordinary user content (which may contain
+// spaces but does not start with a parseable timestamp) as a chunk
+// boundary.
+func getTTYMetadataLength(msg []byte) (int, bool) {
+	idx := bytes.Index(msg, []byte{' '})
+	if idx == -1 {
+		return 0, false
+	}
+	if _, err := time.Parse(time.RFC3339Nano, string(msg[:idx])); err != nil {
+		return 0, false
+	}
+	return idx + 1, true
+}
+
 // isEmptyMessage tests if the entire message is in the form of escaped new line
 // i.e. \\n  or \\r or \\r\\n
 func isEmptyMessage(content []byte) bool {

pkg/logs/internal/parsers/dockerstream/docker_stream_test.go

@@ -165,3 +165,131 @@ func buildPartialMessage(r rune, count int) string {
 func buildMessage(r rune, count int) string {
 	return strings.Repeat(string(r), count)
 }
+
+// ttyTimestamp is a sample RFC3339Nano timestamp used in TTY-mode tests.
+var ttyTimestamp = "2018-06-14T18:27:03.246999277Z"
+
+// buildTTYPartialMessage returns a TTY-mode chunk as Docker would send it:
+// a bare RFC3339Nano timestamp followed by a space and count repetitions of r.
+// There is no 8-byte stream header.
+func buildTTYPartialMessage(r rune, count int) string {
+	return ttyTimestamp + " " + strings.Repeat(string(r), count)
+}
+
+func TestDockerStandaloneParserShouldHandleLargeTtyMessage(t *testing.T) {
+	// A single 16KB TTY log line: no intermediate timestamp injection expected.
+	input := buildTTYPartialMessage('a', dockerBufferSize)
+	logMessage := message.NewMessage([]byte(input), nil, "", 0)
+	msg, err := container1Parser.Parse(logMessage)
+	assert.Nil(t, err)
+	assert.False(t, msg.ParsingExtra.IsPartial)
+	assert.Equal(t, ttyTimestamp, msg.ParsingExtra.Timestamp)
+	assert.Equal(t, message.StatusInfo, msg.Status)
+	assert.Equal(t, []byte(buildMessage('a', dockerBufferSize)), msg.GetContent())
+	assert.Equal(t, dockerBufferSize, len(msg.GetContent()))
+}
+
+func TestDockerStandaloneParserShouldRemoveTTYPartialTimestamps(t *testing.T) {
+	// A TTY log line >16KB: Docker prepends a timestamp to each 16KB chunk.
+	// The parser must strip the intermediate timestamps so users see clean content.
+
+	// two chunks: 16KB + 50 bytes
+	input := buildTTYPartialMessage('a', dockerBufferSize) + buildTTYPartialMessage('b', 50)
+	expectedContent := buildMessage('a', dockerBufferSize) + buildMessage('b', 50)
+	logMessage := message.NewMessage([]byte(input), nil, "", 0)
+	msg, err := container1Parser.Parse(logMessage)
+	assert.Nil(t, err)
+	assert.False(t, msg.ParsingExtra.IsPartial)
+	assert.Equal(t, ttyTimestamp, msg.ParsingExtra.Timestamp)
+	assert.Equal(t, message.StatusInfo, msg.Status)
+	assert.Equal(t, []byte(expectedContent), msg.GetContent())
+	assert.Equal(t, dockerBufferSize+50, len(msg.GetContent()))
+
+	// three full 16KB chunks + a 50-byte tail (mirrors the non-TTY test)
+	input = buildTTYPartialMessage('a', dockerBufferSize) +
+		buildTTYPartialMessage('a', dockerBufferSize) +
+		buildTTYPartialMessage('a', dockerBufferSize) +
+		buildTTYPartialMessage('b', 50)
+	expectedContent = buildMessage('a', 3*dockerBufferSize) + buildMessage('b', 50)
+	logMessage.SetContent([]byte(input))
+	msg, err = container1Parser.Parse(logMessage)
+	assert.Nil(t, err)
+	assert.False(t, msg.ParsingExtra.IsPartial)
+	assert.Equal(t, ttyTimestamp, msg.ParsingExtra.Timestamp)
+	assert.Equal(t, message.StatusInfo, msg.Status)
+	assert.Equal(t, []byte(expectedContent), msg.GetContent())
+	assert.Equal(t, 3*dockerBufferSize+50, len(msg.GetContent()))
+}
+
+// TestDockerStandaloneParserTTYPreservesSpacesInContent guards against the
+// theoretical risk that removeTTYPartialTimestamps could misinterpret a space
+// inside the user's log content as a chunk-boundary timestamp marker. The
+// stripping operates on fixed 16KB chunk windows, not on space lookups inside
+// content, so embedded spaces must round-trip untouched.
+func TestDockerStandaloneParserTTYPreservesSpacesInContent(t *testing.T) {
+	// Build a 16KB chunk whose content contains a space NOT at the boundary.
+	// Layout: 8000 'a' + 1 space + 8383 'b' = 16384 bytes (= dockerBufferSize).
+	chunk1Content := strings.Repeat("a", 8000) + " " + strings.Repeat("b", 8383)
+	assert.Equal(t, dockerBufferSize, len(chunk1Content))
+
+	// Second chunk: shorter, with its own embedded space.
+	chunk2Content := "hello world tail"
+
+	input := ttyTimestamp + " " + chunk1Content + ttyTimestamp + " " + chunk2Content
+	expectedContent := chunk1Content + chunk2Content
+
+	logMessage := message.NewMessage([]byte(input), nil, "", 0)
+	msg, err := container1Parser.Parse(logMessage)
+	assert.Nil(t, err)
+	assert.False(t, msg.ParsingExtra.IsPartial)
+	assert.Equal(t, ttyTimestamp, msg.ParsingExtra.Timestamp)
+	assert.Equal(t, message.StatusInfo, msg.Status)
+	assert.Equal(t, []byte(expectedContent), msg.GetContent())
+	// Confirm the embedded boundary survived — would be lost if the stripper
+	// used space-search inside content instead of fixed 16KB windows.
+	assert.Equal(t, dockerBufferSize+len(chunk2Content), len(msg.GetContent()))
+	assert.Contains(t, string(msg.GetContent()), "aaaa b")
+	assert.Contains(t, string(msg.GetContent()), "hello world tail")
+}
+
+// TestDockerStandaloneParserTTYPreservesUnmatchedTail guards against the
+// codex review finding: when the tail after the first 16KB chunk does NOT
+// begin with a valid RFC3339Nano timestamp + space (for example a single
+// non-Docker frame slightly longer than 16KB, or a truncated frame), the
+// stripper must not drop the tail or misinterpret content-internal spaces
+// as a chunk boundary. The previous (loose) version returned 0 for tails
+// without a space and silently truncated the message.
+func TestDockerStandaloneParserTTYPreservesUnmatchedTail(t *testing.T) {
+	// (a) Tail with no space at all: previous code dropped it entirely.
+	chunk1 := strings.Repeat("a", dockerBufferSize)
+	tail := "tail-without-space" // no space, no TS, must be preserved verbatim
+	input := ttyTimestamp + " " + chunk1 + tail
+	expectedContent := chunk1 + tail
+
+	logMessage := message.NewMessage([]byte(input), nil, "", 0)
+	msg, err := container1Parser.Parse(logMessage)
+	assert.Nil(t, err)
+	assert.False(t, msg.ParsingExtra.IsPartial)
+	assert.Equal(t, ttyTimestamp, msg.ParsingExtra.Timestamp)
+	assert.Equal(t, message.StatusInfo, msg.Status)
+	assert.Equal(t, []byte(expectedContent), msg.GetContent())
+	assert.Equal(t, dockerBufferSize+len(tail), len(msg.GetContent()))
+
+	// (b) Tail with a space but no valid timestamp before it: previous code
+	// would strip the prefix up to the space as if it were a TS marker.
+	tail = "hello world, no timestamp prefix"
+	input = ttyTimestamp + " " + chunk1 + tail
+	expectedContent = chunk1 + tail
+
+	logMessage = message.NewMessage([]byte(input), nil, "", 0)
+	msg, err = container1Parser.Parse(logMessage)
+	assert.Nil(t, err)
+	assert.False(t, msg.ParsingExtra.IsPartial)
+	assert.Equal(t, ttyTimestamp, msg.ParsingExtra.Timestamp)
+	assert.Equal(t, message.StatusInfo, msg.Status)
+	assert.Equal(t, []byte(expectedContent), msg.GetContent())
+	assert.Equal(t, dockerBufferSize+len(tail), len(msg.GetContent()))
+	// Verify the leading "hello" of the tail is NOT stripped (would happen
+	// if the function treated the first space as a chunk-boundary marker).
+	assert.Contains(t, string(msg.GetContent()), "hello world, no timestamp prefix")
+}

releasenotes/notes/fix-docker-tty-large-log-framing-b21e8c4f7d9a6051.yaml

@@ -0,0 +1,15 @@
+# Each section from every release note are combined when the
+# CHANGELOG.rst is rendered. So the text needs to be worded so that
+# it does not depend on any information only available in another
+# section. This may mean repeating some details, but each section
+# must be readable independently of the other.
+#
+# Each section note must be formatted as reStructuredText.
+---
+fixes:
+  - |
+    Fix Docker log parsing for TTY-mode containers when a single log line exceeds the
+    16KB Docker buffer size. Previously, the Agent retained the per-chunk timestamp
+    prefix that Docker inserts at every 16KB boundary, causing those timestamps to
+    appear inside the collected log content. The parser now strips the intermediate
+    timestamps so the log line is reassembled correctly.