Fix SetPaths to be idempotent for global processors

filebeat/tests/integration/cache_processor_test.go

@@ -0,0 +1,118 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build integration
+
+package integration
+
+import (
+	"fmt"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/v7/libbeat/tests/integration"
+)
+
+// TestGlobalCacheProcessorMultipleInputs tests that a global file-backed cache
+// processor works correctly when multiple inputs connect to the pipeline.
+func TestGlobalCacheProcessorMultipleInputs(t *testing.T) {
+	filebeat := integration.NewBeat(
+		t,
+		"filebeat",
+		"../../filebeat.test",
+	)
+	tempDir := filebeat.TempDir()
+
+	// Config with multiple inputs and a global file-backed cache processor.
+	configTemplate := `
+filebeat.inputs:
+  - type: filestream
+    id: input-1
+    enabled: true
+    paths:
+      - %s
+    prospector.scanner.fingerprint.enabled: false
+
+  - type: filestream
+    id: input-2
+    enabled: true
+    paths:
+      - %s
+    prospector.scanner.fingerprint.enabled: false
+
+  - type: filestream
+    id: input-3
+    enabled: true
+    paths:
+      - %s
+    prospector.scanner.fingerprint.enabled: false
+
+# Global cache processor - shared by ALL inputs.
+# Tests that SetPaths works correctly when multiple inputs connect.
+processors:
+  - cache:
+      backend:
+        file:
+          id: test-cache
+          write_interval: 1s
+        capacity: 1000
+      put:
+        key_field: message
+        value_field: message
+        ttl: 1h
+      ignore_missing: true
+
+path.home: %s
+
+output.file:
+  path: ${path.home}
+  filename: output
+  codec.json:
+    pretty: false
+
+logging.level: info
+`
+
+	// Create log files for each input
+	logFile1 := filepath.Join(tempDir, "input1.log")
+	logFile2 := filepath.Join(tempDir, "input2.log")
+	logFile3 := filepath.Join(tempDir, "input3.log")
+	filebeat.WriteConfigFile(fmt.Sprintf(
+		configTemplate,
+		logFile1,
+		logFile2,
+		logFile3,
+		tempDir,
+	))
+
+	// Write test data to all log files
+	integration.WriteLogFile(t, logFile1, 10, false)
+	integration.WriteLogFile(t, logFile2, 10, false)
+	integration.WriteLogFile(t, logFile3, 10, false)
+
+	filebeat.Start()
+	filebeat.WaitPublishedEvents(30*time.Second, 30)
+
+	// Verify the cache file was created in the correct location
+	cacheFile := filepath.Join(tempDir, "data", "cache_processor", "test-cache")
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		assert.FileExists(c, cacheFile)
+	}, 10*time.Second, 100*time.Millisecond)
+}

libbeat/processors/safe_processor.go

@@ -51,6 +51,7 @@ type SafeProcessor struct {
 
 	mu    sync.RWMutex
 	state state
+	paths *paths.Path
 }
 
 // safeProcessorWithClose extends SafeProcessor to also handle Close.
@@ -103,9 +104,13 @@ func (p *SafeProcessor) SetPaths(paths *paths.Path) error {
 	switch p.state {
 	case stateInit:
 		p.state = stateSetPaths
+		p.paths = paths
 		return pathSetter.SetPaths(paths)
 	case stateSetPaths:
-		return ErrPathsAlreadySet
+		if p.paths != paths {
+			return ErrPathsAlreadySet
+		}
+		return nil
 	case stateClosed:
 		return ErrSetPathsOnClosed
 	}

libbeat/processors/safe_processor_test.go

@@ -198,11 +198,17 @@ func TestSafeProcessorSetPathsClose(t *testing.T) {
 		sp, ok = bp.(PathSetter)
 		require.True(t, ok)
 		require.NotNil(t, sp)
-		err = sp.SetPaths(&paths.Path{})
+		testPaths := &paths.Path{}
+		err = sp.SetPaths(testPaths)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, p.setPathsCount)
+
+		// set paths again with the SAME pointer (idempotent for global processors)
+		err = sp.SetPaths(testPaths)
 		assert.NoError(t, err)
 		assert.Equal(t, 1, p.setPathsCount)
 
-		// set paths again
+		// set paths again with a DIFFERENT pointer (should error)
 		err = sp.SetPaths(&paths.Path{})
 		assert.ErrorIs(t, err, ErrPathsAlreadySet)
 		assert.Equal(t, 1, p.setPathsCount)
@@ -282,11 +288,17 @@ func TestSafeProcessorSetPaths(t *testing.T) {
 		sp, ok = bp.(PathSetter)
 		require.True(t, ok)
 		require.NotNil(t, sp)
-		err = sp.SetPaths(&paths.Path{})
+		testPaths := &paths.Path{}
+		err = sp.SetPaths(testPaths)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, p.setPathsCount)
+
+		// set paths again with the SAME pointer (idempotent for global processors)
+		err = sp.SetPaths(testPaths)
 		assert.NoError(t, err)
 		assert.Equal(t, 1, p.setPathsCount)
 
-		// set paths again
+		// set paths again with a DIFFERENT pointer (should error)
 		err = sp.SetPaths(&paths.Path{})
 		assert.ErrorIs(t, err, ErrPathsAlreadySet)
 		assert.Equal(t, 1, p.setPathsCount)

libbeat/tests/integration/framework.go

@@ -1227,9 +1227,10 @@ func StartMockES(
 		addr = "localhost:0"
 	}
 
-	l, err := net.Listen("tcp", addr)
+	lc := net.ListenConfig{}
+	l, err := lc.Listen(t.Context(), "tcp", addr)
 	if err != nil {
-		if l, err = net.Listen("tcp6", addr); err != nil {
+		if l, err = lc.Listen(t.Context(), "tcp6", addr); err != nil {
 			t.Fatalf("failed to listen on a port: %v", err)
 		}
 	}
@@ -1270,14 +1271,10 @@ func (b *BeatProc) WaitPublishedEvents(timeout time.Duration, events int) {
 	t := b.t
 	t.Helper()
 
-	msg := strings.Builder{}
 	path := filepath.Join(b.TempDir(), "output-*.ndjson")
-	assert.Eventually(t, func() bool {
-		got := b.CountFileLines(path)
-		msg.Reset()
-		fmt.Fprintf(&msg, "expecting %d events, got %d", events, got)
-		return got == events
-	}, timeout, 200*time.Millisecond, &msg)
+	assert.EventuallyWithT(t, func(collect *assert.CollectT) {
+		assert.Equal(collect, events, b.CountFileLines(path))
+	}, timeout, 200*time.Millisecond)
 }
 
 // GetEventsFromFileOutput reads all events from file output. If n > 0,