[cache-processor] Set beat paths (#47353)

# Lazy Initialization of the Cache Processor's File Store

## The Problem

The basic problem is that processors often use `paths.Resolve` to find directories like "data" or "logs". This function uses a global variable for the base path, which is fine when a Beat runs as a standalone process.

But when a Beat is embedded as a receiver (e.g., `fbreceiver` in the OTel Collector), this global causes problems. Each receiver needs its own isolated state directory, and a single global path prevents this.

The `cache` processor currently tries to set up its file-based store in its `New` function, which is too early. It only has access to the global path, not the receiver-specific path that gets configured later.

## The Solution

My solution is to initialize the cache's file store lazily.

Instead of creating the store in `cache.New`, I've added a `SetPaths(*paths.Path)` method to the processor. This method creates the file store and is wrapped in a `sync.Once` to make sure it only runs once. The processor's internal store object stays `nil` until `SetPaths` is called during pipeline construction.

## How it Works

The path info gets passed down when a client connects to the pipeline. Here's the flow:

1.  **`x-pack/filebeat/fbreceiver`**: `createReceiver` instantiates the processors (including `cache` with a `nil` store) and calls `instance.NewBeatForReceiver`.
2.  **`x-pack/libbeat/cmd/instance`**: `NewBeatForReceiver` creates the `paths.Path` object from the receiver's specific configuration.
3.  **`libbeat/publisher/pipeline`**: This `paths.Path` object is passed into the pipeline. When a client connects, the path is added to the `beat.ProcessingConfig`.
4.  **`libbeat/publisher/processing`**: The processing builder gets this config and calls `group.SetPaths`, which passes the path down to each processor.
5.  **`libbeat/processors/cache`**: `SetPaths` is finally called on the cache processor instance, and the `sync.Once` guard ensures the file store is created with the correct path.

## Diagram
```mermaid
graph TD
    subgraph "libbeat/processors/cache (init)"
        A["init()"]
    end
    subgraph "libbeat/processors"
        B["processors.RegisterPlugin"]
        C{"registry"}
    end
    A --> B;
    B -- "Save factory" --> C;

    subgraph "x-pack/filebeat/fbreceiver"
        D["createReceiver"]
    end

    subgraph "libbeat/processors"
         E["processors.New(config)"]
         C -. "Lookup 'cache'" .-> E;
    end
    D --> E;
    D --> I;
    E --> G;

    subgraph "libbeat/processors/cache"
        G["cache.New()"] -- store=nil --> H{"cache"};
    end

    subgraph "x-pack/libbeat/cmd/instance"
        I["instance.NewBeatForReceiver"];
        I --> J{"paths.Path object"};
    end

    subgraph "libbeat/publisher/pipeline"
        J --> K["pipeline.New"];
        K --> L["ConnectWith"];
    end

    subgraph "libbeat/publisher/processing"
        L -- "Config w/ paths" --> N["builder.Create"];
        N --> O["group.SetPaths"];
    end

    subgraph "libbeat/processors/cache"
        O --> P["cache.SetPaths"];
        P --> Q["sync.Once"];
        Q -- "initialize store" --> H;
    end
```

## Pros and Cons of This Approach

*   **Pros**:
    *   It's a minimal, targeted change that solves the immediate problem.
    *   It avoids a large-scale, breaking refactoring of all processors.
    *   It maintains backward compatibility for existing processors and downstream consumers of `libbeat`.
*   **Cons**:
    *   Using a type assertion for the `setPaths` interface feels a bit like magic, since the behavior changes at runtime depending on whether a processor implements it.

## Alternatives Considered

### Option 1: Add a `paths` argument to all processor constructors

*   **Pros**:
    *   Simple and direct.
*   **Cons**:
    *   Requires a global refactoring of all processors.
    *   Breaks external downstream libbeat importers like Cloudbeat.
    *   The `paths` argument is not needed in many processors, so adding a rarely used option to the function signature is verbose.

### Option 2: Refactor `processors` to introduce a "V2" interface

*   **Pros**:
    *   Allows for a new, backwards-compatible signature (e.g., using a config struct).
    *   This can still be done later.
    *   We can support both V1 processors and gradually move processors to V2.
*   **Cons**:
    *   Needs a significant refactoring effort.

## Checklist

<!-- Mandatory
Add a checklist of things that are required to be reviewed in order to have the PR approved

List here all the items you have verified BEFORE sending this PR. Please DO NOT remove any item, striking through those that do not apply. (Just in case, strikethrough uses two tildes. ~~Scratch this.~~)
-->

- [x] My code follows the style guidelines of this project
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] ~~I have made corresponding changes to the documentation~~
- [ ] ~~I have made corresponding change to the default configuration files~~
- [x] I have added tests that prove my fix is effective or that my feature works. Where relevant, I have used the [`stresstest.sh`](https://github.com/elastic/beats/blob/main/script/stresstest.sh) script to run them under stress conditions and race detector to verify their stability.
- [ ] ~~I have added an entry in `./changelog/fragments` using the [changelog tool](https://github.com/elastic/elastic-agent-changelog-tool/blob/main/docs/usage.md).~~

## How to test this PR locally
### Configuration

`filebeat-cache-mwe.yml`:

```yaml
path.data: /tmp/data

filebeat.inputs:
  - type: filestream
    id: filestream-input
    enabled: true
    paths:
      - /tmp/logs/*.log
    parsers:
      - ndjson:
          target: ""

processors:
  # PUT: Store metadata when event.type is "source"
  - if:
      equals:
        event.type: "source"
    then:
      - cache:
          backend:
            file:
              id: test_cache
              write_interval: 5s
          put:
            key_field: event.id
            value_field: event.metadata
            ttl: 1h

  # GET: Retrieve metadata when event.type is "target"
  - if:
      equals:
        event.type: "target"
    then:
      - cache:
          backend:
            file:
              id: test_cache
          get:
            key_field: event.id
            target_field: cached_metadata

output.console:
  enabled: true
```

### Setup

```bash
# Create directory
#rm -rf /tmp/data /tmp/logs
mkdir -p /tmp/logs

# Create test data
cat > /tmp/logs/test.log <<'EOF'
{"event":{"type":"source","id":"001","metadata":{"user":"user-1","role":"admin","sequence":1,"data":{"ip":"192.168.1.1","session":"session-001"}}},"message":"source event 1"}
{"event":{"type":"source","id":"002","metadata":{"user":"user-2","role":"admin","sequence":2,"data":{"ip":"192.168.1.2","session":"session-002"}}},"message":"source event 2"}
{"event":{"type":"source","id":"003","metadata":{"user":"user-3","role":"admin","sequence":3,"data":{"ip":"192.168.1.3","session":"session-003"}}},"message":"source event 3"}
{"event":{"type":"source","id":"004","metadata":{"user":"user-4","role":"admin","sequence":4,"data":{"ip":"192.168.1.4","session":"session-004"}}},"message":"source event 4"}
{"event":{"type":"source","id":"005","metadata":{"user":"user-5","role":"admin","sequence":5,"data":{"ip":"192.168.1.5","session":"session-005"}}},"message":"source event 5"}
{"event":{"type":"target","id":"001"},"message":"target event 1"}
{"event":{"type":"target","id":"002"},"message":"target event 2"}
{"event":{"type":"target","id":"003"},"message":"target event 3"}
{"event":{"type":"target","id":"004"},"message":"target event 4"}
{"event":{"type":"target","id":"005"},"message":"target event 5"}
EOF

# Run filebeat
./x-pack/filebeat/filebeat -e -c filebeat-cache-mwe.yml
```

### Expected Output

Target events should have `cached_metadata` field populated:

```json
{
  "event": {
    "type": "target",
    "id": "001"
  },
  "message": "target event 1",
  "cached_metadata": {
    "user": "user-1",
    "role": "admin",
    "sequence": 1,
    "data": {
      "ip": "192.168.1.1",
      "session": "session-001"
    }
  }
}
```

### Cache Files

After running filebeat, check cache files:

```bash
cat /tmp/data/cache_processor/test_cache
```
example:
```json
{"key":"001","val":{"data":{"ip":"192.168.1.1","session":"session-001"},"role":"admin","sequence":1,"user":"user-1"},"expires":"2025-11-20T15:02:32.865896537+01:00"}
{"key":"002","val":{"data":{"ip":"192.168.1.2","session":"session-002"},"role":"admin","sequence":2,"user":"user-2"},"expires":"2025-11-20T15:02:32.865950973+01:00"}
{"key":"003","val":{"data":{"ip":"192.168.1.3","session":"session-003"},"role":"admin","sequence":3,"user":"user-3"},"expires":"2025-11-20T15:02:32.865972408+01:00"}
{"key":"004","val":{"data":{"ip":"192.168.1.4","session":"session-004"},"role":"admin","sequence":4,"user":"user-4"},"expires":"2025-11-20T15:02:32.865988843+01:00"}
{"key":"005","val":{"data":{"ip":"192.168.1.5","session":"session-005"},"role":"admin","sequence":5,"user":"user-5"},"expires":"2025-11-20T15:02:32.866006958+01:00"}
```

## Related issues

- Closes #46985

Author: orestisfl

libbeat/cmd/instance/beat.go

@@ -389,6 +389,7 @@ func (b *Beat) createBeater(bt beat.Creator) (beat.Beater, error) {
 		WaitClose:      time.Second,
 		Processors:     b.processors,
 		InputQueueSize: b.InputQueueSize,
+		Paths:          b.Paths,
 	}
 	publisher, err = pipeline.LoadWithSettings(b.Info, monitors, b.Config.Pipeline, outputFactory, settings)
 	if err != nil {

libbeat/processors/cache/cache.go

@@ -69,38 +69,31 @@ func New(cfg *conf.C, log *logp.Logger) (beat.Processor, error) {
 	// Logging (each processor instance has a unique ID).
 	id := int(instanceID.Add(1))
 	log = log.Named(name).With("instance_id", id)
+	log.Infow("cache processor created", "config", config)
 
-	src, cancel, err := getStoreFor(config, log)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get the store for %s: %w", name, err)
-	}
-
-	p := &cache{
+	return &cache{
 		config: config,
-		store:  src,
-		cancel: cancel,
+		store:  nil, // initialized in SetPaths
 		log:    log,
-	}
-	p.log.Infow("initialized cache processor", "details", p)
-	return p, nil
+	}, nil
 }
 
 // getStoreFor returns a backing store for the provided configuration,
 // and a context cancellation that releases the cache resource when it
 // is no longer required. The cancellation should be called when the
 // processor is closed.
-func getStoreFor(cfg config, log *logp.Logger) (Store, context.CancelFunc, error) {
+func getStoreFor(cfg config, log *logp.Logger, path *paths.Path) (Store, context.CancelFunc, error) {
 	switch {
 	case cfg.Store.Memory != nil:
 		s, cancel := memStores.get(cfg.Store.Memory.ID, cfg)
 		return s, cancel, nil
 
 	case cfg.Store.File != nil:
-		err := os.MkdirAll(paths.Resolve(paths.Data, "cache_processor"), 0o700)
+		err := os.MkdirAll(path.Resolve(paths.Data, "cache_processor"), 0o700)
 		if err != nil {
 			return nil, noop, fmt.Errorf("cache processor could not create store directory: %w", err)
 		}
-		s, cancel := fileStores.get(cfg.Store.File.ID, cfg, log)
+		s, cancel := fileStores.get(cfg.Store.File.ID, cfg, log, path)
 		return s, cancel, nil
 
 	default:
@@ -131,8 +124,12 @@ type CacheEntry struct {
 	index   int
 }
 
-// Run enriches the given event with the host metadata.
+// Run enriches the given event with cached metadata.
 func (p *cache) Run(event *beat.Event) (*beat.Event, error) {
+	if p.store == nil {
+		return event, fmt.Errorf("cache processor store not initialized")
+	}
+
 	switch {
 	case p.config.Put != nil:
 		p.log.Debugw("put", "backend_id", p.store, "config", p.config.Put)
@@ -186,6 +183,21 @@ func (p *cache) Run(event *beat.Event) (*beat.Event, error) {
 	}
 }
 
+// SetPaths initializes the cache store with the provided paths configuration.
+// This method must be called before the processor can be used.
+func (p *cache) SetPaths(path *paths.Path) error {
+	src, cancel, err := getStoreFor(p.config, p.log, path)
+	if err != nil {
+		return fmt.Errorf("cache processor could not create store for %s: %w", name, err)
+	}
+
+	p.store = src
+	p.cancel = cancel
+
+	p.log.Infow("initialized cache processor", "details", p)
+	return nil
+}
+
 // putFrom takes the configured value from the event and stores it in the cache
 // if it exists.
 func (p *cache) putFrom(event *beat.Event) (err error) {
@@ -302,7 +314,9 @@ func (p *cache) deleteFor(event *beat.Event) (err error) {
 }
 
 func (p *cache) Close() error {
-	p.cancel()
+	if p.cancel != nil {
+		p.cancel()
+	}
 	return nil
 }
 

libbeat/processors/cache/cache_test.go

@@ -19,6 +19,7 @@ package cache
 
 import (
 	"errors"
+	"strings"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -28,6 +29,7 @@ import (
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/mapstr"
+	"github.com/elastic/elastic-agent-libs/paths"
 )
 
 type cacheTestStep struct {
@@ -662,7 +664,19 @@ func TestCache(t *testing.T) {
 				t.Log(p)
 				c, ok := p.(*cache)
 				if !ok {
-					t.Fatalf("processor %d is not an *cache", i)
+					t.Fatalf("processor %d is not a *cache", i)
+				}
+
+				// Initialize the store with paths
+				tmpDir := t.TempDir()
+				err = c.SetPaths(&paths.Path{
+					Home:   tmpDir,
+					Config: tmpDir,
+					Data:   tmpDir,
+					Logs:   tmpDir,
+				})
+				if err != nil {
+					t.Errorf("unexpected error from SetPaths: %v", err)
 				}
 
 				defer func() {
@@ -702,3 +716,44 @@ func TestCache(t *testing.T) {
 		})
 	}
 }
+
+func TestSetPathsUninitialized(t *testing.T) {
+	cfg, err := conf.NewConfigFrom(mapstr.M{
+		"backend": mapstr.M{
+			"memory": mapstr.M{
+				"id": "test",
+			},
+		},
+		"get": mapstr.M{
+			"key_field":    "key",
+			"target_field": "target",
+		},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error from NewConfigFrom: %v", err)
+	}
+
+	p, err := New(cfg, logptest.NewTestingLogger(t, ""))
+	if err != nil {
+		t.Fatalf("unexpected error from New: %v", err)
+	}
+
+	c, ok := p.(*cache)
+	if !ok {
+		t.Fatal("processor is not a *cache")
+	}
+	defer func() {
+		if err := c.Close(); err != nil {
+			t.Errorf("unexpected error from c.Close(): %v", err)
+		}
+	}()
+
+	// Try to use without SetPaths - should fail
+	event, err := c.Run(&beat.Event{})
+	if event == nil {
+		t.Error("expected non-nil event")
+	}
+	if err == nil || !strings.Contains(err.Error(), "cache processor store not initialized") {
+		t.Fatalf("expected error containing 'cache processor store not initialized', got: %v", err)
+	}
+}

libbeat/processors/cache/file_store.go

@@ -47,12 +47,12 @@ type fileStoreSet struct {
 // and its reference count is increased. The returned context.CancelFunc
 // reduces the reference count and deletes the fileStore from the set if the
 // count reaches zero.
-func (s *fileStoreSet) get(id string, cfg config, log *logp.Logger) (*fileStore, context.CancelFunc) {
+func (s *fileStoreSet) get(id string, cfg config, log *logp.Logger, path *paths.Path) (*fileStore, context.CancelFunc) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	store, ok := s.stores[id]
 	if !ok {
-		store = newFileStore(cfg, id, pathFromConfig(cfg, log), log)
+		store = newFileStore(cfg, id, pathFromConfig(cfg, log, path), log)
 		s.stores[store.id] = store
 	}
 	store.add(cfg)
@@ -63,10 +63,10 @@ func (s *fileStoreSet) get(id string, cfg config, log *logp.Logger) (*fileStore,
 }
 
 // pathFromConfig returns the mapping form a config to a file-system path.
-func pathFromConfig(cfg config, log *logp.Logger) string {
-	path := filepath.Join(paths.Resolve(paths.Data, "cache_processor"), cleanFilename(cfg.Store.File.ID))
-	log.Infow("mapping file-backed cache processor config to file path", "id", cfg.Store.File.ID, "path", path)
-	return path
+func pathFromConfig(cfg config, log *logp.Logger, path *paths.Path) string {
+	resultPath := filepath.Join(path.Resolve(paths.Data, "cache_processor"), cleanFilename(cfg.Store.File.ID))
+	log.Infow("mapping file-backed cache processor config to file path", "id", cfg.Store.File.ID, "path", resultPath)
+	return resultPath
 }
 
 // cleanFilename replaces illegal printable characters (and space or dot) in

libbeat/processors/conditionals.go

@@ -26,6 +26,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/conditions"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/paths"
 )
 
 // NewConditional returns a constructor suitable for registering when conditionals as a plugin.
@@ -77,6 +78,14 @@ func (r *WhenProcessor) Run(event *beat.Event) (*beat.Event, error) {
 	return r.p.Run(event)
 }
 
+func (r *WhenProcessor) SetPaths(paths *paths.Path) error {
+	pathSetter, ok := r.p.(PathSetter)
+	if ok {
+		return pathSetter.SetPaths(paths)
+	}
+	return nil
+}
+
 func (r *WhenProcessor) String() string {
 	return fmt.Sprintf("%v, condition=%v", r.p.String(), r.condition.String())
 }
@@ -86,7 +95,7 @@ func (r *WhenProcessor) String() string {
 // processors, one with `Close` and one without.  The decision of
 // which to return is determined if the underlying processors require
 // `Close`.  This is useful because some places in the code base
-// (eg. javascript processors) require stateless processors (no Close
+// (e.g. javascript processors) require stateless processors (no Close
 // method).
 type ClosingWhenProcessor struct {
 	WhenProcessor
@@ -199,6 +208,27 @@ func (p *IfThenElseProcessor) Run(event *beat.Event) (*beat.Event, error) {
 	return event, nil
 }
 
+func (p *IfThenElseProcessor) SetPaths(paths *paths.Path) error {
+	var err error
+	for _, proc := range p.then.List {
+		if procWithSet, ok := proc.(PathSetter); ok {
+			err = errors.Join(err, procWithSet.SetPaths(paths))
+		}
+	}
+
+	if p.els == nil {
+		return err
+	}
+
+	for _, proc := range p.els.List {
+		if procWithSet, ok := proc.(PathSetter); ok {
+			err = errors.Join(err, procWithSet.SetPaths(paths))
+		}
+	}
+
+	return err
+}
+
 func (p *IfThenElseProcessor) String() string {
 	var sb strings.Builder
 	sb.WriteString("if ")

libbeat/processors/conditionals_test.go

@@ -26,6 +26,8 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/elastic/elastic-agent-libs/paths"
+
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/conditions"
 	conf "github.com/elastic/elastic-agent-libs/config"
@@ -248,6 +250,18 @@ func (c *errorProcessor) Close() error {
 	return ErrProcessorClose
 }
 
+var ErrSetPathsProcessor = fmt.Errorf("error processor set paths error")
+
+type setPathsProcessor struct{}
+
+func (c *setPathsProcessor) Run(e *beat.Event) (*beat.Event, error) {
+	return e, nil
+}
+func (c *setPathsProcessor) String() string { return "error_processor" }
+func (c *setPathsProcessor) SetPaths(p *paths.Path) error {
+	return fmt.Errorf("error_processor set paths error: %s", p)
+}
+
 func TestConditionRuleClose(t *testing.T) {
 	const whenCondition = `
 contains.a: b
@@ -289,16 +303,73 @@ then:
 
 	beatProcessor, err := NewIfElseThenProcessor(c, logptest.NewTestingLogger(t, ""))
 	require.NoError(t, err)
+	requireImplements[Closer](t, beatProcessor)
 
 	// Verify we got a ClosingIfThenElseProcessor
 	closingProc := requireAs[*ClosingIfThenElseProcessor](t, beatProcessor)
 	assert.Nil(t, closingProc.els, "els should be nil when no else clause is provided")
-	assert.Implements(t, (*Closer)(nil), beatProcessor)
 
 	err = closingProc.Close()
 	require.NoError(t, err)
 }
 
+func TestIfThenElseProcessorSetPaths(t *testing.T) {
+	logger := logptest.NewTestingLogger(t, "")
+	thenProcessors := &Processors{
+		List: []beat.Processor{&setPathsProcessor{}},
+		log:  logger,
+	}
+	elsProcessors := &Processors{
+		List: []beat.Processor{&setPathsProcessor{}},
+		log:  logger,
+	}
+	proc := &IfThenElseProcessor{
+		cond: nil,
+		then: thenProcessors,
+		els:  elsProcessors,
+	}
+
+	// SetPaths should not panic when then is nil
+	tmpDir := t.TempDir()
+	beatPaths := &paths.Path{
+		Home:   tmpDir,
+		Config: tmpDir,
+		Data:   tmpDir,
+		Logs:   tmpDir,
+	}
+	err := proc.SetPaths(beatPaths)
+	require.ErrorAs(t, err, &ErrSetPathsProcessor)
+	require.ErrorContains(t, err, ErrSetPathsProcessor.Error())
+	require.ErrorContains(t, err, beatPaths.String())
+}
+
+func TestIfThenElseProcessorSetPathsNil(t *testing.T) {
+	const cfg = `
+if:
+  equals.test: value
+then:
+  - add_fields: {target: "", fields: {test_field: test_value}}
+`
+	c, err := conf.NewConfigWithYAML([]byte(cfg), "if-then config")
+	require.NoError(t, err)
+
+	beatProcessor, err := NewIfElseThenProcessor(c, logptest.NewTestingLogger(t, ""))
+	require.NoError(t, err)
+
+	proc := requireImplements[PathSetter](t, beatProcessor)
+
+	// SetPaths should not panic when then is nil
+	tmpDir := t.TempDir()
+	beatPaths := &paths.Path{
+		Home:   tmpDir,
+		Config: tmpDir,
+		Data:   tmpDir,
+		Logs:   tmpDir,
+	}
+	err = proc.SetPaths(beatPaths)
+	require.NoError(t, err)
+}
+
 // requireAs performs a type assertion and requires it to succeed.
 func requireAs[T any](t *testing.T, v any) T {
 	t.Helper()
@@ -310,3 +381,14 @@ func requireAs[T any](t *testing.T, v any) T {
 
 	return result
 }
+
+func requireImplements[T any](t *testing.T, v any) T {
+	t.Helper()
+	expected := (*T)(nil)
+	require.Implements(t, expected, v)
+
+	result, ok := v.(T)
+	require.True(t, ok, "sanity check: expected %T, got %T", expected, v)
+
+	return result
+}