Sharding

event-exporter/README.md

@@ -74,6 +74,35 @@ spec:
         - '/event-exporter'
 ```
 
+## Sharding
+
+On large clusters a single replica may not fit in its memory limit, mostly
+because of the cluster-wide pod metadata cache used for owner labels. Event
+exporter can be sharded across multiple replicas:
+
+```
+-total-shards int
+    Total number of event-exporter replicas (shards). Each event is exported
+    by exactly one shard, chosen by hashing the involved object's UID. All
+    replicas must run with the same value. 1 disables sharding. (default 1)
+-shard-id int
+    ID of this shard, in [0, total-shards). -1 derives the ID from the
+    ordinal suffix of the pod hostname, which works for StatefulSet
+    replicas. (default -1)
+```
+
+Deploy the shards as a StatefulSet with `replicas` equal to `-total-shards`;
+each pod picks up its shard ID from its hostname ordinal. Each replica still
+receives the full event and pod watch streams from the apiserver and filters
+them locally, so sharding divides memory usage per replica but multiplies
+apiserver watch load by the number of shards. When `-enable-pod-owner-label`
+is on, the pod label cache is sharded by the same key (events are sharded by
+involved object UID, which for pod events is the pod UID), so each replica
+only caches the pods whose events it exports.
+
+Note: changing `-total-shards` reassigns shard ownership, so events may be
+duplicated or missed while the rollout is in progress.
+
 ## Notes
 ### ClusterRoleBinding
 This pod's service account should be authorized to get events, you

event-exporter/event_exporter.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/kubernetes/watchers"
 	"github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/kubernetes/watchers/events"
+	"github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/sharding"
 	"github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/sinks"
 	"github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/utils"
 )
@@ -37,18 +38,22 @@ func (e *eventExporter) Run(stopCh <-chan struct{}) {
 	utils.RunConcurrentlyUntil(stopCh, e.sink.Run, e.watcher.Run)
 }
 
-func newEventExporter(client kubernetes.Interface, sink sinks.Sink, resyncPeriod time.Duration, eventLabelSelector labels.Selector, listerWatcherOptionsLimit int64, listerWatcherEnableStreaming bool, storageType watchers.StorageType) *eventExporter {
+func newEventExporter(client kubernetes.Interface, sink sinks.Sink, resyncPeriod time.Duration, eventLabelSelector labels.Selector, listerWatcherOptionsLimit int64, listerWatcherEnableStreaming bool, storageType watchers.StorageType, sharder *sharding.Sharder) *eventExporter {
 	return &eventExporter{
 		sink:    sink,
-		watcher: createWatcher(client, sink, resyncPeriod, eventLabelSelector, listerWatcherOptionsLimit, listerWatcherEnableStreaming, storageType),
+		watcher: createWatcher(client, sink, resyncPeriod, eventLabelSelector, listerWatcherOptionsLimit, listerWatcherEnableStreaming, storageType, sharder),
 	}
 }
 
-func createWatcher(client kubernetes.Interface, sink sinks.Sink, resyncPeriod time.Duration, eventLabelSelector labels.Selector, listerWatcherOptionsLimit int64, listerWatcherEnableStreaming bool, storageType watchers.StorageType) watchers.Watcher {
+func createWatcher(client kubernetes.Interface, sink sinks.Sink, resyncPeriod time.Duration, eventLabelSelector labels.Selector, listerWatcherOptionsLimit int64, listerWatcherEnableStreaming bool, storageType watchers.StorageType, sharder *sharding.Sharder) watchers.Watcher {
+	var handler events.EventHandler = sink
+	if sharder.Enabled() {
+		handler = events.NewShardingHandler(sink, sharder.Owns)
+	}
 	return events.NewEventWatcher(client, &events.EventWatcherConfig{
 		OnList:                       sink.OnList,
 		ResyncPeriod:                 resyncPeriod,
-		Handler:                      sink,
+		Handler:                      handler,
 		EventLabelSelector:           eventLabelSelector,
 		ListerWatcherOptionsLimit:    listerWatcherOptionsLimit,
 		ListerWatcherEnableStreaming: listerWatcherEnableStreaming,

event-exporter/kubernetes/podlabels/label_utils.go

@@ -12,6 +12,16 @@ const (
 	jobsetUIDLabelKey            = "jobset.sigs.k8s.io/jobset-uid"
 )
 
+// ownerKindsWithLabels is the set of owner reference kinds from which
+// getLabelsFromMeta derives owner labels. Owner references of other kinds
+// don't need to be cached. Keep in sync with the switch in getLabelsFromMeta.
+var ownerKindsWithLabels = map[string]struct{}{
+	"DaemonSet":   {},
+	"StatefulSet": {},
+	"ReplicaSet":  {},
+	"Job":         {},
+}
+
 // matches suffixes containing number between 20000000 to 59999999
 // These 2 numbers are chosen because the convenience of regex matching:
 // 20000000: Thu Jan 10 2008 21:20:00 GMT+0000  in minutes since Unix Epoch

event-exporter/kubernetes/podlabels/pod_labels_informer.go

@@ -5,6 +5,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	clientfeatures "k8s.io/client-go/features"
 	"k8s.io/client-go/metadata"
 
@@ -56,7 +57,11 @@ func (c *customFeatureGates) Enabled(key clientfeatures.Feature) bool {
 	return c.Gates.Enabled(key)
 }
 
-func NewPodLabelsSharedInformerFactory(client metadata.Interface, ignoredNamespaces []string, enableWatchListClient bool) *PodLabelsSharedInformerFactory {
+// NewPodLabelsSharedInformerFactory creates a factory for the pod labels
+// informer. If owns is non-nil, only pods whose UID satisfies owns are
+// cached; events are sharded by involved object UID, which for pod events is
+// the pod UID, so the shard that exports an event always caches its pod.
+func NewPodLabelsSharedInformerFactory(client metadata.Interface, ignoredNamespaces []string, enableWatchListClient bool, owns func(types.UID) bool) *PodLabelsSharedInformerFactory {
 	// Set the custom feature gates based on the flag
 	clientfeatures.ReplaceFeatureGates(&customFeatureGates{
 		Gates:                 clientfeatures.FeatureGates(),
@@ -76,6 +81,9 @@ func NewPodLabelsSharedInformerFactory(client metadata.Interface, ignoredNamespa
 				if _, ok := ignoredNamespacesMap[meta.Namespace]; ok {
 					return nil, nil
 				}
+				if owns != nil && !owns(meta.UID) {
+					return nil, nil
+				}
 				labels := make(map[string]string)
 				if v, ok := meta.Labels["pod-template-hash"]; ok {
 					labels["pod-template-hash"] = v
@@ -89,11 +97,24 @@ func NewPodLabelsSharedInformerFactory(client metadata.Interface, ignoredNamespa
 				if v, ok := meta.Labels[jobsetUIDLabelKey]; ok {
 					labels[jobsetUIDLabelKey] = v
 				}
+				if len(labels) == 0 {
+					labels = nil
+				}
+				// Cache only the owner reference fields getLabelsFromMeta reads.
+				var owners []metav1.OwnerReference
+				for _, owner := range meta.OwnerReferences {
+					if _, ok := ownerKindsWithLabels[owner.Kind]; ok {
+						owners = append(owners, metav1.OwnerReference{
+							Kind: owner.Kind,
+							Name: owner.Name,
+						})
+					}
+				}
 				return &metav1.PartialObjectMetadata{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:            meta.Name,
 						Namespace:       meta.Namespace,
-						OwnerReferences: meta.OwnerReferences,
+						OwnerReferences: owners,
 						Labels:          labels,
 					},
 				}, nil

event-exporter/kubernetes/podlabels/pod_labels_informer_test.go

@@ -10,6 +10,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	metadatafake "k8s.io/client-go/metadata/fake"
 )
 
@@ -234,7 +235,7 @@ func TestGetLabelsCacheOperations(t *testing.T) {
 	}
 	fakeMetadataClient := metadatafake.NewSimpleMetadataClient(scheme, pod1, pod2)
 
-	factory := NewPodLabelsSharedInformerFactory(fakeMetadataClient, nil, false)
+	factory := NewPodLabelsSharedInformerFactory(fakeMetadataClient, nil, false, nil)
 	collector := factory.NewPodLabelsSharedInformer()
 	stopCh := make(chan struct{})
 	defer close(stopCh)
@@ -322,3 +323,54 @@ func TestGetLabelsCacheOperations(t *testing.T) {
 		t.Errorf("At this point, cacheOpsCount with operation=evict should be 1, but got %d", count)
 	}
 }
+
+func TestGetLabelsCacheSharding(t *testing.T) {
+	scheme := runtime.NewScheme()
+	metav1.AddMetaToScheme(scheme)
+
+	ownedPod := &metav1.PartialObjectMetadata{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "owned-pod",
+			Namespace: "default",
+			UID:       "owned-uid",
+			OwnerReferences: []metav1.OwnerReference{{
+				Kind: "DaemonSet",
+				Name: "my-agent",
+			}},
+		},
+	}
+	foreignPod := &metav1.PartialObjectMetadata{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "foreign-pod",
+			Namespace: "default",
+			UID:       "foreign-uid",
+			OwnerReferences: []metav1.OwnerReference{{
+				Kind: "DaemonSet",
+				Name: "my-agent",
+			}},
+		},
+	}
+	fakeMetadataClient := metadatafake.NewSimpleMetadataClient(scheme, ownedPod, foreignPod)
+
+	owns := func(uid types.UID) bool { return uid == "owned-uid" }
+	factory := NewPodLabelsSharedInformerFactory(fakeMetadataClient, nil, false, owns)
+	collector := factory.NewPodLabelsSharedInformer()
+	stopCh := make(chan struct{})
+	defer close(stopCh)
+	factory.Run(stopCh)
+
+	if labels := collector.GetLabels("default", "owned-pod"); len(labels) != 2 {
+		t.Errorf("GetLabels() for owned pod returned unexpected labels %v", labels)
+	}
+	if labels := collector.GetLabels("default", "foreign-pod"); labels != nil {
+		t.Errorf("GetLabels() for pod of another shard should miss, but returned %v", labels)
+	}
+}

event-exporter/kubernetes/watchers/events/metrics.go

@@ -0,0 +1,37 @@
+/*
+Copyright 2026 Google Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package events
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var shardingFilteredEventsCount = prometheus.NewCounter(
+	prometheus.CounterOpts{
+		Name:      "filtered_events_count",
+		Help:      "Number of events skipped because they belong to another shard",
+		Subsystem: "sharding",
+	},
+)
+
+func init() {
+	prometheus.MustRegister(shardingFilteredEventsCount)
+}
+
+func recordShardingFilteredEvent() {
+	shardingFilteredEventsCount.Inc()
+}

event-exporter/kubernetes/watchers/events/sharding_handler.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2026 Google Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package events
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// shardingHandler forwards only the events owned by this shard to the
+// delegate handler. Events are sharded by the involved object's UID: for pod
+// events that is the pod UID, so an event is always handled by the same shard
+// that caches the pod's metadata for owner label lookup.
+type shardingHandler struct {
+	delegate EventHandler
+	owns     func(types.UID) bool
+}
+
+// NewShardingHandler wraps delegate so that only events whose involved
+// object UID satisfies owns are passed through.
+func NewShardingHandler(delegate EventHandler, owns func(types.UID) bool) EventHandler {
+	return &shardingHandler{
+		delegate: delegate,
+		owns:     owns,
+	}
+}
+
+func (s *shardingHandler) OnAdd(event *corev1.Event) {
+	if !s.owns(event.InvolvedObject.UID) {
+		recordShardingFilteredEvent()
+		return
+	}
+	s.delegate.OnAdd(event)
+}
+
+func (s *shardingHandler) OnUpdate(oldEvent *corev1.Event, newEvent *corev1.Event) {
+	if !s.owns(newEvent.InvolvedObject.UID) {
+		recordShardingFilteredEvent()
+		return
+	}
+	s.delegate.OnUpdate(oldEvent, newEvent)
+}
+
+func (s *shardingHandler) OnDelete(event *corev1.Event) {
+	if !s.owns(event.InvolvedObject.UID) {
+		recordShardingFilteredEvent()
+		return
+	}
+	s.delegate.OnDelete(event)
+}

event-exporter/kubernetes/watchers/events/sharding_handler_test.go

@@ -0,0 +1,73 @@
+/*
+Copyright 2026 Google Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package events
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+func eventForUID(uid types.UID) *corev1.Event {
+	return &corev1.Event{
+		InvolvedObject: corev1.ObjectReference{
+			UID: uid,
+		},
+	}
+}
+
+func TestShardingHandlerFiltersByInvolvedObjectUID(t *testing.T) {
+	const ownedUID = types.UID("owned-uid")
+	owns := func(uid types.UID) bool { return uid == ownedUID }
+
+	testCases := []struct {
+		desc     string
+		uid      types.UID
+		expected bool
+	}{
+		{"owned event is forwarded", ownedUID, true},
+		{"foreign event is dropped", "foreign-uid", false},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			addTriggered := false
+			updateTriggered := false
+			deleteTriggered := false
+			handler := NewShardingHandler(&fakeEventHandler{
+				onAddFunc:    func(*corev1.Event) { addTriggered = true },
+				onUpdateFunc: func(*corev1.Event, *corev1.Event) { updateTriggered = true },
+				onDeleteFunc: func(*corev1.Event) { deleteTriggered = true },
+			}, owns)
+
+			handler.OnAdd(eventForUID(tc.uid))
+			handler.OnUpdate(nil, eventForUID(tc.uid))
+			handler.OnDelete(eventForUID(tc.uid))
+
+			if addTriggered != tc.expected {
+				t.Errorf("Add is triggered = %v, expected %v", addTriggered, tc.expected)
+			}
+			if updateTriggered != tc.expected {
+				t.Errorf("Update is triggered = %v, expected %v", updateTriggered, tc.expected)
+			}
+			if deleteTriggered != tc.expected {
+				t.Errorf("Delete is triggered = %v, expected %v", deleteTriggered, tc.expected)
+			}
+		})
+	}
+}