Fix handling of multiplexed long lines in Docker logs (#4713)

* Improve docker's multiplexed long lines handling

* Fix `dockerChunkWriter` buffer flushing and add test for it

* Update `CHANGELOG.md`

* Fix potential panic in `extractTsFromBytes` and `dockerChunkWriter` buffer allocation

Author: axd1x8a

CHANGELOG.md

@@ -58,6 +58,8 @@ Main (unreleased)
 
 - Remove extraneous `output` stage from the `cri` stage pipeline in `loki.process`. (@kalleep)
 
+- Fix Docker log corruption for multiplexed long lines. (@axd1x8a)
+
 v1.12.0
 -----------------
 

internal/component/loki/source/docker/internal/dockertarget/target.go

@@ -6,13 +6,15 @@ package dockertarget
 
 import (
 	"bufio"
+	"bytes"
 	"context"
 	"fmt"
 	"io"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
+	"unsafe"
 
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/client"
@@ -34,6 +36,7 @@ const (
 	dockerLabel                = model.MetaLabelPrefix + "docker_"
 	dockerLabelContainerPrefix = dockerLabel + "container_"
 	dockerLabelLogStream       = dockerLabelContainerPrefix + "log_stream"
+	dockerMaxChunkSize         = 16384
 )
 
 // Target enables reading Docker container logs.
@@ -110,7 +113,12 @@ func (t *Target) processLoop(ctx context.Context, tty bool, reader io.ReadCloser
 		if tty {
 			written, err = io.Copy(wstdout, reader)
 		} else {
-			written, err = stdcopy.StdCopy(wstdout, wstderr, reader)
+			// For non-TTY, wrap the pipe writers with our chunk writer to reassemble frames.
+			wcstdout := newChunkWriter(wstdout, t.logger)
+			defer wcstdout.Close()
+			wcstderr := newChunkWriter(wstderr, t.logger)
+			defer wcstderr.Close()
+			written, err = stdcopy.StdCopy(wcstdout, wcstderr, reader)
 		}
 		if err != nil {
 			level.Warn(t.logger).Log("msg", "could not transfer logs", "written", written, "container", t.containerName, "err", err)
@@ -128,53 +136,38 @@ func (t *Target) processLoop(ctx context.Context, tty bool, reader io.ReadCloser
 	level.Debug(t.logger).Log("msg", "done processing Docker logs", "container", t.containerName)
 }
 
-// extractTs tries for read the timestamp from the beginning of the log line.
-// It's expected to follow the format 2006-01-02T15:04:05.999999999Z07:00.
-func extractTs(line string) (time.Time, string, error) {
-	pair := strings.SplitN(line, " ", 2)
-	if len(pair) != 2 {
-		return time.Now(), line, fmt.Errorf("could not find timestamp in '%s'", line)
-	}
-	ts, err := time.Parse("2006-01-02T15:04:05.999999999Z07:00", pair[0])
-	if err != nil {
-		return time.Now(), line, fmt.Errorf("could not parse timestamp from '%s': %w", pair[0], err)
-	}
-	return ts, pair[1], nil
-}
+// extractTsFromBytes parses an RFC3339Nano timestamp from the byte slice.
+func extractTsFromBytes(line []byte) (time.Time, []byte, error) {
+	const timestampLayout = "2006-01-02T15:04:05.999999999Z07:00"
 
-// https://devmarkpro.com/working-big-files-golang
-func readLine(r *bufio.Reader) (string, error) {
-	var (
-		isPrefix = true
-		err      error
-		line, ln []byte
-	)
-
-	for isPrefix && err == nil {
-		line, isPrefix, err = r.ReadLine()
-		ln = append(ln, line...)
+	spaceIdx := bytes.IndexByte(line, ' ')
+	if spaceIdx == -1 || spaceIdx >= len(line)-1 {
+		return time.Time{}, nil, fmt.Errorf("could not find timestamp in bytes")
 	}
 
-	return string(ln), err
+	// The unsafe.String is used here to avoid allocation and string conversion when parsing the timestamp
+	// This is safe because:
+	// 1. spaceIdx > 0 and spaceIdx < len(line)-1 is guaranteed by the check above
+	// 2. time.Parse doesn't retain the string after returning
+	// 3. The underlying bytes aren't modified during parsing
+	ts, err := time.Parse(timestampLayout, unsafe.String(&line[0], spaceIdx))
+	if err != nil {
+		return time.Time{}, nil, fmt.Errorf("could not parse timestamp: %w", err)
+	}
+	return ts, line[spaceIdx+1:], nil
 }
 
 func (t *Target) process(r io.Reader, logStreamLset model.LabelSet) {
-	defer func() {
-		t.wg.Done()
-	}()
+	defer t.wg.Done()
 
-	reader := bufio.NewReader(r)
-	for {
-		line, err := readLine(reader)
-		if err != nil {
-			if err == io.EOF {
-				break
-			}
-			level.Error(t.logger).Log("msg", "error reading docker log line, skipping line", "err", err)
-			t.metrics.dockerErrors.Inc()
-		}
+	scanner := bufio.NewScanner(r)
+	const maxCapacity = dockerMaxChunkSize * 64
+	buf := make([]byte, 0, maxCapacity)
+	scanner.Buffer(buf, maxCapacity)
+	for scanner.Scan() {
+		line := scanner.Bytes()
 
-		ts, line, err := extractTs(line)
+		ts, content, err := extractTsFromBytes(line)
 		if err != nil {
 			level.Error(t.logger).Log("msg", "could not extract timestamp, skipping line", "err", err)
 			t.metrics.dockerErrors.Inc()
@@ -185,7 +178,7 @@ func (t *Target) process(r io.Reader, logStreamLset model.LabelSet) {
 			Labels: logStreamLset,
 			Entry: push.Entry{
 				Timestamp: ts,
-				Line:      line,
+				Line:      string(content),
 			},
 		}
 		t.metrics.dockerEntries.Inc()
@@ -200,6 +193,10 @@ func (t *Target) process(r io.Reader, logStreamLset model.LabelSet) {
 		t.since.Store(ts.Unix())
 		t.last.Store(time.Now().Unix())
 	}
+	if err := scanner.Err(); err != nil {
+		level.Error(t.logger).Log("msg", "error reading docker log line", "err", err)
+		t.metrics.dockerErrors.Inc()
+	}
 }
 
 // StartIfNotRunning starts processing container logs. The operation is idempotent , i.e. the processing cannot be started twice.
@@ -321,3 +318,66 @@ func (t *Target) getStreamLabels(logStream string) model.LabelSet {
 
 	return filtered
 }
+
+// dockerChunkWriter implements io.Writer to preprocess and reassemble Docker log frames.
+type dockerChunkWriter struct {
+	writer      io.Writer
+	logger      log.Logger
+	buffer      *bytes.Buffer
+	isBuffering bool
+}
+
+var bufferPool = sync.Pool{
+	New: func() interface{} {
+		return bytes.NewBuffer(make([]byte, 0, dockerMaxChunkSize*2))
+	},
+}
+
+func newChunkWriter(writer io.Writer, logger log.Logger) *dockerChunkWriter {
+	return &dockerChunkWriter{
+		writer: writer,
+		logger: logger,
+		buffer: bufferPool.Get().(*bytes.Buffer),
+	}
+}
+
+func (fw *dockerChunkWriter) Close() error {
+	if fw.buffer != nil {
+		fw.buffer.Reset()
+		bufferPool.Put(fw.buffer)
+		fw.buffer = nil
+	}
+	return nil
+}
+
+func (fw *dockerChunkWriter) Write(p []byte) (int, error) {
+	if !fw.isBuffering {
+		if len(p) < dockerMaxChunkSize || p[len(p)-1] == 0x0A {
+			// Short or complete frame: write directly without buffering.
+			return fw.writer.Write(p)
+		}
+		// Long frame start: buffer the first chunk.
+		fw.buffer.Write(p)
+		fw.isBuffering = true
+		return len(p), nil
+	}
+
+	// Continuation chunk: strip redundant timestamp and append content.
+	_, content, err := extractTsFromBytes(p)
+	if err != nil {
+		// Should not normally happen, but flog.log has entries like this for some reason.
+		level.Warn(fw.logger).Log("msg", "could not strip timestamp from continuation chunk", "err", err)
+		fw.buffer.Write(p)
+	} else {
+		fw.buffer.Write(content)
+	}
+	// If this is the last continuation chunk (ends with newline), flush the buffer
+	if len(p) > 0 && p[len(p)-1] == 0x0A {
+		if _, err := fw.writer.Write(fw.buffer.Bytes()); err != nil {
+			return 0, err
+		}
+		fw.buffer.Reset()
+		fw.isBuffering = false
+	}
+	return len(p), nil
+}

internal/component/loki/source/docker/internal/dockertarget/target_test.go

@@ -5,6 +5,7 @@ package dockertarget
 // read logs from Docker containers and forward them to other loki components.
 
 import (
+	"bytes"
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
@@ -136,6 +137,44 @@ func TestStartStopStressTest(t *testing.T) {
 	wg.Wait()
 }
 
+func TestDockerChunkWriter(t *testing.T) {
+	logger := log.NewNopLogger()
+	var buf bytes.Buffer
+	writer := newChunkWriter(&buf, logger)
+
+	timestamp := []byte("2023-12-09T12:00:00.000000000Z ")
+	shortLine := []byte("short log line\n")
+
+	var longContent []byte
+	for range 50 * 1024 {
+		longContent = append(longContent, 'a')
+	}
+	longContent = append(longContent, '\n')
+
+	// First part of long line
+	chunk1 := append(timestamp, longContent[:32*1024]...)
+	_, err := writer.Write(chunk1)
+	require.NoError(t, err)
+
+	// Second part of long line
+	chunk2 := append(timestamp, longContent[32*1024:]...)
+	_, err = writer.Write(chunk2)
+	require.NoError(t, err)
+
+	// Start a new short line
+	chunk3 := append(timestamp, shortLine...)
+	_, err = writer.Write(chunk3)
+	require.NoError(t, err)
+
+	err = writer.Close()
+	require.NoError(t, err)
+
+	expected := append(timestamp, longContent...)
+	expected = append(expected, chunk3...)
+
+	assert.Equal(t, expected, buf.Bytes())
+}
+
 func newDockerServer(t *testing.T) *httptest.Server {
 	h := func(w http.ResponseWriter, r *http.Request) {
 		path := r.URL.Path