Ensure all WatchFunc enable watch and boomarks

AllowWatchBookmarks is generally pretty safe to enable as it has been
available in Kuberentes for a long while, and the server ignores the
flag if it doesn't implement it (per docs).

Signed-off-by: Vince Prignano <[email protected]>

Author: vincepri

pkg/cache/cache.go

@@ -223,6 +223,16 @@ type Options struct {
 	// DefaultNamespaces.
 	DefaultUnsafeDisableDeepCopy *bool
 
+	// DefaultEnableWatchBookmarks requests watch events with type "BOOKMARK".
+	// Servers that do not implement bookmarks may ignore this flag and
+	// bookmarks are sent at the server's discretion. Clients should not
+	// assume bookmarks are returned at any specific interval, nor may they
+	// assume the server will send any BOOKMARK event during a session.
+	//
+	// This will be used for all object types, unless it is set in ByObject or
+	// DefaultNamespaces.
+	DefaultEnableWatchBookmarks *bool
+
 	// ByObject restricts the cache's ListWatch to the desired fields per GVK at the specified object.
 	// If unset, this will fall through to the Default* settings.
 	ByObject map[client.Object]ByObject
@@ -273,6 +283,13 @@ type ByObject struct {
 	// Be very careful with this, when enabled you must DeepCopy any object before mutating it,
 	// otherwise you will mutate the object in the cache.
 	UnsafeDisableDeepCopy *bool
+
+	// EnableWatchBookmarks requests watch events with type "BOOKMARK".
+	// Servers that do not implement bookmarks may ignore this flag and
+	// bookmarks are sent at the server's discretion. Clients should not
+	// assume bookmarks are returned at any specific interval, nor may they
+	// assume the server will send any BOOKMARK event during a session.
+	EnableWatchBookmarks *bool
 }
 
 // Config describes all potential options for a given watch.
@@ -299,6 +316,13 @@ type Config struct {
 	// UnsafeDisableDeepCopy specifies if List and Get requests against the
 	// cache should not DeepCopy. A nil value allows to default this.
 	UnsafeDisableDeepCopy *bool
+
+	// EnableWatchBookmarks requests watch events with type "BOOKMARK".
+	// Servers that do not implement bookmarks may ignore this flag and
+	// bookmarks are sent at the server's discretion. Clients should not
+	// assume bookmarks are returned at any specific interval, nor may they
+	// assume the server will send any BOOKMARK event during a session.
+	EnableWatchBookmarks *bool
 }
 
 // NewCacheFunc - Function for creating a new cache from the options and a rest config.
@@ -368,6 +392,7 @@ func optionDefaultsToConfig(opts *Options) Config {
 		FieldSelector:         opts.DefaultFieldSelector,
 		Transform:             opts.DefaultTransform,
 		UnsafeDisableDeepCopy: opts.DefaultUnsafeDisableDeepCopy,
+		EnableWatchBookmarks:  opts.DefaultEnableWatchBookmarks,
 	}
 }
 
@@ -377,6 +402,7 @@ func byObjectToConfig(byObject ByObject) Config {
 		FieldSelector:         byObject.Field,
 		Transform:             byObject.Transform,
 		UnsafeDisableDeepCopy: byObject.UnsafeDisableDeepCopy,
+		EnableWatchBookmarks:  byObject.EnableWatchBookmarks,
 	}
 }
 
@@ -399,6 +425,7 @@ func newCache(restConfig *rest.Config, opts Options) newCacheFunc {
 				Transform:             config.Transform,
 				WatchErrorHandler:     opts.DefaultWatchErrorHandler,
 				UnsafeDisableDeepCopy: ptr.Deref(config.UnsafeDisableDeepCopy, false),
+				EnableWatchBookmarks:  ptr.Deref(config.EnableWatchBookmarks, true),
 				NewInformer:           opts.newInformer,
 			}),
 			readerFailOnMissingInformer: opts.ReaderFailOnMissingInformer,
@@ -508,6 +535,11 @@ func defaultOpts(config *rest.Config, opts Options) (Options, error) {
 	if opts.SyncPeriod == nil {
 		opts.SyncPeriod = &defaultSyncPeriod
 	}
+
+	// Default enable watch bookmarks, unless set.
+	if opts.DefaultEnableWatchBookmarks == nil {
+		opts.DefaultEnableWatchBookmarks = ptr.To(true)
+	}
 	return opts, nil
 }
 

pkg/cache/internal/informers.go

@@ -51,6 +51,7 @@ type InformersOpts struct {
 	Selector              Selector
 	Transform             cache.TransformFunc
 	UnsafeDisableDeepCopy bool
+	EnableWatchBookmarks  bool
 	WatchErrorHandler     cache.WatchErrorHandler
 }
 
@@ -78,6 +79,7 @@ func NewInformers(config *rest.Config, options *InformersOpts) *Informers {
 		selector:              options.Selector,
 		transform:             options.Transform,
 		unsafeDisableDeepCopy: options.UnsafeDisableDeepCopy,
+		enableWatchBookmarks:  options.EnableWatchBookmarks,
 		newInformer:           newInformer,
 		watchErrorHandler:     options.WatchErrorHandler,
 	}
@@ -174,6 +176,7 @@ type Informers struct {
 	selector              Selector
 	transform             cache.TransformFunc
 	unsafeDisableDeepCopy bool
+	enableWatchBookmarks  bool
 
 	// NewInformer allows overriding of the shared index informer constructor for testing.
 	newInformer func(cache.ListerWatcher, runtime.Object, time.Duration, cache.Indexers) cache.SharedIndexInformer
@@ -361,8 +364,10 @@ func (ip *Informers) addInformerToMap(gvk schema.GroupVersionKind, obj runtime.O
 			return listWatcher.ListFunc(opts)
 		},
 		WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
-			ip.selector.ApplyToList(&opts)
 			opts.Watch = true // Watch needs to be set to true separately
+			opts.AllowWatchBookmarks = ip.enableWatchBookmarks
+
+			ip.selector.ApplyToList(&opts)
 			return listWatcher.WatchFunc(opts)
 		},
 	}, obj, calculateResyncPeriod(ip.resync), cache.Indexers{
@@ -444,6 +449,9 @@ func (ip *Informers) makeListWatcher(gvk schema.GroupVersionKind, obj runtime.Ob
 			},
 			// Setup the watch function
 			WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
+				opts.Watch = true // Watch needs to be set to true separately
+				opts.AllowWatchBookmarks = ip.enableWatchBookmarks
+
 				if namespace != "" {
 					return resources.Namespace(namespace).Watch(ip.ctx, opts)
 				}
@@ -486,6 +494,9 @@ func (ip *Informers) makeListWatcher(gvk schema.GroupVersionKind, obj runtime.Ob
 			},
 			// Setup the watch function
 			WatchFunc: func(opts metav1.ListOptions) (watcher watch.Interface, err error) {
+				opts.Watch = true // Watch needs to be set to true separately
+				opts.AllowWatchBookmarks = ip.enableWatchBookmarks
+
 				if namespace != "" {
 					watcher, err = resources.Namespace(namespace).Watch(ip.ctx, opts)
 				} else {
@@ -527,6 +538,9 @@ func (ip *Informers) makeListWatcher(gvk schema.GroupVersionKind, obj runtime.Ob
 			},
 			// Setup the watch function
 			WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
+				opts.Watch = true // Watch needs to be set to true separately
+				opts.AllowWatchBookmarks = ip.enableWatchBookmarks
+
 				// Build the request.
 				req := client.Get().Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec)
 				if namespace != "" {