diff --git a/executor/pkg/plugin/k8s/event_watcher.go b/executor/pkg/plugin/k8s/event_watcher.go index 3a9e3f6ec0..db436e0bc0 100644 --- a/executor/pkg/plugin/k8s/event_watcher.go +++ b/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 diff --git a/executor/pkg/plugin/k8s/event_watcher_test.go b/executor/pkg/plugin/k8s/event_watcher_test.go index be2ea5c8d0..3f8a8863de 100644 --- a/executor/pkg/plugin/k8s/event_watcher_test.go +++ b/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)) +}