libbeat/processors/cache: add file-backed cache

elastic · Sep 27, 2023 · 0afe564 · 0afe564
1 parent 58db08f
commit 0afe564
Show file tree

Hide file tree

Showing 7 changed files with 757 additions and 77 deletions.
diff --git a/CHANGELOG-developer.next.asciidoc b/CHANGELOG-developer.next.asciidoc
@@ -173,6 +173,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Add device handling to Okta API package for entity analytics. {pull}35980[35980]
 - Make Filebeat HTTPJSON input process responses sequentially. {pull}36493[36493]
 - Add initial infrastructure for a caching enrichment processor. {pull}36619[36619]
+- Add file-backed cache for cache enrichment processor. {pull}[]
 
 ==== Deprecated
 

diff --git a/libbeat/processors/cache/cache.go b/libbeat/processors/cache/cache.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -29,6 +30,7 @@ import (
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
+	"github.com/elastic/elastic-agent-libs/paths"
 )
 
 const name = "cache"
@@ -64,15 +66,15 @@ func New(cfg *conf.C) (beat.Processor, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to unpack the %s configuration: %w", name, err)
 	}
-	src, cancel, err := getStoreFor(config)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get the store for %s: %w", name, err)
-	}
-
 	// Logging (each processor instance has a unique ID).
 	id := int(instanceID.Inc())
 	log := logp.NewLogger(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{
 		config: config,
 		store:  src,
@@ -87,16 +89,18 @@ func New(cfg *conf.C) (beat.Processor, error) {
 // 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) (Store, context.CancelFunc, error) {
+func getStoreFor(cfg config, log *logp.Logger) (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:
-		logp.L().Warn("using memory store when file is configured")
-		// TODO: Replace place-holder code with a file-store.
-		s, cancel := fileStores.get(cfg.Store.File.ID, cfg)
+		err := os.MkdirAll(paths.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)
 		return s, cancel, nil
 
 	default:
@@ -105,11 +109,6 @@ func getStoreFor(cfg config) (Store, context.CancelFunc, error) {
 	}
 }
 
-var (
-	memStores  = memStoreSet{stores: map[string]*memStore{}, typ: "memory"}
-	fileStores = memStoreSet{stores: map[string]*memStore{}, typ: "file"} // This is a temporary mock.
-)
-
 // noop is a no-op context.CancelFunc.
 func noop() {}
 
@@ -126,9 +125,9 @@ type Store interface {
 }
 
 type CacheEntry struct {
-	key     string
-	value   any
-	expires time.Time
+	Key     string `json:"key"`
+	Value   any    `json:"val"`
+	Expires time.Time
 	index   int
 }
 

diff --git a/libbeat/processors/cache/cache_test.go b/libbeat/processors/cache/cache_test.go
@@ -130,7 +130,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -191,7 +191,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -210,7 +210,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -271,7 +271,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -290,7 +290,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -351,7 +351,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -379,7 +379,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: errors.New("error applying cache get processor: target field 'crowdstrike.metadata_new' already exists and overwrite_keys is false"),
 			},
@@ -441,7 +441,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -465,7 +465,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -527,7 +527,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: nil,
 			},
@@ -547,7 +547,7 @@ var cacheTests = []struct {
 					},
 				},
 				wantCacheVal: map[string]*CacheEntry{
-					"one": {key: "one", value: "metadata_value"},
+					"one": {Key: "one", Value: "metadata_value"},
 				},
 				wantErr: errors.New("error applying cache get processor: expected map but type is string"),
 			},
@@ -613,7 +613,7 @@ func TestCache(t *testing.T) {
 					switch got := p.(*cache).store.(type) {
 					case *memStore:
 						allow := cmp.AllowUnexported(CacheEntry{})
-						ignore := cmpopts.IgnoreFields(CacheEntry{}, "expires", "index")
+						ignore := cmpopts.IgnoreFields(CacheEntry{}, "Expires", "index")
 						if !cmp.Equal(step.wantCacheVal, got.cache, allow, ignore) {
 							t.Errorf("unexpected cache state result %d:\n--- want\n+++ got\n%s", i, cmp.Diff(step.wantCacheVal, got.cache, allow, ignore))
 						}