Use client-go's native streaming instead of hand-crafted streaming

event-exporter/event_exporter.go

@@ -37,21 +37,20 @@ 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, storageType watchers.StorageType) *eventExporter {
 	return &eventExporter{
 		sink:    sink,
-		watcher: createWatcher(client, sink, resyncPeriod, eventLabelSelector, listerWatcherOptionsLimit, listerWatcherEnableStreaming, storageType),
+		watcher: createWatcher(client, sink, resyncPeriod, eventLabelSelector, listerWatcherOptionsLimit, storageType),
 	}
 }
 
-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, storageType watchers.StorageType) watchers.Watcher {
 	return events.NewEventWatcher(client, &events.EventWatcherConfig{
-		OnList:                       sink.OnList,
-		ResyncPeriod:                 resyncPeriod,
-		Handler:                      sink,
-		EventLabelSelector:           eventLabelSelector,
-		ListerWatcherOptionsLimit:    listerWatcherOptionsLimit,
-		ListerWatcherEnableStreaming: listerWatcherEnableStreaming,
-		StorageType:                  storageType,
+		OnList:                    sink.OnList,
+		ResyncPeriod:              resyncPeriod,
+		Handler:                   sink,
+		EventLabelSelector:        eventLabelSelector,
+		ListerWatcherOptionsLimit: listerWatcherOptionsLimit,
+		StorageType:               storageType,
 	})
 }

event-exporter/kubernetes/podlabels/pod_labels_informer.go

@@ -5,7 +5,6 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	clientfeatures "k8s.io/client-go/features"
 	"k8s.io/client-go/metadata"
 
 	"k8s.io/client-go/metadata/metadatainformer"
@@ -43,26 +42,11 @@ func (f *PodLabelsSharedInformerFactory) NewPodLabelsSharedInformer() *PodLabels
 	}
 }
 
-type customFeatureGates struct {
-	clientfeatures.Gates
-	enableWatchListClient bool
-}
-
-func (c *customFeatureGates) Enabled(key clientfeatures.Feature) bool {
-	if key == clientfeatures.WatchListClient {
-		klog.Info("Enabled feature gate WatchListClient: ", c.enableWatchListClient)
-		return c.enableWatchListClient
-	}
-	return c.Gates.Enabled(key)
-}
-
-func NewPodLabelsSharedInformerFactory(client metadata.Interface, ignoredNamespaces []string, enableWatchListClient bool) *PodLabelsSharedInformerFactory {
-	// Set the custom feature gates based on the flag
-	clientfeatures.ReplaceFeatureGates(&customFeatureGates{
-		Gates:                 clientfeatures.FeatureGates(),
-		enableWatchListClient: enableWatchListClient,
-	})
-
+// NewPodLabelsSharedInformerFactory creates a factory for the pod-labels
+// informer. Whether the informer syncs via a watch-list stream or paginated
+// lists is controlled by the global client-go WatchListClient feature gate,
+// set up in main.
+func NewPodLabelsSharedInformerFactory(client metadata.Interface, ignoredNamespaces []string) *PodLabelsSharedInformerFactory {
 	ignoredNamespacesMap := make(map[string]struct{})
 	for _, ns := range ignoredNamespaces {
 		ignoredNamespacesMap[ns] = struct{}{}

event-exporter/kubernetes/podlabels/pod_labels_informer_test.go

@@ -234,7 +234,7 @@ func TestGetLabelsCacheOperations(t *testing.T) {
 	}
 	fakeMetadataClient := metadatafake.NewSimpleMetadataClient(scheme, pod1, pod2)
 
-	factory := NewPodLabelsSharedInformerFactory(fakeMetadataClient, nil, false)
+	factory := NewPodLabelsSharedInformerFactory(fakeMetadataClient, nil)
 	collector := factory.NewPodLabelsSharedInformer()
 	stopCh := make(chan struct{})
 	defer close(stopCh)

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

@@ -18,8 +18,6 @@ package events
 
 import (
 	"context"
-	"errors"
-	"regexp"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -31,7 +29,6 @@ import (
 	"k8s.io/client-go/tools/cache"
 
 	"github.com/GoogleCloudPlatform/k8s-stackdriver/event-exporter/kubernetes/watchers"
-	"github.com/golang/glog"
 )
 
 const (
@@ -51,47 +48,41 @@ const (
 	eventWatchListPageSize = 10000
 )
 
-// OnListFunc represent an action on the initial list of object received
-// from the Kubernetes API server before starting watching for the updates.
+// OnListFunc represent an action on the initial sync with the Kubernetes
+// API server, before starting watching for the updates. The list passed
+// to it carries only the resource version of the sync, not the items.
 type OnListFunc func(*corev1.EventList)
 
 // EventWatcherConfig represents the configuration for the watcher that
 // only watches the events resource.
 type EventWatcherConfig struct {
-	// Note, that this action will be executed on each List request, of which
+	// Note, that this action will be executed on each initial sync, of which
 	// there can be many, e.g. because of network problems. Note also, that
-	// items in the List response WILL NOT trigger OnAdd method in handler,
-	// instead Store contents will be completely replaced.
-	OnList                       OnListFunc
-	ResyncPeriod                 time.Duration
-	Handler                      EventHandler
-	EventLabelSelector           labels.Selector
-	ListerWatcherOptionsLimit    int64
-	ListerWatcherEnableStreaming bool
-	StorageType                  watchers.StorageType
+	// objects received during the sync WILL NOT trigger OnAdd method in
+	// handler, instead Store contents will be completely replaced.
+	OnList                    OnListFunc
+	ResyncPeriod              time.Duration
+	Handler                   EventHandler
+	EventLabelSelector        labels.Selector
+	ListerWatcherOptionsLimit int64
+	StorageType               watchers.StorageType
 }
 
 // NewEventWatcher create a new watcher that only watches the events resource.
+// When the WatchListClient client-go feature gate is enabled, the underlying
+// Reflector establishes the initial state via a watch-list stream
+// (https://kep.k8s.io/3157) and falls back to the paginated list below if the
+// API server does not support it.
 func NewEventWatcher(client kubernetes.Interface, config *EventWatcherConfig) watchers.Watcher {
-	watchListFeatureGateEnabled := IsFeatureGateEnabled(client, "WatchList")
-	glog.Infof("Feature gate WatchList is enabled: %v, config.ListerWatcherEnableStreaming: %v", watchListFeatureGateEnabled, config.ListerWatcherEnableStreaming)
 	return watchers.NewWatcher(&watchers.WatcherConfig{
 		// List and watch events in all namespaces.
 		ListerWatcher: &cache.ListWatch{
 			ListFunc: func(options meta_v1.ListOptions) (runtime.Object, error) {
-				if config.ListerWatcherEnableStreaming && watchListFeatureGateEnabled {
-					return streamingListEvents(client, config, options)
-				} else {
-					if config.ListerWatcherOptionsLimit > 0 {
-						options.Limit = config.ListerWatcherOptionsLimit
-					}
-					options.LabelSelector = config.EventLabelSelector.String()
-					list, err := client.CoreV1().Events(meta_v1.NamespaceAll).List(context.TODO(), options)
-					if err == nil {
-						config.OnList(list)
-					}
-					return list, err
+				if config.ListerWatcherOptionsLimit > 0 {
+					options.Limit = config.ListerWatcherOptionsLimit
 				}
+				options.LabelSelector = config.EventLabelSelector.String()
+				return client.CoreV1().Events(meta_v1.NamespaceAll).List(context.TODO(), options)
 			},
 			WatchFunc: func(options meta_v1.ListOptions) (watch.Interface, error) {
 				options.LabelSelector = config.EventLabelSelector.String()
@@ -100,107 +91,20 @@ func NewEventWatcher(client kubernetes.Interface, config *EventWatcherConfig) wa
 		},
 		ExpectedType: &corev1.Event{},
 		StoreConfig: &watchers.WatcherStoreConfig{
-			KeyFunc:     cache.DeletionHandlingMetaNamespaceKeyFunc,
-			Handler:     newEventHandlerWrapper(config.Handler),
+			KeyFunc: cache.DeletionHandlingMetaNamespaceKeyFunc,
+			Handler: newEventHandlerWrapper(config.Handler),
+			// Replace is called once per initial sync, both by the list path
+			// and at the watch-list bookmark. The items are deliberately not
+			// copied into the EventList to keep this O(1) in cluster size.
+			OnReplace: func(items []interface{}, resourceVersion string) {
+				config.OnList(&corev1.EventList{
+					ListMeta: meta_v1.ListMeta{ResourceVersion: resourceVersion},
+				})
+			},
 			StorageType: config.StorageType,
 			StorageTTL:  eventStorageTTL,
 		},
 		ResyncPeriod:      config.ResyncPeriod,
 		WatchListPageSize: eventWatchListPageSize,
 	})
 }
-
-// streamingListEvents uses Streaming List (SendInitialEvents=true) to avoid buffering.
-// This allows us to process initial events incrementally.
-func streamingListEvents(client kubernetes.Interface, config *EventWatcherConfig, options meta_v1.ListOptions) (runtime.Object, error) {
-	sendInitialEvents := true
-	options.SendInitialEvents = &sendInitialEvents
-	options.ResourceVersionMatch = meta_v1.ResourceVersionMatchNotOlderThan
-	options.Watch = true
-	options.LabelSelector = config.EventLabelSelector.String()
-	options.AllowWatchBookmarks = true
-	glog.Infof("streamingListEvents started watching events with options: %v", options)
-
-	// Perform the streaming list (actually a Watch)
-	watcher, err := client.CoreV1().Events(meta_v1.NamespaceAll).Watch(context.TODO(), options)
-	if err != nil {
-		glog.Errorf("streamingListEvents failed to watch events: %v", err)
-		return nil, err
-	}
-	defer watcher.Stop()
-	glog.Infof("streamingListEvents started watching events")
-
-	// Call OnList once to start the sink (it just logs "Started watching")
-	config.OnList(&corev1.EventList{})
-
-	lastRV := ""
-	bookmarkReceived := false
-
-eventLoop:
-	for event := range watcher.ResultChan() {
-		if meta, ok := event.Object.(meta_v1.Object); ok {
-			lastRV = meta.GetResourceVersion()
-		}
-
-		switch event.Type {
-		case watch.Added:
-			if e, ok := event.Object.(*corev1.Event); ok {
-				// Manually pass to handler since we bypass Reflector's store
-				config.Handler.OnAdd(e)
-			}
-		case watch.Bookmark:
-			// Check for the annotation that signals the initial list is done.
-			if m, ok := event.Object.(meta_v1.Object); ok {
-				if val, ok := m.GetAnnotations()["k8s.io/initial-events-end"]; ok && val == "true" {
-					// Close the channel and break the loop
-					bookmarkReceived = true
-					break eventLoop
-				}
-			}
-		case watch.Error:
-			// If we get an error, Reflector will retry ListFunc anyway.
-			// We can return the error here to trigger that retry.
-			if status, ok := event.Object.(*meta_v1.Status); ok {
-				return nil, errors.New(status.Message)
-			}
-		}
-	}
-
-	if !bookmarkReceived {
-		// If we exited the loop without receiving the bookmark, something went wrong.
-		return nil, errors.New("streaming list ended without receiving initial-events-end bookmark")
-	}
-
-	// Return an empty list with the correct ResourceVersion.
-	// Reflector will then start Watching from this version.
-	return &corev1.EventList{
-		ListMeta: meta_v1.ListMeta{
-			ResourceVersion: lastRV,
-		},
-		Items: []corev1.Event{},
-	}, nil
-}
-
-func IsFeatureGateEnabled(client kubernetes.Interface, featureName string) bool {
-	// Request raw metrics from the API server
-	data, err := client.CoreV1().RESTClient().Get().
-		AbsPath("/metrics").
-		SetHeader("Accept", "text/plain").
-		DoRaw(context.TODO())
-
-	if err != nil {
-		glog.Errorf("fail to get raw metrics: %v", err)
-		return false
-	}
-
-	// Pattern explained:
-	// 1. Match the metric name and the feature name label
-	// 2. [^}]* matches any other labels (like stage="BETA")
-	// 3. \s+ matches the whitespace before the value
-	// 4. (1(\.0+)?) matches "1" or "1.0", "1.00", etc.
-	// 5. (\s+|$) ensures it's the end of the value (whitespace or end of line)
-	pattern := `kubernetes_feature_enabled\{name="` + featureName + `"[^}]*\}\s+(1(\.0+)?)(\s+|$)`
-	re := regexp.MustCompile(pattern)
-
-	return re.Match(data)
-}

event-exporter/kubernetes/watchers/storage.go

@@ -39,16 +39,21 @@ const (
 
 // WatcherStoreConfig represents the configuration of the storage backing the watcher.
 type WatcherStoreConfig struct {
-	KeyFunc     cache.KeyFunc
-	Handler     cache.ResourceEventHandler
+	KeyFunc cache.KeyFunc
+	Handler cache.ResourceEventHandler
+	// OnReplace is called after the Reflector has successfully established
+	// the initial state, either via a list request or via a watch-list
+	// stream. Objects passed here do not go through the Handler.
+	OnReplace   func(items []interface{}, resourceVersion string)
 	StorageType StorageType
 	StorageTTL  time.Duration
 }
 
 type watcherStore struct {
 	cache.ReflectorStore
 
-	handler cache.ResourceEventHandler
+	handler   cache.ResourceEventHandler
+	onReplace func(items []interface{}, resourceVersion string)
 }
 
 func (s *watcherStore) Add(obj interface{}) error {
@@ -66,6 +71,16 @@ func (s *watcherStore) Delete(obj interface{}) error {
 	return nil
 }
 
+func (s *watcherStore) Replace(items []interface{}, resourceVersion string) error {
+	if err := s.ReflectorStore.Replace(items, resourceVersion); err != nil {
+		return err
+	}
+	if s.onReplace != nil {
+		s.onReplace(items, resourceVersion)
+	}
+	return nil
+}
+
 func newWatcherStore(config *WatcherStoreConfig) *watcherStore {
 	var cacheStorage cache.ReflectorStore
 	switch config.StorageType {
@@ -81,5 +96,6 @@ func newWatcherStore(config *WatcherStoreConfig) *watcherStore {
 	return &watcherStore{
 		ReflectorStore: cacheStorage,
 		handler:        config.Handler,
+		onReplace:      config.OnReplace,
 	}
 }

event-exporter/main.go

@@ -33,6 +33,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
 	"k8s.io/apimachinery/pkg/labels"
+	clientfeatures "k8s.io/client-go/features"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/metadata"
 	"k8s.io/client-go/rest"
@@ -49,10 +50,25 @@ var (
 	enablePodOwnerLabel          = flag.Bool("enable-pod-owner-label", true, "Whether to enable the pod label collector to add pod owner labels to log entries")
 	eventLabelSelector           = flag.String("event-label-selector", "", "Export events only if they match the given label selector. Same syntax as kubectl label")
 	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.")
+	listerWatcherEnableStreaming = flag.Bool("lister-watcher-enable-streaming", false, "Establish the initial state via the WatchList streaming API (KEP-3157) instead of paginated list requests. Falls back to regular list requests if the API server does not support streaming; in that case lister-watcher-options-limit still applies.")
 	storageType                  = flag.String("storage-type", "DeltaFIFOStorage", "What storage should be used as a cache for the watcher. Supported sotrage type: SimpleStorage, TTLStorage and DeltaFIFOStorage.")
 )
 
+// watchListClientFeatureGates overrides the client-go WatchListClient feature
+// gate, which makes Reflectors establish their initial state via a watch-list
+// stream (KEP-3157) instead of paginated list requests.
+type watchListClientFeatureGates struct {
+	clientfeatures.Gates
+	enableWatchListClient bool
+}
+
+func (g *watchListClientFeatureGates) Enabled(key clientfeatures.Feature) bool {
+	if key == clientfeatures.WatchListClient {
+		return g.enableWatchListClient
+	}
+	return g.Gates.Enabled(key)
+}
+
 func newSystemStopChannel() chan struct{} {
 	ch := make(chan struct{})
 	go func() {
@@ -96,6 +112,14 @@ func main() {
 	defer glog.Flush()
 	flag.Parse()
 
+	// The gate is global and read by every Reflector at construction, both
+	// the events watcher and the pod-labels informer.
+	clientfeatures.ReplaceFeatureGates(&watchListClientFeatureGates{
+		Gates:                 clientfeatures.FeatureGates(),
+		enableWatchListClient: *listerWatcherEnableStreaming,
+	})
+	glog.Infof("Client-go feature gate WatchListClient enabled: %v", *listerWatcherEnableStreaming)
+
 	client, err := newKubernetesClient()
 	if err != nil {
 		glog.Fatalf("Failed to initialize kubernetes client: %v", err)
@@ -109,7 +133,7 @@ func main() {
 	var informer podlabels.PodLabelCollector = nil
 	stopCh := newSystemStopChannel()
 	if *enablePodOwnerLabel {
-		factory := podlabels.NewPodLabelsSharedInformerFactory(metadataClient, strings.Split(*systemNamespaces, ","), *listerWatcherEnableStreaming)
+		factory := podlabels.NewPodLabelsSharedInformerFactory(metadataClient, strings.Split(*systemNamespaces, ","))
 		informer = factory.NewPodLabelsSharedInformer()
 		factory.Run(stopCh)
 	}
@@ -136,7 +160,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, st)
 
 	// Expose the Prometheus http endpoint
 	go func() {