From e257c14ec162ff747ed84e687dff994bb551955e Mon Sep 17 00:00:00 2001
From: Dane Strandboge <136023093+DStrand1@users.noreply.github.com>
Date: Fri, 18 Oct 2024 13:22:02 -0500
Subject: [PATCH] fix(agent): Use a unique WAL file for plugin instances of the
 same type (#15966)

---
 docs/CONFIGURATION.md       |  2 +-
 models/buffer.go            |  4 ++--
 models/buffer_disk.go       | 11 +++++++++--
 models/buffer_disk_test.go  |  3 ++-
 models/buffer_mem_test.go   |  2 +-
 models/buffer_suite_test.go |  2 +-
 models/running_output.go    |  2 +-
 7 files changed, 17 insertions(+), 9 deletions(-)

diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md
index bb4b42a37..2d40f5707 100644
--- 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
index dc6363cc0..2f0aac7ce 100644
--- 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
index dabd377cd..2031e5739 100644
--- 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
index d650471ad..d7d97be81 100644
--- 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
index eec7c8b39..014f62631 100644
--- 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
index fff061694..39814d657 100644
--- 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
index c89f78c6d..420af4d76 100644
--- 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)
 	}