Adjust log input and output plugins to support 1MB log lines and resolve issues with truncation logic

Author: TravisStark

plugins/inputs/logfile/constants/constants.go

@@ -0,0 +1,15 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package constants
+
+const (
+	// DefaultMaxEventSize is the default maximum size for log events (1MB)
+	DefaultMaxEventSize = 1024 * 1024
+
+	// PerEventHeaderBytes is the bytes required for metadata for each log event
+	PerEventHeaderBytes = 200
+
+	// DefaultTruncateSuffix is the suffix added to truncated log messages
+	DefaultTruncateSuffix = "[Truncated...]"
+)

plugins/inputs/logfile/fileconfig.go

@@ -18,14 +18,10 @@ import (
 
 	"github.com/aws/amazon-cloudwatch-agent/internal/logscommon"
 	"github.com/aws/amazon-cloudwatch-agent/logs"
+	"github.com/aws/amazon-cloudwatch-agent/plugins/inputs/logfile/constants"
 	"github.com/aws/amazon-cloudwatch-agent/profiler"
 )
 
-const (
-	defaultMaxEventSize   = 1024 * 256 //256KB
-	defaultTruncateSuffix = "[Truncated...]"
-)
-
 // The file config presents the structure of configuration for a file to be tailed.
 type FileConfig struct {
 	//The file path for input log file.
@@ -154,11 +150,10 @@ func (config *FileConfig) init() error {
 	}
 
 	if config.MaxEventSize == 0 {
-		config.MaxEventSize = defaultMaxEventSize
+		config.MaxEventSize = constants.DefaultMaxEventSize - constants.PerEventHeaderBytes
 	}
-
 	if config.TruncateSuffix == "" {
-		config.TruncateSuffix = defaultTruncateSuffix
+		config.TruncateSuffix = constants.DefaultTruncateSuffix
 	}
 	if config.RetentionInDays == 0 {
 		config.RetentionInDays = -1

plugins/inputs/logfile/logfile.go

@@ -86,8 +86,8 @@ const sampleConfig = `
       ## Whether file is a named pipe
       pipe = false
       destination = "cloudwatchlogs"
-      ## Max size of each log event, defaults to 262144 (256KB)
-      max_event_size = 262144
+      ## Max size of each log event, defaults to 1048576 (1MB)
+      max_event_size = 1048576
       ## Suffix to be added to truncated logline to indicate its truncation, defaults to "[Truncated...]"
       truncate_suffix = "[Truncated...]"
 

plugins/inputs/logfile/tail/tail.go

@@ -18,6 +18,7 @@ import (
 	"gopkg.in/tomb.v1"
 
 	"github.com/aws/amazon-cloudwatch-agent/internal/state"
+	"github.com/aws/amazon-cloudwatch-agent/plugins/inputs/logfile/constants"
 	"github.com/aws/amazon-cloudwatch-agent/plugins/inputs/logfile/tail/watch"
 )
 
@@ -306,8 +307,12 @@ func (tail *Tail) readlineUtf16() (string, error) {
 			}
 			cur = append(cur, nextByte)
 		}
-		// 262144 => 256KB
-		if resSize+len(cur) >= 262144 {
+		// Use MaxLineSize if configured, otherwise use DefaultMaxEventSize
+		maxSize := constants.DefaultMaxEventSize // Use the constant defined in constants package
+		if tail.MaxLineSize > 0 {
+			maxSize = tail.MaxLineSize
+		}
+		if resSize+len(cur) >= maxSize {
 			break
 		}
 		buf := make([]byte, len(cur))
@@ -534,8 +539,7 @@ func (tail *Tail) waitForChanges() error {
 func (tail *Tail) openReader() {
 	tail.lk.Lock()
 	if tail.MaxLineSize > 0 {
-		// add 2 to account for newline characters
-		tail.reader = bufio.NewReaderSize(tail.file, tail.MaxLineSize+2)
+		tail.reader = bufio.NewReaderSize(tail.file, tail.MaxLineSize)
 	} else {
 		tail.reader = bufio.NewReader(tail.file)
 	}

plugins/inputs/logfile/tail/tail_test.go

@@ -7,11 +7,13 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 const linesWrittenToFile int = 10
@@ -197,3 +199,100 @@ func tearDown(tmpfile *os.File) {
 	exitOnDeletionCheckDuration = time.Minute
 	exitOnDeletionWaitDuration = 5 * time.Minute
 }
+
+func TestUtf16LineSize(t *testing.T) {
+	tmpfile, err := os.CreateTemp("", "")
+	require.NoError(t, err)
+	defer os.Remove(tmpfile.Name())
+
+	// Create a UTF-16 BOM
+	_, err = tmpfile.Write([]byte{0xFE, 0xFF})
+	require.NoError(t, err)
+
+	// Create a tail with a small MaxLineSize
+	maxLineSize := 100
+	tail, err := TailFile(tmpfile.Name(), Config{
+		MaxLineSize: maxLineSize,
+		Follow:      true,
+		ReOpen:      false,
+		Poll:        true,
+	})
+	require.NoError(t, err)
+	defer tail.Stop()
+
+	// Write a UTF-16 encoded line that exceeds MaxLineSize when decoded
+	// Each 'a' will be 2 bytes in UTF-16
+	utf16Line := make([]byte, 0, maxLineSize*4)
+	for i := 0; i < maxLineSize*2; i++ {
+		utf16Line = append(utf16Line, 0x00, 'a')
+	}
+	utf16Line = append(utf16Line, 0x00, '\n')
+
+	_, err = tmpfile.Write(utf16Line)
+	require.NoError(t, err)
+	err = tmpfile.Sync()
+	require.NoError(t, err)
+
+	// Read the line and verify it's truncated
+	select {
+	case line := <-tail.Lines:
+		// The line should be truncated to maxLineSize
+		assert.LessOrEqual(t, len(line.Text), maxLineSize)
+	case <-time.After(1 * time.Second):
+		t.Fatal("timeout waiting for line")
+	}
+}
+
+func TestTail_DefaultBuffer(t *testing.T) {
+	// Test that default buffer works with normal-sized log lines
+	tempDir := t.TempDir()
+	filename := filepath.Join(tempDir, "test.log")
+
+	// Create a file with a normal-sized line (1KB - well within default buffer)
+	normalContent := strings.Repeat("b", 1024) // 1KB
+	err := os.WriteFile(filename, []byte(normalContent+"\n"), 0600)
+	require.NoError(t, err)
+
+	tail, err := TailFile(filename, Config{
+		Follow:    false,
+		MustExist: true,
+		// MaxLineSize not set - should use default buffer
+	})
+	require.NoError(t, err)
+	defer tail.Stop()
+
+	select {
+	case line := <-tail.Lines:
+		assert.NoError(t, line.Err)
+		assert.Equal(t, normalContent, line.Text)
+	case <-time.After(time.Second):
+		t.Fatal("Timeout waiting for line")
+	}
+}
+
+func TestTail_1MBWithExplicitMaxLineSize(t *testing.T) {
+	// Test that large lines work when MaxLineSize is explicitly set
+	tempDir := t.TempDir()
+	filename := filepath.Join(tempDir, "test.log")
+
+	// Create a file with a 512KB line
+	largeContent := strings.Repeat("b", 512*1024) // 512KB
+	err := os.WriteFile(filename, []byte(largeContent+"\n"), 0600)
+	require.NoError(t, err)
+
+	tail, err := TailFile(filename, Config{
+		Follow:      false,
+		MustExist:   true,
+		MaxLineSize: 1024 * 1024, // Explicitly set 1MB buffer
+	})
+	require.NoError(t, err)
+	defer tail.Stop()
+
+	select {
+	case line := <-tail.Lines:
+		assert.NoError(t, line.Err)
+		assert.Equal(t, largeContent, line.Text)
+	case <-time.After(time.Second):
+		t.Fatal("Timeout waiting for line")
+	}
+}

plugins/inputs/logfile/tailersrc.go

@@ -238,10 +238,6 @@ func (ts *tailerSrc) runTail() {
 			} else {
 				msgBuf.WriteString("\n")
 				msgBuf.WriteString(text)
-				if msgBuf.Len() > ts.maxEventSize {
-					msgBuf.Truncate(ts.maxEventSize - len(ts.truncateSuffix))
-					msgBuf.WriteString(ts.truncateSuffix)
-				}
 				fo.ShiftInt64(line.Offset)
 				continue
 			}