diff --git a/event-exporter/README.md b/event-exporter/README.md index cfe2ddc06..f265eee3c 100644 --- a/event-exporter/README.md +++ b/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 diff --git a/event-exporter/event_exporter.go b/event-exporter/event_exporter.go index cf1250fae..ecda83d1c 100644 --- a/event-exporter/event_exporter.go +++ b/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, diff --git a/event-exporter/kubernetes/podlabels/label_utils.go b/event-exporter/kubernetes/podlabels/label_utils.go index 6a658d91b..a6edf6c71 100644 --- a/event-exporter/kubernetes/podlabels/label_utils.go +++ b/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 diff --git a/event-exporter/kubernetes/podlabels/pod_labels_informer.go b/event-exporter/kubernetes/podlabels/pod_labels_informer.go index 658caa11e..969fec739 100644 --- a/event-exporter/kubernetes/podlabels/pod_labels_informer.go +++ b/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 diff --git a/event-exporter/kubernetes/podlabels/pod_labels_informer_test.go b/event-exporter/kubernetes/podlabels/pod_labels_informer_test.go index 3a95375ba..c921a3c4e 100644 --- a/event-exporter/kubernetes/podlabels/pod_labels_informer_test.go +++ b/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) + } +} diff --git a/event-exporter/kubernetes/watchers/events/metrics.go b/event-exporter/kubernetes/watchers/events/metrics.go new file mode 100644 index 000000000..0d44ddcdc --- /dev/null +++ b/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() +} diff --git a/event-exporter/kubernetes/watchers/events/sharding_handler.go b/event-exporter/kubernetes/watchers/events/sharding_handler.go new file mode 100644 index 000000000..56ae9035f --- /dev/null +++ b/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) +} diff --git a/event-exporter/kubernetes/watchers/events/sharding_handler_test.go b/event-exporter/kubernetes/watchers/events/sharding_handler_test.go new file mode 100644 index 000000000..8698f7fc8 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/event-exporter/main.go b/event-exporter/main.go index 343838ec5..0f9e1cbe1 100644 --- a/event-exporter/main.go +++ b/event-exporter/main.go @@ -28,11 +28,15 @@ import ( "github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/kubernetes/podlabels" "github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/kubernetes/watchers" + "github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/sharding" + "github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/sinks" + "github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/sinks/local" "github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/sinks/stackdriver" "github.com/golang/glog" "github.com/prometheus/client_golang/prometheus/promhttp" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes" "k8s.io/client-go/metadata" "k8s.io/client-go/rest" @@ -51,6 +55,11 @@ var ( listerWatcherOptionsLimit = flag.Int64("lister-watcher-options-limit", 100, "Maximum number of responses to return for a list call on events watch. Larger the number, higher the memory event-exporter will consume. No limits when set to 0.") listerWatcherEnableStreaming = flag.Bool("lister-watcher-enable-streaming", false, "Enable watch streaming for lister watcher to prevent all the unhandled events get loaded into memory at once. Instead, events will be processed one by one. If this flag is set to true, lister-watcher-options-limit will be ignored.") storageType = flag.String("storage-type", "DeltaFIFOStorage", "What storage should be used as a cache for the watcher. Supported sotrage type: SimpleStorage, TTLStorage and DeltaFIFOStorage.") + + totalShards = flag.Int("total-shards", 1, "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.") + shardID = flag.Int("shard-id", -1, "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.") + + sinkType = flag.String("sink", "stackdriver", "Sink to export events to. Supported: stackdriver, local (writes events as JSON lines to stdout, for testing).") ) func newSystemStopChannel() chan struct{} { @@ -106,15 +115,39 @@ func main() { glog.Fatalf("Failed to initialize metadata client: %v", err) } + sharder, err := sharding.NewFromFlags(*shardID, *totalShards) + if err != nil { + glog.Fatalf("Failed to initialize sharding: %v", err) + } + if sharder.Enabled() { + glog.Infof("Sharding enabled: this replica is shard %d of %d, events are sharded by involved object UID", sharder.ShardID(), sharder.TotalShards()) + } + var informer podlabels.PodLabelCollector = nil stopCh := newSystemStopChannel() if *enablePodOwnerLabel { - factory := podlabels.NewPodLabelsSharedInformerFactory(metadataClient, strings.Split(*systemNamespaces, ","), *listerWatcherEnableStreaming) + // Shard the pod label cache by the same key as events, so each + // replica only caches the pods whose events it exports. + var owns func(types.UID) bool + if sharder.Enabled() { + owns = sharder.Owns + } + factory := podlabels.NewPodLabelsSharedInformerFactory(metadataClient, strings.Split(*systemNamespaces, ","), *listerWatcherEnableStreaming, owns) informer = factory.NewPodLabelsSharedInformer() factory.Run(stopCh) } - sink, err := stackdriver.NewSdSinkFactory().CreateNew(strings.Split(*sinkOpts, " "), informer) + var sinkFactory sinks.SinkFactory + switch *sinkType { + case "stackdriver": + sinkFactory = stackdriver.NewSdSinkFactory() + case "local": + sinkFactory = local.NewFactory() + default: + glog.Fatalf("Unsupported sink type: %v", *sinkType) + } + + sink, err := sinkFactory.CreateNew(strings.Split(*sinkOpts, " "), informer) if err != nil { glog.Fatalf("Failed to initialize sink: %v", err) } @@ -136,7 +169,7 @@ func main() { glog.Fatalf("Unsupported storage type:%v.", *storageType) } - eventExporter := newEventExporter(client, sink, *resyncPeriod, parsedLabelSelector, *listerWatcherOptionsLimit, *listerWatcherEnableStreaming, st) + eventExporter := newEventExporter(client, sink, *resyncPeriod, parsedLabelSelector, *listerWatcherOptionsLimit, *listerWatcherEnableStreaming, st, sharder) // Expose the Prometheus http endpoint go func() { diff --git a/event-exporter/sharding/sharding.go b/event-exporter/sharding/sharding.go new file mode 100644 index 000000000..017639675 --- /dev/null +++ b/event-exporter/sharding/sharding.go @@ -0,0 +1,110 @@ +/* +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 sharding + +import ( + "fmt" + "hash/fnv" + "os" + "regexp" + "strconv" + + "k8s.io/apimachinery/pkg/types" +) + +// Sharder deterministically assigns objects to one of totalShards shards by +// hashing their UID. All replicas must run with the same totalShards so that +// every object is owned by exactly one replica. +type Sharder struct { + shardID uint32 + totalShards uint32 +} + +// New creates a Sharder for the shard shardID out of totalShards. +func New(shardID, totalShards int) (*Sharder, error) { + if totalShards < 1 { + return nil, fmt.Errorf("total shards must be at least 1, got %d", totalShards) + } + if shardID < 0 || shardID >= totalShards { + return nil, fmt.Errorf("shard ID must be in [0, %d), got %d", totalShards, shardID) + } + return &Sharder{ + shardID: uint32(shardID), + totalShards: uint32(totalShards), + }, nil +} + +// NewFromFlags creates a Sharder from the command line flag values. A shardID +// of -1 derives the shard ID from the ordinal suffix of the pod hostname, +// which works out of the box for StatefulSet replicas. +func NewFromFlags(shardID, totalShards int) (*Sharder, error) { + if shardID == -1 { + if totalShards == 1 { + shardID = 0 + } else { + hostname, err := os.Hostname() + if err != nil { + return nil, fmt.Errorf("failed to get hostname to derive shard ID: %v", err) + } + shardID, err = shardIDFromHostname(hostname) + if err != nil { + return nil, err + } + } + } + return New(shardID, totalShards) +} + +// Enabled reports whether sharding is active, i.e. there is more than one +// shard. A nil Sharder behaves as a single shard that owns everything. +func (s *Sharder) Enabled() bool { + return s != nil && s.totalShards > 1 +} + +// ShardID returns the ID of this shard. +func (s *Sharder) ShardID() int { + return int(s.shardID) +} + +// TotalShards returns the total number of shards. +func (s *Sharder) TotalShards() int { + return int(s.totalShards) +} + +// Owns reports whether the object with the given UID belongs to this shard. +// An empty UID deterministically maps to one fixed shard, so replicas never +// disagree on ownership. +func (s *Sharder) Owns(uid types.UID) bool { + if !s.Enabled() { + return true + } + h := fnv.New32a() + h.Write([]byte(uid)) + return h.Sum32()%s.totalShards == s.shardID +} + +var hostnameOrdinalMatcher = regexp.MustCompile(`-([0-9]+)$`) + +// shardIDFromHostname extracts the shard ID from a StatefulSet pod hostname +// of the form -. +func shardIDFromHostname(hostname string) (int, error) { + m := hostnameOrdinalMatcher.FindStringSubmatch(hostname) + if m == nil { + return 0, fmt.Errorf("hostname %q has no ordinal suffix to derive shard ID from; set shard-id explicitly", hostname) + } + return strconv.Atoi(m[1]) +} diff --git a/event-exporter/sharding/sharding_test.go b/event-exporter/sharding/sharding_test.go new file mode 100644 index 000000000..41c861c84 --- /dev/null +++ b/event-exporter/sharding/sharding_test.go @@ -0,0 +1,124 @@ +/* +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 sharding + +import ( + "fmt" + "testing" + + "k8s.io/apimachinery/pkg/types" +) + +func TestNewValidation(t *testing.T) { + testCases := []struct { + shardID int + totalShards int + wantErr bool + }{ + {0, 1, false}, + {2, 3, false}, + {0, 0, true}, + {-1, 3, true}, + {3, 3, true}, + } + for _, tc := range testCases { + _, err := New(tc.shardID, tc.totalShards) + if gotErr := err != nil; gotErr != tc.wantErr { + t.Errorf("New(%d, %d) error = %v, wantErr %v", tc.shardID, tc.totalShards, err, tc.wantErr) + } + } +} + +func TestEveryUIDOwnedByExactlyOneShard(t *testing.T) { + const totalShards = 3 + sharders := make([]*Sharder, totalShards) + for i := range sharders { + s, err := New(i, totalShards) + if err != nil { + t.Fatalf("New(%d, %d) failed: %v", i, totalShards, err) + } + sharders[i] = s + } + + ownedPerShard := make([]int, totalShards) + for i := 0; i < 1000; i++ { + uid := types.UID(fmt.Sprintf("uid-%d", i)) + owners := 0 + for shard, s := range sharders { + if s.Owns(uid) { + owners++ + ownedPerShard[shard]++ + } + } + if owners != 1 { + t.Errorf("UID %q owned by %d shards, want exactly 1", uid, owners) + } + } + + // FNV-1a should spread UIDs roughly evenly; guard against a degenerate + // distribution rather than asserting exact counts. + for shard, count := range ownedPerShard { + if count < 200 { + t.Errorf("shard %d owns only %d of 1000 UIDs, distribution is too skewed", shard, count) + } + } +} + +func TestDisabledSharderOwnsEverything(t *testing.T) { + s, err := New(0, 1) + if err != nil { + t.Fatalf("New(0, 1) failed: %v", err) + } + if s.Enabled() { + t.Error("Sharder with a single shard should not be enabled") + } + if !s.Owns("any-uid") || !s.Owns("") { + t.Error("Sharder with a single shard should own every UID") + } + + var nilSharder *Sharder + if nilSharder.Enabled() { + t.Error("nil Sharder should not be enabled") + } + if !nilSharder.Owns("any-uid") { + t.Error("nil Sharder should own every UID") + } +} + +func TestShardIDFromHostname(t *testing.T) { + testCases := []struct { + hostname string + want int + wantErr bool + }{ + {"event-exporter-0", 0, false}, + {"event-exporter-12", 12, false}, + {"event-exporter-v0.4.1-5", 5, false}, + {"event-exporter", 0, true}, + {"", 0, true}, + } + for _, tc := range testCases { + got, err := shardIDFromHostname(tc.hostname) + if gotErr := err != nil; gotErr != tc.wantErr { + t.Errorf("shardIDFromHostname(%q) error = %v, wantErr %v", tc.hostname, err, tc.wantErr) + continue + } + if !tc.wantErr && got != tc.want { + t.Errorf("shardIDFromHostname(%q) = %d, want %d", tc.hostname, got, tc.want) + } + } +} diff --git a/event-exporter/sinks/local/sink.go b/event-exporter/sinks/local/sink.go new file mode 100644 index 000000000..2a94f1b9b --- /dev/null +++ b/event-exporter/sinks/local/sink.go @@ -0,0 +1,118 @@ +/* +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 local provides a sink that writes exported events as JSON lines to +// stdout. It is meant for testing the export pipeline (e.g. in a kind +// cluster) without Stackdriver access; each line is prefixed with +// EXPORTED_EVENT so it can be extracted from the pod logs. +package local + +import ( + "encoding/json" + "fmt" + "sync" + + "github.com/golang/glog" + corev1 "k8s.io/api/core/v1" + + "github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/kubernetes/podlabels" + "github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/sinks" +) + +const exportedEventPrefix = "EXPORTED_EVENT" + +type exportedEvent struct { + Action string `json:"action"` + EventUID string `json:"eventUID"` + EventNamespace string `json:"eventNamespace"` + EventName string `json:"eventName"` + ResourceVersion string `json:"resourceVersion"` + Count int32 `json:"count"` + Reason string `json:"reason"` + InvolvedKind string `json:"involvedKind"` + InvolvedName string `json:"involvedName"` + InvolvedUID string `json:"involvedUID"` + PodLabels map[string]string `json:"podLabels,omitempty"` +} + +type localSink struct { + podLabelCollector podlabels.PodLabelCollector + + mu sync.Mutex +} + +// NewFactory creates a factory for the local testing sink. +func NewFactory() sinks.SinkFactory { + return &localSinkFactory{} +} + +type localSinkFactory struct{} + +func (f *localSinkFactory) CreateNew(opts []string, podLabelCollector podlabels.PodLabelCollector) (sinks.Sink, error) { + return &localSink{ + podLabelCollector: podLabelCollector, + }, nil +} + +func (s *localSink) OnAdd(event *corev1.Event) { + s.export("ADD", event) +} + +func (s *localSink) OnUpdate(_ *corev1.Event, newEvent *corev1.Event) { + s.export("UPDATE", newEvent) +} + +func (s *localSink) OnDelete(*corev1.Event) { + // Deletions are not exported, matching the Stackdriver sink. +} + +func (s *localSink) OnList(*corev1.EventList) { + glog.Info("Local sink received list, started watching") +} + +func (s *localSink) Run(stopCh <-chan struct{}) { + glog.Info("Starting local sink") + <-stopCh + glog.Info("Local sink received stop signal") +} + +func (s *localSink) export(action string, event *corev1.Event) { + exported := exportedEvent{ + Action: action, + EventUID: string(event.UID), + EventNamespace: event.Namespace, + EventName: event.Name, + ResourceVersion: event.ResourceVersion, + Count: event.Count, + Reason: event.Reason, + InvolvedKind: event.InvolvedObject.Kind, + InvolvedName: event.InvolvedObject.Name, + InvolvedUID: string(event.InvolvedObject.UID), + } + // Enrich pod events with owner labels the same way the Stackdriver + // sink does, so label lookup can be tested end to end. + if event.InvolvedObject.Kind == "Pod" && s.podLabelCollector != nil { + exported.PodLabels = s.podLabelCollector.GetLabels(event.InvolvedObject.Namespace, event.InvolvedObject.Name) + } + line, err := json.Marshal(exported) + if err != nil { + glog.Warningf("Failed to marshal exported event %+v: %v", exported, err) + return + } + s.mu.Lock() + defer s.mu.Unlock() + fmt.Printf("%s %s\n", exportedEventPrefix, line) +}