diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 313c6fa91163a..ded16750b078f 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -361,7 +361,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 dc6363cc0747f..2f0aac7ce493e 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 dabd377cd9e03..2031e57399eab 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 d650471ad249d..d7d97be812bb1 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 eec7c8b39c01f..014f626315924 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 fff061694b549..39814d657825c 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 c89f78c6dd963..420af4d764092 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) }