libbeat/processors/cache: don't write cache states that have not been altered

CHANGELOG-developer.next.asciidoc

@@ -173,7 +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}36686[36686]
+- Add file-backed cache for cache enrichment processor. {pull}36686[36686] {pull}36696[36696]
 
 ==== Deprecated
 

libbeat/processors/cache/file_store.go

@@ -190,8 +190,8 @@
 		var e CacheEntry
 		err = dec.Decode(&e)
 		if err != nil {
 			if err != io.EOF {
 				switch err := err.(type) {
 				case *json.SyntaxError:
 					c.log.Errorw("failed to read state element", "error", err, "path", c.path, "offset", err.Offset)
 				default:
@@ -202,6 +202,7 @@
 		}
 		if e.Expires.Before(time.Now()) {
 			// Don't retain expired elements.
+			c.dirty = true // The cache now does not reflect the file.
 			continue
 		}
 		c.cache[e.Key] = &e
@@ -239,6 +240,9 @@
 // writeState writes the current cache state to the backing file.
 // If final is true and the cache is empty, the file will be deleted.
 func (c *fileStore) writeState(final bool) {
+	if !c.dirty {
+		return
+	}
 	if len(c.cache) == 0 && final {
 		err := os.Remove(c.path)
 		if err != nil {
@@ -291,4 +295,6 @@
 			return
 		}
 	}
+	// Only mark as not dirty if we succeeded in the write.
+	c.dirty = false
 }

libbeat/processors/cache/file_store_test.go

@@ -196,6 +196,7 @@
 					id:     "test",
 					cache:  map[string]*CacheEntry{},
 					refs:   2,
+					dirty:  false,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -221,6 +222,7 @@
 						{Key: "three", Value: int(3), index: 2},
 					},
 					refs:   2,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -247,6 +249,7 @@
 						{Key: "three", Value: int(3), index: 2},
 					},
 					refs:   2,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -267,6 +270,7 @@
 						{Key: "three", Value: int(3), index: 1},
 					},
 					refs:   2,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -291,6 +295,7 @@
 						{Key: "three", Value: int(3), index: 1},
 					},
 					refs:   2,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -315,6 +320,7 @@
 						{Key: "three", Value: int(3), index: 1},
 					},
 					refs:   1,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -333,6 +339,7 @@
 					cache:    nil, // assistively nil-ed.
 					expiries: nil, // assistively nil-ed.
 					refs:     0,
+					dirty:    false,
 					ttl:      time.Second,
 					cap:      1000,
 					effort:   10,
@@ -403,7 +410,7 @@
 				var e CacheEntry
 				err = dec.Decode(&e)
 				if err != nil {
 					if err != io.EOF {
 						t.Fatalf("unexpected error reading persisted cache data: %v", err)
 					}
 					break

libbeat/processors/cache/mem_store.go

@@ -67,6 +67,9 @@ type memStore struct {
 	expiries expiryHeap
 	ttl      time.Duration // ttl is the time entries are valid for in the cache.
 	refs     int           // refs is the number of processors referring to this store.
+	// dirty marks the cache as changed from the
+	// state in a backing file if it exists.
+	dirty bool
 
 	// id is the index into global cache store for the cache.
 	id string
@@ -176,6 +179,7 @@ func (c *memStore) Put(key string, val any) error {
 	}
 	c.cache[key] = e
 	heap.Push(&c.expiries, e)
+	c.dirty = true
 	return nil
 }
 
@@ -212,6 +216,7 @@ func (c *memStore) Delete(key string) error {
 	}
 	heap.Remove(&c.expiries, v.index)
 	delete(c.cache, key)
+	c.dirty = true
 	return nil
 }
 

libbeat/processors/cache/mem_store_test.go

@@ -184,6 +184,7 @@ var memStoreTests = []struct {
 					id:     "test",
 					cache:  map[string]*CacheEntry{},
 					refs:   2,
+					dirty:  false,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -209,6 +210,7 @@ var memStoreTests = []struct {
 						{Key: "three", Value: int(3), index: 2},
 					},
 					refs:   2,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -235,6 +237,7 @@ var memStoreTests = []struct {
 						{Key: "three", Value: int(3), index: 2},
 					},
 					refs:   2,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -255,6 +258,7 @@ var memStoreTests = []struct {
 						{Key: "three", Value: int(3), index: 1},
 					},
 					refs:   2,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -279,6 +283,7 @@ var memStoreTests = []struct {
 						{Key: "three", Value: int(3), index: 1},
 					},
 					refs:   2,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -303,6 +308,7 @@ var memStoreTests = []struct {
 						{Key: "three", Value: int(3), index: 1},
 					},
 					refs:   1,
+					dirty:  true,
 					ttl:    time.Second,
 					cap:    1000,
 					effort: 10,
@@ -321,6 +327,7 @@ var memStoreTests = []struct {
 					cache:    nil, // assistively nil-ed.
 					expiries: nil, // assistively nil-ed.
 					refs:     0,
+					dirty:    true,
 					ttl:      time.Second,
 					cap:      1000,
 					effort:   10,