fix(agent): Use a unique WAL file for plugin instances of the same type (#15966)

mstrandboge · web-flow · commit e257c14ec162 · 2024-10-18T20:22:02.000+02:00
diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md
@@ -362,7 +362,7 @@ The agent table configures Telegraf and the defaults used across all plugins.
 
 - **buffer_directory**:
   The directory to use when in `disk` buffer mode. Each output plugin will make
-  another subdirectory in this directory with the output plugin's name.
+  another subdirectory in this directory with the output plugin's ID.
 
 ## Plugins
 
diff --git a/models/buffer.go b/models/buffer.go
@@ -49,7 +49,7 @@ type BufferStats struct {
 }
 
 // NewBuffer returns a new empty Buffer with the given capacity.
-func NewBuffer(name string, alias string, capacity int, strategy string, path string) (Buffer, error) {
+func NewBuffer(name, id, alias string, capacity int, strategy, path string) (Buffer, error) {
 	registerGob()
 
 	bs := NewBufferStats(name, alias, capacity)
@@ -58,7 +58,7 @@ func NewBuffer(name string, alias string, capacity int, strategy string, path st
 	case "", "memory":
 		return NewMemoryBuffer(capacity, bs)
 	case "disk":
-		return NewDiskBuffer(name, path, bs)
+		return NewDiskBuffer(name, id, path, bs)
 	}
 	return nil, fmt.Errorf("invalid buffer strategy %q", strategy)
 }
diff --git a/models/buffer_disk.go b/models/buffer_disk.go
@@ -33,12 +33,19 @@ type DiskBuffer struct {
 	isEmpty bool
 }
 
-func NewDiskBuffer(name string, path string, stats BufferStats) (*DiskBuffer, error) {
-	filePath := filepath.Join(path, name)
+func NewDiskBuffer(name, id, path string, stats BufferStats) (*DiskBuffer, error) {
+	filePath := filepath.Join(path, id)
 	walFile, err := wal.Open(filePath, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open wal file: %w", err)
 	}
+	//nolint:errcheck // cannot error here
+	if index, _ := walFile.FirstIndex(); index == 0 {
+		// simple way to test if the walfile is freshly initialized, meaning no existing file was found
+		log.Printf("I! WAL file not found for plugin outputs.%s (%s), "+
+			"this can safely be ignored if you added this plugin instance for the first time", name, id)
+	}
+
 	buf := &DiskBuffer{
 		BufferStats: stats,
 		file:        walFile,
diff --git a/models/buffer_disk_test.go b/models/buffer_disk_test.go
@@ -22,7 +22,7 @@ func newTestDiskBuffer(t testing.TB) Buffer {
 
 func newTestDiskBufferWithPath(t testing.TB, name string, path string) Buffer {
 	t.Helper()
-	buf, err := NewBuffer(name, "", 0, "disk", path)
+	buf, err := NewBuffer(name, "123", "", 0, "disk", path)
 	require.NoError(t, err)
 	buf.Stats().MetricsAdded.Set(0)
 	buf.Stats().MetricsWritten.Set(0)
@@ -45,6 +45,7 @@ func TestBuffer_RetainsTrackingInformation(t *testing.T) {
 func TestBuffer_TrackingDroppedFromOldWal(t *testing.T) {
 	path, err := os.MkdirTemp("", "*-buffer-test")
 	require.NoError(t, err)
+	path = filepath.Join(path, "123")
 	walfile, err := wal.Open(path, nil)
 	require.NoError(t, err)
 
diff --git a/models/buffer_mem_test.go b/models/buffer_mem_test.go
@@ -8,7 +8,7 @@ import (
 
 func newTestMemoryBuffer(t testing.TB, capacity int) Buffer {
 	t.Helper()
-	buf, err := NewBuffer("test", "", capacity, "memory", "")
+	buf, err := NewBuffer("test", "123", "", capacity, "memory", "")
 	require.NoError(t, err)
 	buf.Stats().MetricsAdded.Set(0)
 	buf.Stats().MetricsWritten.Set(0)
diff --git a/models/buffer_suite_test.go b/models/buffer_suite_test.go
@@ -84,7 +84,7 @@ func MetricTime(sec int64) telegraf.Metric {
 
 func (s *BufferSuiteTest) newTestBuffer(capacity int) Buffer {
 	s.T().Helper()
-	buf, err := NewBuffer("test", "", capacity, s.bufferType, s.bufferPath)
+	buf, err := NewBuffer("test", "123", "", capacity, s.bufferType, s.bufferPath)
 	s.Require().NoError(err)
 	buf.Stats().MetricsAdded.Set(0)
 	buf.Stats().MetricsWritten.Set(0)
diff --git a/models/running_output.go b/models/running_output.go
@@ -104,7 +104,7 @@ func NewRunningOutput(
 		batchSize = DefaultMetricBatchSize
 	}
 
-	b, err := NewBuffer(config.Name, config.Alias, bufferLimit, config.BufferStrategy, config.BufferDirectory)
+	b, err := NewBuffer(config.Name, config.ID, config.Alias, bufferLimit, config.BufferStrategy, config.BufferDirectory)
 	if err != nil {
 		panic(err)
 	}

Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ type BufferStats struct {`
`49`	`49`	`}`
`50`	`50`
`51`	`51`	`// NewBuffer returns a new empty Buffer with the given capacity.`
`52`		`-func NewBuffer(name string, alias string, capacity int, strategy string, path string) (Buffer, error) {`
	`52`	`+func NewBuffer(name, id, alias string, capacity int, strategy, path string) (Buffer, error) {`
`53`	`53`	`registerGob()`
`54`	`54`
`55`	`55`	`bs := NewBufferStats(name, alias, capacity)`
`@@ -58,7 +58,7 @@ func NewBuffer(name string, alias string, capacity int, strategy string, path st`
`58`	`58`	`case "", "memory":`
`59`	`59`	`return NewMemoryBuffer(capacity, bs)`
`60`	`60`	`case "disk":`
`61`		`- return NewDiskBuffer(name, path, bs)`
	`61`	`+ return NewDiskBuffer(name, id, path, bs)`
`62`	`62`	`}`
`63`	`63`	`return nil, fmt.Errorf("invalid buffer strategy %q", strategy)`
`64`	`64`	`}`
Original file line number	Diff line number	Diff line change
`@@ -104,7 +104,7 @@ func NewRunningOutput(`
`104`	`104`	`batchSize = DefaultMetricBatchSize`
`105`	`105`	`}`
`106`	`106`
`107`		`- b, err := NewBuffer(config.Name, config.Alias, bufferLimit, config.BufferStrategy, config.BufferDirectory)`
	`107`	`+ b, err := NewBuffer(config.Name, config.ID, config.Alias, bufferLimit, config.BufferStrategy, config.BufferDirectory)`
`108`	`108`	`if err != nil {`
`109`	`109`	`panic(err)`
`110`	`110`	`}`