fix(executor): drop event watcher cache buckets when their last event goes

executor/pkg/plugin/k8s/event_watcher.go

@@ -47,6 +47,12 @@ type controllerRuntimeEventWatcher struct {
 type eventObjects struct {
 	mu         sync.RWMutex
 	eventInfos map[k8stypes.NamespacedName]*eventInfo
+	// evicted marks a bucket that has been taken out of objectCache because its last
+	// event was deleted. A writer that loaded the bucket before it was removed must not
+	// add to it: the bucket is no longer reachable, so the event would be dropped. It
+	// sees this flag under the same lock the eviction is done under and retries against
+	// a fresh bucket instead.
+	evicted bool
 }
 
 func newControllerRuntimeEventWatcher(ctx context.Context, cache ctrlcache.Cache) (*controllerRuntimeEventWatcher, error) {
@@ -94,11 +100,6 @@ func (w *controllerRuntimeEventWatcher) store(obj interface{}) {
 
 	eventKey := k8stypes.NamespacedName{Namespace: event.Namespace, Name: event.Name}
 
-	value, _ := w.objectCache.LoadOrStore(objectKey, &eventObjects{
-		eventInfos: make(map[k8stypes.NamespacedName]*eventInfo),
-	})
-	eventInfos := value.(*eventObjects)
-
 	info := &eventInfo{
 		Message:        event.Note,
 		CreatedAt:      event.CreationTimestamp.Time,
@@ -108,19 +109,36 @@ func (w *controllerRuntimeEventWatcher) store(obj interface{}) {
 		LastObservedAt: lastObservedTime(event),
 	}
 
-	eventInfos.mu.Lock()
-	defer eventInfos.mu.Unlock()
+	// A bucket can be evicted between the load and the lock, so the store is retried
+	// until it lands in one that is still reachable. The loop turns at most once per
+	// concurrent eviction of this object's bucket, and an eviction only happens when the
+	// bucket is empty, so it cannot spin.
+	for {
+		value, _ := w.objectCache.LoadOrStore(objectKey, &eventObjects{
+			eventInfos: make(map[k8stypes.NamespacedName]*eventInfo),
+		})
+		eventInfos := value.(*eventObjects)
+
+		eventInfos.mu.Lock()
+		if eventInfos.evicted {
+			eventInfos.mu.Unlock()
+			continue
+		}
 
-	if existing, ok := eventInfos.eventInfos[eventKey]; ok {
-		info.CreatedAt = existing.CreatedAt
-		info.RecordedAt = existing.RecordedAt
-		if existing.LastObservedAt.After(info.LastObservedAt) {
-			info.LastObservedAt = existing.LastObservedAt
+		stored := *info
+		if existing, ok := eventInfos.eventInfos[eventKey]; ok {
+			stored.CreatedAt = existing.CreatedAt
+			stored.RecordedAt = existing.RecordedAt
+			if existing.LastObservedAt.After(stored.LastObservedAt) {
+				stored.LastObservedAt = existing.LastObservedAt
+			}
 		}
+		// The entry is replaced rather than mutated: List hands out these pointers, and a
+		// reader may still be looking at the one it got.
+		eventInfos.eventInfos[eventKey] = &stored
+		eventInfos.mu.Unlock()
+		return
 	}
-	// The entry is replaced rather than mutated: List hands out these pointers, and a
-	// reader may still be looking at the one it got.
-	eventInfos.eventInfos[eventKey] = info
 }
 
 // lastObservedTime is the freshest occurrence the event reports. An aggregated event
@@ -178,8 +196,20 @@ func (w *controllerRuntimeEventWatcher) OnDelete(obj interface{}) {
 	defer eventInfos.mu.Unlock()
 
 	delete(eventInfos.eventInfos, eventKey)
-	// We intentionally do not delete empty buckets from objectCache. This avoids races where
-	// a new event is being added to the bucket while the top-level map entry is concurrently removed.
+
+	// The bucket goes when its last event does, so objectCache holds an entry only for
+	// objects with a live event rather than for every object seen since startup.
+	//
+	// Removing it is safe against a concurrent store because both the flag and the
+	// removal happen under this bucket's write lock: a writer that already loaded this
+	// bucket blocks here, then sees evicted and retries against a fresh one, and a writer
+	// that has not loaded it yet cannot reach it once it is out of the map. Taking the
+	// map's lock while holding the bucket's cannot deadlock, because every other path
+	// releases the map's lock before it takes a bucket's.
+	if len(eventInfos.eventInfos) == 0 {
+		eventInfos.evicted = true
+		w.objectCache.Delete(objectKey)
+	}
 }
 
 // List returns the cached events for an object, ordered by when they were created. The

executor/pkg/plugin/k8s/event_watcher_test.go

@@ -1,6 +1,8 @@
 package k8s
 
 import (
+	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -102,3 +104,84 @@ func TestEventWatcherOnUpdateKeepsTheFreshestObservation(t *testing.T) {
 	require.Len(t, events, 1)
 	assert.WithinDuration(t, lastObserved, events[0].LastObservedAt, time.Microsecond)
 }
+
+func cachedBuckets(w *controllerRuntimeEventWatcher) int {
+	count := 0
+	w.objectCache.Range(func(_, _ any) bool {
+		count++
+		return true
+	})
+	return count
+}
+
+func TestEventWatcherDropsBucketWhenLastEventIsDeleted(t *testing.T) {
+	watcher := &controllerRuntimeEventWatcher{}
+	createdAt := time.Now()
+
+	watcher.OnAdd(testEvent("event-1", "pod-uid", createdAt), false)
+	watcher.OnAdd(testEvent("event-2", "pod-uid", createdAt), false)
+	require.Equal(t, 1, cachedBuckets(watcher))
+
+	// The bucket stays while any event for the object is still cached.
+	watcher.OnDelete(testEvent("event-1", "pod-uid", createdAt))
+	assert.Equal(t, 1, cachedBuckets(watcher))
+
+	// The last deletion takes the bucket with it, so an object whose events have all
+	// expired stops costing anything.
+	watcher.OnDelete(testEvent("event-2", "pod-uid", createdAt))
+	assert.Equal(t, 0, cachedBuckets(watcher))
+	assert.Empty(t, watcher.List(watchedObjectKey{Namespace: "ns", Name: "pod", Kind: "Pod"}, time.Time{}, time.Time{}))
+}
+
+func TestEventWatcherDoesNotGrowAcrossObjectChurn(t *testing.T) {
+	watcher := &controllerRuntimeEventWatcher{}
+	createdAt := time.Now()
+
+	// Every object the cluster emits an event about used to leave a bucket behind for
+	// the life of the process; the count now follows only what is still live.
+	for i := 0; i < 1000; i++ {
+		event := testEvent("event-1", "pod-uid", createdAt)
+		event.Regarding.Name = fmt.Sprintf("pod-%d", i)
+		watcher.OnAdd(event, false)
+	}
+	require.Equal(t, 1000, cachedBuckets(watcher))
+
+	for i := 0; i < 1000; i++ {
+		event := testEvent("event-1", "pod-uid", createdAt)
+		event.Regarding.Name = fmt.Sprintf("pod-%d", i)
+		watcher.OnDelete(event)
+	}
+	assert.Equal(t, 0, cachedBuckets(watcher))
+}
+
+func TestEventWatcherStoreSurvivesConcurrentEviction(t *testing.T) {
+	watcher := &controllerRuntimeEventWatcher{}
+	createdAt := time.Now()
+
+	// A store that loses its bucket to a concurrent eviction has to land in a fresh one
+	// rather than write into a bucket nothing can reach again. Run under -race.
+	var wg sync.WaitGroup
+	for i := 0; i < 200; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			watcher.OnAdd(testEvent("event-keep", "pod-uid", createdAt), false)
+		}()
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			watcher.OnAdd(testEvent("event-churn", "pod-uid", createdAt), false)
+			watcher.OnDelete(testEvent("event-churn", "pod-uid", createdAt))
+		}()
+	}
+	wg.Wait()
+
+	// The surviving event must still be reachable: an eviction that raced a store must
+	// not have swallowed it.
+	watcher.OnAdd(testEvent("event-keep", "pod-uid", createdAt), false)
+	events := watcher.List(watchedObjectKey{Namespace: "ns", Name: "pod", Kind: "Pod"}, time.Time{}, time.Time{})
+	require.Len(t, events, 1)
+	assert.Equal(t, k8stypes.UID("pod-uid"), events[0].RegardingUID)
+	assert.Equal(t, 1, cachedBuckets(watcher))
+}