Introduce a ConsistentWatchFromEtcd feature gate, enabled by default,…

… that can be used to disable direct watches to etcd
kubernetes · Mar 14, 2024 · c586bca · c586bca
1 parent d1a2a13
commit c586bca
Show file tree

Hide file tree

Showing 3 changed files with 28 additions and 1 deletion.
diff --git a/staging/src/k8s.io/apiserver/pkg/features/kube_features.go b/staging/src/k8s.io/apiserver/pkg/features/kube_features.go
@@ -108,6 +108,12 @@ const (
 	// Enables expression validation in Admission Control
 	ValidatingAdmissionPolicy featuregate.Feature = "ValidatingAdmissionPolicy"
 
+	// owner: @serathius
+	// beta: 1.30
+	// Enables consistent watches to be served from storage directly.
+	// Used to prevent https://github.com/kubernetes/kubernetes/issues/123072 until etcd fixes the issue.
+	ConsistentWatchFromStorage featuregate.Feature = "ConsistentWatchFromStorage"
+
 	// owner: @cici37
 	// kep: https://kep.k8s.io/2876
 	// alpha: v1.23
@@ -315,6 +321,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 
 	ValidatingAdmissionPolicy: {Default: true, PreRelease: featuregate.GA, LockToDefault: true}, // remove in 1.32
 
+	ConsistentWatchFromStorage: {Default: true, PreRelease: featuregate.Beta},
+
 	CustomResourceValidationExpressions: {Default: true, PreRelease: featuregate.GA, LockToDefault: true}, // remove in 1.31
 
 	EfficientWatchResumption: {Default: true, PreRelease: featuregate.GA, LockToDefault: true},

diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go
@@ -523,7 +523,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 		opts.SendInitialEvents = nil
 	}
 	// TODO: we should eventually get rid of this legacy case
-	if opts.SendInitialEvents == nil && opts.ResourceVersion == "" {
+	if utilfeature.DefaultFeatureGate.Enabled(features.ConsistentWatchFromStorage) && opts.SendInitialEvents == nil && opts.ResourceVersion == "" {
 		return c.storage.Watch(ctx, key, opts)
 	}
 	requestedWatchRV, err := c.versioner.ParseResourceVersion(opts.ResourceVersion)

diff --git a/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go b/staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go
@@ -355,6 +355,25 @@ func TestWatchCacheBypass(t *testing.T) {
 	if err != errDummy {
 		t.Errorf("Watch with unset RV should bypass cacher: %v", err)
 	}
+
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ConsistentWatchFromStorage, true)()
+	// With consistent watch from etcd enabled, check if cacher is bypassed.
+	_, err = cacher.Watch(context.TODO(), "pod/ns", storage.ListOptions{
+		ResourceVersion: "",
+	})
+	if err != errDummy {
+		t.Errorf("With ConsistentWatchFromStorage enabled, Watch with unset RV should bypass cacher: %v", err)
+	}
+
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ConsistentWatchFromStorage, false)()
+	// With consistent watch from etcd disabled, check if cacher is not bypassed.
+	_, err = cacher.Watch(context.TODO(), "pod/ns", storage.ListOptions{
+		ResourceVersion: "",
+		Predicate:       storage.Everything,
+	})
+	if err != nil {
+		t.Errorf("With ConsistentWatchFromStorage disabled, watch with unset RV should be served from cache: %v", err)
+	}
 }
 
 func TestEmptyWatchEventCache(t *testing.T) {