Zero-allocation optimization for sendLine and readLine

plugins/inputs/logfile/tail/tail.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"unsafe"
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/models"
@@ -37,11 +38,6 @@ type Line struct {
 	Offset int64 // offset of current reader
 }
 
-// NewLine returns a Line with present time.
-func NewLine(text string, offset int64) *Line {
-	return &Line{text, time.Now(), nil, offset}
-}
-
 // SeekInfo represents arguments to `os.Seek`
 type SeekInfo struct {
 	Offset int64
@@ -91,6 +87,8 @@ type Tail struct {
 	lk sync.Mutex
 
 	FileDeletedCh chan struct{}
+
+	linePool sync.Pool
 }
 
 // TailFile begins tailing the file. Output stream is made available
@@ -107,6 +105,9 @@ func TailFile(filename string, config Config) (*Tail, error) {
 		Lines:         make(chan *Line),
 		Config:        config,
 		FileDeletedCh: make(chan struct{}),
+		linePool: sync.Pool{
+			New: func() any { return &Line{} },
+		},
 	}
 
 	// when Logger was not specified in config, create new one
@@ -260,7 +261,7 @@ func (tail *Tail) readLine() (string, error) {
 		}
 		line = line[:len(line)-drop]
 	}
-	return string(line), err
+	return unsafe.String(unsafe.SliceData(line), len(line)), err
 }
 
 func (tail *Tail) readlineUtf16() (string, error) {
@@ -418,7 +419,13 @@ func (tail *Tail) tailFileSync() {
 				// Wait a second before seeking till the end of
 				// file when rate limit is reached.
 				msg := "Too much log activity; waiting a second before resuming tailing"
-				tail.Lines <- &Line{msg, time.Now(), errors.New(msg), tail.curOffset}
+				// Warning: Make sure to release line once done!
+				lineObject := tail.linePool.Get().(*Line)
+				lineObject.Text = msg
+				lineObject.Time = time.Now()
+				lineObject.Err = errors.New(msg)
+				lineObject.Offset = tail.curOffset
+				tail.Lines <- lineObject
 				select {
 				case <-time.After(time.Second):
 				case <-tail.Dying():
@@ -574,12 +581,18 @@ func (tail *Tail) sendLine(line string, offset int64) bool {
 
 	for i, line := range lines {
 		// This select is to avoid blockage on the tail.Lines chan
+		// Warning: Make sure to release line once done!
+		lineObject := tail.linePool.Get().(*Line)
+		lineObject.Text = line
+		lineObject.Time = now
+		lineObject.Err = nil
+		lineObject.Offset = offset
 		select {
-		case tail.Lines <- &Line{line, now, nil, offset}:
+		case tail.Lines <- lineObject:
 		case <-tail.Dying():
 			if tail.Err() == errStopAtEOF {
 				// Try sending, even if it blocks.
-				tail.Lines <- &Line{line, now, nil, offset}
+				tail.Lines <- lineObject
 			} else {
 				tail.dropCnt += len(lines) - i
 				return true
@@ -627,6 +640,11 @@ func (tail *Tail) unreadByte() (err error) {
 	return
 }
 
+func (tail *Tail) ReleaseLine(line *Line) {
+	*line = Line{}
+	tail.linePool.Put(line)
+}
+
 // A wrapper of tomb Err()
 func (tail *Tail) UnexpectedError() (err error) {
 	err = tail.Err()

plugins/inputs/logfile/tail/tail_test.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -105,7 +106,8 @@ func TestStopAtEOF(t *testing.T) {
 
 	// Read to EOF
 	for i := 0; i < linesWrittenToFile-3; i++ {
-		<-tail.Lines
+		line := <-tail.Lines
+		tail.ReleaseLine(line)
 	}
 
 	// Verify tail.Wait() has completed.
@@ -163,11 +165,13 @@ func readThreelines(t *testing.T, tail *Tail) {
 		line := <-tail.Lines
 		if line.Err != nil {
 			t.Errorf("error tailing test file: %v", line.Err)
+			tail.ReleaseLine(line) // Release even on error
 			continue
 		}
 		if !strings.HasSuffix(line.Text, "some log line") {
 			t.Errorf("wrong line from tail found: '%v'", line.Text)
 		}
+		tail.ReleaseLine(line) // Release line back to pool
 	}
 	// If file was readable, then expect it to exist.
 	assert.Equal(t, int64(1), OpenFileCount.Load())
@@ -238,6 +242,7 @@ func TestUtf16LineSize(t *testing.T) {
 	case line := <-tail.Lines:
 		// The line should be truncated to maxLineSize
 		assert.LessOrEqual(t, len(line.Text), maxLineSize)
+		tail.ReleaseLine(line)
 	case <-time.After(1 * time.Second):
 		t.Fatal("timeout waiting for line")
 	}
@@ -265,6 +270,7 @@ func TestTail_DefaultBuffer(t *testing.T) {
 	case line := <-tail.Lines:
 		assert.NoError(t, line.Err)
 		assert.Equal(t, normalContent, line.Text)
+		tail.ReleaseLine(line)
 	case <-time.After(time.Second):
 		t.Fatal("Timeout waiting for line")
 	}
@@ -292,7 +298,146 @@ func TestTail_1MBWithExplicitMaxLineSize(t *testing.T) {
 	case line := <-tail.Lines:
 		assert.NoError(t, line.Err)
 		assert.Equal(t, largeContent, line.Text)
+		tail.ReleaseLine(line)
 	case <-time.After(time.Second):
 		t.Fatal("Timeout waiting for line")
 	}
 }
+
+// TestLinePooling verifies that Line objects are properly pooled and reused
+func TestLinePooling(t *testing.T) {
+	tmpfile, err := os.CreateTemp("", "pool_test")
+	require.NoError(t, err)
+	defer os.Remove(tmpfile.Name())
+
+	content := "line1\nline2\nline3\n"
+	_, err = tmpfile.WriteString(content)
+	require.NoError(t, err)
+	require.NoError(t, tmpfile.Close())
+
+	tail, err := TailFile(tmpfile.Name(), Config{
+		Follow:    false,
+		MustExist: true,
+	})
+	require.NoError(t, err)
+	defer tail.Stop()
+
+	var lines []*Line
+	for i := 0; i < 3; i++ {
+		select {
+		case line := <-tail.Lines:
+			lines = append(lines, line)
+		case <-time.After(time.Second):
+			t.Fatal("Timeout waiting for line")
+		}
+	}
+
+	assert.Equal(t, "line1", lines[0].Text)
+	assert.Equal(t, "line2", lines[1].Text)
+	assert.Equal(t, "line3", lines[2].Text)
+
+	// Release all lines back to pool
+	for _, line := range lines {
+		tail.ReleaseLine(line)
+	}
+
+	// Line object should be zeroed out because we released it
+	pooledLine := tail.linePool.Get().(*Line)
+	assert.Empty(t, pooledLine.Text, "Pooled line should remain zeroed")
+	assert.Empty(t, pooledLine.Time, "Pooled line should remain zeroed")
+	assert.Empty(t, pooledLine.Err, "Pooled line should remain zeroed")
+	assert.Empty(t, pooledLine.Offset, "Pooled line should remain zeroed")
+	tail.ReleaseLine(pooledLine)
+}
+
+// TestConcurrentLinePoolAccess tests that the line pool is thread-safe
+func TestConcurrentLinePoolAccess(t *testing.T) {
+	tmpfile, err := os.CreateTemp("", "concurrent_test")
+	require.NoError(t, err)
+	defer os.Remove(tmpfile.Name())
+
+	numLines := 100
+	for i := 0; i < numLines; i++ {
+		_, err = tmpfile.WriteString("concurrent test line\n")
+		require.NoError(t, err)
+	}
+	require.NoError(t, tmpfile.Close())
+
+	tail, err := TailFile(tmpfile.Name(), Config{
+		Follow:    false,
+		MustExist: true,
+	})
+	require.NoError(t, err)
+	defer tail.Stop()
+
+	// Process lines concurrently
+	var wg sync.WaitGroup
+	linesChan := make(chan *Line, numLines)
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < numLines; i++ {
+			select {
+			case line := <-tail.Lines:
+				linesChan <- line
+			case <-time.After(5 * time.Second):
+				t.Errorf("Timeout waiting for line %d", i)
+				return
+			}
+		}
+		close(linesChan)
+	}()
+
+	numWorkers := 5
+	for w := 0; w < numWorkers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for line := range linesChan {
+				assert.Equal(t, "concurrent test line", line.Text)
+				tail.ReleaseLine(line) // Release back to pool
+			}
+		}()
+	}
+
+	wg.Wait()
+}
+
+// TestUnsafeStringConversion tests the zero-copy string conversion using unsafe.String
+func TestUnsafeStringConversion(t *testing.T) {
+	tmpfile, err := os.CreateTemp("", "unsafe_test")
+	require.NoError(t, err)
+	defer os.Remove(tmpfile.Name())
+
+	// Write test content with different line types
+	testContent := "simple line\nline with spaces   \nline\r\nline\nunicode: 你好世界\nspecial chars: !@#$%^&*()\n"
+	err = os.WriteFile(tmpfile.Name(), []byte(testContent), 0600)
+	require.NoError(t, err)
+
+	tail, err := TailFile(tmpfile.Name(), Config{
+		Follow:    false,
+		MustExist: true,
+	})
+	require.NoError(t, err)
+	defer tail.Stop()
+
+	expectedLines := []string{
+		"simple line",
+		"line with spaces   ",
+		"line",
+		"line",
+		"unicode: 你好世界",
+		"special chars: !@#$%^&*()",
+	}
+
+	for i, expected := range expectedLines {
+		select {
+		case line := <-tail.Lines:
+			assert.Equal(t, expected, line.Text, "Line %d mismatch", i)
+			tail.ReleaseLine(line)
+		case <-time.After(2 * time.Second):
+			t.Fatalf("Timeout waiting for line %d", i)
+		}
+	}
+}

plugins/inputs/logfile/tailersrc.go

@@ -199,6 +199,7 @@ func (ts *tailerSrc) runTail() {
 
 	for {
 		select {
+		// Warning: Make sure to release line once done!
 		case line, ok := <-ts.tailer.Lines:
 			if !ok {
 				ts.publishEvent(msgBuf, fo)
@@ -207,6 +208,7 @@ func (ts *tailerSrc) runTail() {
 
 			if line.Err != nil {
 				log.Printf("E! [logfile] Error tailing line in file %s, Error: %s\n", ts.tailer.Filename, line.Err)
+				ts.tailer.ReleaseLine(line)
 				continue
 			}
 
@@ -216,6 +218,7 @@ func (ts *tailerSrc) runTail() {
 				text, err = ts.enc.NewDecoder().String(text)
 				if err != nil {
 					log.Printf("E! [logfile] Cannot decode the log file content for %s: %v\n", ts.tailer.Filename, err)
+					ts.tailer.ReleaseLine(line)
 					continue
 				}
 			}
@@ -231,14 +234,17 @@ func (ts *tailerSrc) runTail() {
 			} else if ignoreUntilNextEvent || msgBuf.Len() >= ts.maxEventSize {
 				ignoreUntilNextEvent = true
 				fo.ShiftInt64(line.Offset)
+				ts.tailer.ReleaseLine(line)
 				continue
 			} else {
 				msgBuf.WriteString("\n")
 				msgBuf.WriteString(text)
 				fo.ShiftInt64(line.Offset)
+				ts.tailer.ReleaseLine(line)
 				continue
 			}
 
+			ts.tailer.ReleaseLine(line)
 			ts.publishEvent(msgBuf, fo)
 			msgBuf.Reset()
 			msgBuf.WriteString(init)