Fix container ignoring max bytes when parsing CRI partial messages (#49743)

GenAI-Assisted: Yes
Human-Reviewed: Yes
Tool: Claude-CLI, Model: Sonet 4.6


---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: belimawr

changelog/fragments/1774630083-fix-container-max-bytes.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Fix container input not respecting max bytes when parsing CRI partial lines
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: filebeat
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: https://github.com/elastic/beats/issues/49259

filebeat/input/log/harvester.go

@@ -529,7 +529,7 @@ func (h *Harvester) openFile() error {
 
 	f, err := file_helper.ReadOpen(h.state.Source)
 	if err != nil {
-		return fmt.Errorf("Failed opening %s: %w", h.state.Source, err)
+		return fmt.Errorf("failed opening %s: %w", h.state.Source, err)
 	}
 
 	harvesterOpenFiles.Add(1)
@@ -551,11 +551,11 @@ func (h *Harvester) validateFile(f *os.File) error {
 
 	info, err := f.Stat()
 	if err != nil {
-		return fmt.Errorf("Failed getting stats for file %s: %w", h.state.Source, err)
+		return fmt.Errorf("failed getting stats for file %s: %w", h.state.Source, err)
 	}
 
 	if !info.Mode().IsRegular() {
-		return fmt.Errorf("Tried to open non regular file: %q %s", info.Mode(), info.Name())
+		return fmt.Errorf("tried to open non regular file: %q %s", info.Mode(), info.Name())
 	}
 
 	// Compares the stat of the opened file to the state given by the input. Abort if not match.
@@ -687,7 +687,7 @@ func (h *Harvester) newLogFileReader() (reader.Reader, error) {
 
 	if h.config.DockerJSON != nil {
 		// Docker json-file format, add custom parsing to the pipeline
-		r = readjson.New(r, h.config.DockerJSON.Stream, h.config.DockerJSON.Partial, h.config.DockerJSON.Format, h.config.DockerJSON.CRIFlags, h.logger)
+		r = readjson.New(r, h.config.DockerJSON.Stream, h.config.DockerJSON.Partial, h.config.DockerJSON.Format, h.config.DockerJSON.CRIFlags, h.config.MaxBytes, h.logger)
 	}
 
 	if h.config.JSON != nil {

libbeat/reader/parser/parser.go

@@ -173,7 +173,7 @@ func (c *Config) Create(in reader.Reader, log *logp.Logger) Parser {
 			if err != nil {
 				return p
 			}
-			p = readjson.NewContainerParser(p, &config, log)
+			p = readjson.NewContainerParser(p, &config, int(c.pCfg.MaxBytes), log)
 		case "syslog":
 			config := syslog.DefaultConfig()
 			cfg := ns.Config()

libbeat/reader/readjson/docker_json.go

@@ -43,6 +43,11 @@ type DockerJSONReader struct {
 	// parse CRI flags
 	criflags bool
 
+	// maximum number of bytes to use when reassembling partial CRI/docker log lines;
+	// limits growth while joining fragments but does not cap the size of the initial chunk.
+	// A value of 0 means no limit is applied during reassembly.
+	maxBytes int
+
 	parseLine func(message *reader.Message, msg *logLine) error
 
 	stripNewLine func(msg *reader.Message)
@@ -60,12 +65,13 @@ type logLine struct {
 }
 
 // New creates a new reader renaming a field
-func New(r reader.Reader, stream string, partial bool, format string, CRIFlags bool, logger *logp.Logger) *DockerJSONReader {
+func New(r reader.Reader, stream string, partial bool, format string, CRIFlags bool, maxBytes int, logger *logp.Logger) *DockerJSONReader {
 	reader := DockerJSONReader{
 		stream:   stream,
 		partial:  partial,
 		reader:   r,
 		criflags: CRIFlags,
+		maxBytes: maxBytes,
 		logger:   logger.Named("reader_docker_json"),
 	}
 
@@ -87,12 +93,13 @@ func New(r reader.Reader, stream string, partial bool, format string, CRIFlags b
 	return &reader
 }
 
-func NewContainerParser(r reader.Reader, config *ContainerJSONConfig, logger *logp.Logger) *DockerJSONReader {
+func NewContainerParser(r reader.Reader, config *ContainerJSONConfig, maxBytes int, logger *logp.Logger) *DockerJSONReader {
 	reader := DockerJSONReader{
 		stream:   config.Stream.String(),
 		partial:  true,
 		reader:   r,
 		criflags: true,
+		maxBytes: maxBytes,
 		logger:   logger.Named("parser_container"),
 	}
 
@@ -233,6 +240,7 @@ func (p *DockerJSONReader) Next() (reader.Message, error) {
 		}
 
 		// Handle multiline messages, join partial lines
+		truncated := false
 		for p.partial && logLine.Partial {
 			next, err := p.reader.Next()
 
@@ -248,7 +256,24 @@ func (p *DockerJSONReader) Next() (reader.Message, error) {
 				p.logger.Errorf("Parse line error: %v", err)
 				continue
 			}
-			message.Content = append(message.Content, next.Content...)
+
+			// Enforce max_bytes during partial line reassembly to prevent unbounded
+			// memory growth. Once the limit is reached, drain remaining partial
+			// chunks (updating the byte counter only) so the reader stays aligned
+			// to logical line boundaries for the next Next() call.
+			if truncated {
+				continue
+			}
+			if p.maxBytes > 0 && len(message.Content)+len(next.Content) > p.maxBytes {
+				remaining := p.maxBytes - len(message.Content)
+				if remaining > 0 {
+					message.Content = append(message.Content, next.Content[:remaining]...)
+				}
+				_ = message.AddFlagsWithKey("log.flags", "truncated")
+				truncated = true
+			} else {
+				message.Content = append(message.Content, next.Content...)
+			}
 		}
 
 		if p.stream != "all" && p.stream != logLine.Stream {

libbeat/reader/readjson/docker_json_test.go

@@ -18,13 +18,15 @@
 package readjson
 
 import (
+	"fmt"
 	"io"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/beats/v7/libbeat/reader"
+	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
@@ -351,7 +353,7 @@ func TestDockerJSON(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			r := &mockReader{messages: test.input}
-			json := New(r, test.stream, test.partial, test.format, test.criflags, logger)
+			json := New(r, test.stream, test.partial, test.format, test.criflags, 0, logger)
 			message, err := json.Next()
 
 			if test.expectedError != nil {
@@ -370,6 +372,41 @@ func TestDockerJSON(t *testing.T) {
 	}
 }
 
+func TestDockerJSONMaxBytes(t *testing.T) {
+	// Simulate many CRI partial chunks that would exceed max_bytes in aggregate.
+	// The reader must truncate the assembled message and drain remaining partials
+	// without allocating unbounded memory.
+	chunkContent := "abcdefghij" // 10 bytes per chunk
+	numChunks := 5
+	maxBytes := 25 // limit is less than 5*10 = 50 bytes
+
+	var inputs [][]byte
+	for i := range numChunks {
+		flag := "P"
+		if i == numChunks-1 {
+			flag = "F"
+		}
+		line := fmt.Sprintf("2017-10-12T13:32:21.232861448Z stdout %s %s", flag, chunkContent)
+		inputs = append(inputs, []byte(line))
+	}
+
+	r := &mockReader{messages: inputs}
+	json := New(r, "stdout", true, "cri", true, maxBytes, logp.NewNopLogger())
+	message, err := json.Next()
+
+	assert.NoError(t, err)
+	assert.Len(t, message.Content, maxBytes, "content should be capped at maxBytes")
+
+	flags, err := message.Fields.GetValue("log.flags")
+	assert.NoError(t, err, "'log.flags' not present in event")
+	assert.Contains(t, flags, "truncated", "truncated flag should be set")
+
+	// All partial chunks up to and including the final F line must have been
+	// consumed by this single Next() call. If any remain, subsequent calls
+	// would emit orphaned partial chunks as separate events, breaking alignment.
+	assert.Empty(t, r.messages, "all partial chunks must be drained before returning")
+}
+
 type mockReader struct {
 	messages [][]byte
 }