[cache-processor] Set beat paths

libbeat/beat/pipeline.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/elastic/elastic-agent-libs/mapstr"
+	"github.com/elastic/elastic-agent-libs/paths"
 )
 
 // Pipeline provides access to libbeat event publishing by creating a Client
@@ -125,6 +126,9 @@ type ProcessingConfig struct {
 	// Private contains additional information to be passed to the processing
 	// pipeline builder.
 	Private interface{}
+
+	// Paths TODO
+	Paths *paths.Path
 }
 
 // ClientListener provides access to internal client events.

libbeat/processors/cache/cache.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -57,6 +58,7 @@ type cache struct {
 	store  Store
 	cancel context.CancelFunc
 	log    *logp.Logger
+	once   sync.Once
 }
 
 // Resulting processor implements `Close()` to release the cache resources.
@@ -70,37 +72,28 @@ func New(cfg *conf.C, log *logp.Logger) (beat.Processor, error) {
 	id := int(instanceID.Add(1))
 	log = log.Named(name).With("instance_id", id)
 
-	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,
 		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:
@@ -133,6 +126,8 @@ type CacheEntry struct {
 
 // Run enriches the given event with the host metadata.
 func (p *cache) Run(event *beat.Event) (*beat.Event, error) {
+	p.SetPaths(paths.Paths) // set default if paths is not initialized
+
 	switch {
 	case p.config.Put != nil:
 		p.log.Debugw("put", "backend_id", p.store, "config", p.config.Put)
@@ -182,6 +177,21 @@ func (p *cache) Run(event *beat.Event) (*beat.Event, error) {
 	}
 }
 
+func (p *cache) SetPaths(path *paths.Path) {
+	p.once.Do(func() {
+		src, cancel, err := getStoreFor(p.config, p.log, path)
+		if err != nil {
+			p.log.Errorf("error getting store for %s: %v", name, err)
+			return
+		}
+
+		p.store = src
+		p.cancel = cancel
+
+		p.log.Infow("initialized cache processor", "details", p)
+	})
+}
+
 // putFrom takes the configured value from the event and stores it in the cache
 // if it exists.
 func (p *cache) putFrom(event *beat.Event) error {

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 @@
 	"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.
@@ -48,6 +49,13 @@
 	p         beat.Processor
 }
 
+func (p *WhenProcessor) SetPaths(paths *paths.Path) {
+	setPather, ok := p.p.(setPaths)
+	if ok {
+		setPather.SetPaths(paths)
+	}
+}
+
 // NewConditionRule returns a processor that will execute the provided processor if the condition is true.
 func NewConditionRule(
 	c conditions.Config,
@@ -70,7 +78,7 @@
 }
 
 // Run executes this WhenProcessor.
 func (r *WhenProcessor) Run(event *beat.Event) (*beat.Event, error) {
 	if !(r.condition).Check(event) {
 		return event, nil
 	}

libbeat/processors/safe_processor.go

@@ -24,6 +24,7 @@
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/paths"
 )
 
 var ErrClosed = errors.New("attempt to use a closed processor")
@@ -46,10 +47,22 @@
 	if atomic.CompareAndSwapUint32(&p.closed, 0, 1) {
 		return Close(p.Processor)
 	}
 	logp.L().Warnf("tried to close already closed %q processor", p.Processor.String())
 	return nil
 }
 
+// TODO: common
+type setPaths interface {
+	SetPaths(*paths.Path)
+}
+
+func (p *SafeProcessor) SetPaths(paths *paths.Path) {
+	setPather, ok := p.Processor.(setPaths)
+	if ok {
+		setPather.SetPaths(paths)
+	}
+}
+
 // SafeWrap makes sure that the processor handles all the required edge-cases.
 //
 // Each processor might end up in multiple processor groups.

libbeat/publisher/pipeline/pipeline.go

@@ -35,6 +35,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/publisher/queue/memqueue"
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/paths"
 )
 
 // Pipeline implementation providint all beats publisher functionality.
@@ -71,6 +72,9 @@ type Pipeline struct {
 	forceCloseQueue bool
 
 	processors processing.Supporter
+
+	// paths TODO
+	paths *paths.Path
 }
 
 // Settings is used to pass additional settings to a newly created pipeline instance.
@@ -84,6 +88,9 @@ type Settings struct {
 	Processors processing.Supporter
 
 	InputQueueSize int
+
+	// Paths TODO
+	Paths *paths.Path
 }
 
 // WaitCloseMode enumerates the possible behaviors of WaitClose in a pipeline.
@@ -135,6 +142,7 @@ func New(
 		observer:         nilObserver,
 		waitCloseTimeout: settings.WaitClose,
 		processors:       settings.Processors,
+		paths:            settings.Paths,
 	}
 	switch settings.WaitCloseMode {
 	case WaitOnPipelineClose, WaitOnPipelineCloseThenForce:
@@ -282,6 +290,7 @@ func (p *Pipeline) createEventProcessing(cfg beat.ProcessingConfig, noPublish bo
 	if p.processors == nil {
 		return nil, nil
 	}
+	cfg.Paths = p.paths // TODO: ehm bad?
 	return p.processors.Create(cfg, noPublish)
 }
 

libbeat/publisher/processing/default.go

@@ -383,6 +383,7 @@ func (b *builder) Create(cfg beat.ProcessingConfig, drop bool) (beat.Processor,
 	// setup 8: pipeline processors list
 	if b.processors != nil {
 		// Add the global pipeline as a function processor, so clients cannot close it
+		b.processors.SetPaths(cfg.Paths)
 		processors.add(newProcessor(b.processors.title, b.processors.Run))
 	}
 

libbeat/publisher/processing/processors.go

@@ -30,6 +30,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/processors"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
+	"github.com/elastic/elastic-agent-libs/paths"
 )
 
 type group struct {
@@ -116,6 +117,20 @@ func (p *group) All() []beat.Processor {
 	return p.list
 }
 
+// TODO: move to processors
+type setPaths interface {
+	SetPaths(*paths.Path)
+}
+
+func (p *group) SetPaths(paths *paths.Path) {
+	for _, processor := range p.list {
+		setPather, ok := processor.(setPaths)
+		if ok {
+			setPather.SetPaths(paths)
+		}
+	}
+}
+
 func (p *group) Run(event *beat.Event) (*beat.Event, error) {
 	if p == nil || len(p.list) == 0 {
 		return event, nil

x-pack/libbeat/cmd/instance/beat.go

@@ -289,6 +289,7 @@ func NewBeatForReceiver(settings instance.Settings, receiverConfig map[string]an
 		InputQueueSize: b.InputQueueSize,
 		WaitCloseMode:  pipeline.WaitOnPipelineCloseThenForce,
 		WaitClose:      receiverPublisherCloseTimeout,
+		Paths:          b.Paths,
 	}
 	publisher, err := pipeline.LoadWithSettings(b.Info, monitors, b.Config.Pipeline, outputFactory, pipelineSettings)
 	if err != nil {