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

[Daksha1611]

Tracking issue

Related to #7992 (fixes the second of the two problems reported there)

Why are the changes needed?

controllerRuntimeEventWatcher.objectCache had no eviction path at all. store creates one bucket per (namespace, name, kind), OnDelete removes the event from the bucket but deliberately leaves the bucket behind, and nothing anywhere else removes one — no Delete, no LoadAndDelete, no reaper.

The existing comment explains why the bucket was left in place:

// 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.

That race is real, but avoiding it this way means the map grows for the entire life of the process. Kubernetes expires Events after ~1h so the inner maps drain, but the outer buckets never go. The key space is every object the cluster emitted an event about since the executor started, and because buildGeneratedName makes pod names unique per attempt ({action}-{retry}), even the Flyte-only subset grows with cumulative attempts rather than with concurrency — it never reaches a steady state.

Driving store and OnDelete for 200k distinct objects, deleting every event, then forcing a GC:

distinct objects seen:                                    200000
buckets retained in objectCache AFTER all events deleted: 200000
event entries retained inside buckets:                    0
retained heap delta:                                      115.9 MB  (~608 bytes per bucket)

Every inner map is empty and all 200k buckets are still held. On a busy cluster this is a slow, unbounded climb toward the OOMKill that #7831 described.

What changes were proposed in this pull request?

The bucket is now dropped when its last event is deleted, which closes the race rather than avoiding it.

eventObjects gains an evicted flag. Both setting the flag and removing the map entry happen under that bucket's own write lock:

if len(eventInfos.eventInfos) == 0 {
	eventInfos.evicted = true
	w.objectCache.Delete(objectKey)
}

and store retries when it finds the bucket it loaded has been evicted:

for {
	value, _ := w.objectCache.LoadOrStore(objectKey, &eventObjects{...})
	eventInfos := value.(*eventObjects)

	eventInfos.mu.Lock()
	if eventInfos.evicted {
		eventInfos.mu.Unlock()
		continue
	}
	...
}

This is correct in both directions:

• A store that already loaded the doomed bucket blocks on its lock, then observes evicted and retries. Because the entry is already out of the map by then, LoadOrStore gives it a fresh bucket rather than the dead one.
• A store that has not loaded it yet cannot reach the dead bucket at all, since it is no longer in the map.

The loop cannot spin: it turns at most once per concurrent eviction of that object's bucket, and eviction only happens when the bucket is empty.

The lock order is bucket-then-map here, and map-then-bucket nowhere — every other path (Load/LoadOrStore in store, OnDelete, List) releases the map's internal lock before taking a bucket's — so taking objectCache.Delete under the bucket lock cannot deadlock.

One incidental change: store now writes a copy of the eventInfo per attempt instead of mutating and inserting a single shared value, so a retry cannot carry over the merge it did against the bucket it lost.

objectCache is now bounded by objects with a live event instead of by every object seen since startup.

Deliberately not included

This does not address the other half of #7992 — that the Event informer is registered with no ByObject entry and so does a full unfiltered LIST/WATCH of every Event in the cluster. Scoping it needs a maintainer decision about which events the executor should watch (a regarding.kind=Pod field selector, namespace scoping, or a transform), so I have left it to the issue rather than guessing here. Happy to follow up with a second PR once there is a direction.

How was this patch tested?

Added to event_watcher_test.go:

• TestEventWatcherDropsBucketWhenLastEventIsDeleted — the bucket survives while any event for the object remains and goes with the last one.
• TestEventWatcherDoesNotGrowAcrossObjectChurn — 1000 distinct objects added then deleted leaves 0 buckets.
• TestEventWatcherStoreSurvivesConcurrentEviction — 400 goroutines racing stores against evictions on the same object; asserts the surviving event is still reachable, i.e. an eviction that raced a store did not swallow it.

Results:

go test ./executor/pkg/plugin/k8s/ -count=1          ok
go test ./executor/pkg/plugin/k8s/ -race -count=4    ok
go build ./executor/...                              ok
go vet ./executor/pkg/plugin/k8s/                    ok
gofmt -l executor/pkg/plugin/k8s/                    (clean)

Re-running the 200k probe from the issue against this branch:

distinct objects seen:                                    200000
buckets retained in objectCache AFTER all events deleted: 0
retained heap delta:                                      ~0 MB

All pre-existing tests in the package still pass unchanged.

Labels

fixed

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.