Implement watchcache returning error from etcd that caused cache reinitialization

Author: serathius

staging/src/k8s.io/apiextensions-apiserver/test/integration/conversion/conversion_test.go

@@ -686,18 +686,14 @@ func expectConversionFailureMessage(id, message string) func(t *testing.T, ctc *
 		objv1beta2 := newConversionMultiVersionFixture(ns, id, "v1beta2")
 		meta, _, _ := unstructured.NestedFieldCopy(obj.Object, "metadata")
 		unstructured.SetNestedField(objv1beta2.Object, meta, "metadata")
-		lastRV := objv1beta2.GetResourceVersion()
 
 		for _, verb := range []string{"get", "list", "create", "update", "patch", "delete", "deletecollection"} {
 			t.Run(verb, func(t *testing.T) {
 				switch verb {
 				case "get":
 					_, err = clients["v1beta2"].Get(context.TODO(), obj.GetName(), metav1.GetOptions{})
 				case "list":
-					// With ResilientWatchcCacheInitialization feature, List requests are rejected with 429 if watchcache is not initialized.
-					// However, in some of these tests that install faulty converter webhook, watchcache will never initialize by definition (as list will never succeed due to faulty converter webook).
-					// In such case, the returned error will differ from the one returned from the etcd, so we need to force the request to go to etcd.
-					_, err = clients["v1beta2"].List(context.TODO(), metav1.ListOptions{ResourceVersion: lastRV, ResourceVersionMatch: metav1.ResourceVersionMatchExact})
+					_, err = clients["v1beta2"].List(context.TODO(), metav1.ListOptions{})
 				case "create":
 					_, err = clients["v1beta2"].Create(context.TODO(), newConversionMultiVersionFixture(ns, id, "v1beta2"), metav1.CreateOptions{})
 				case "update":

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go

@@ -463,31 +463,19 @@ func NewCacherFromConfig(config Config) (*Cacher, error) {
 }
 
 func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
-	// The 'usable' lock is always 'RLock'able when it is safe to use the cache.
-	// It is safe to use the cache after a successful list until a disconnection.
-	// We start with usable (write) locked. The below OnReplace function will
-	// unlock it after a successful list. The below defer will then re-lock
-	// it when this function exits (always due to disconnection), only if
-	// we actually got a successful list. This cycle will repeat as needed.
-	successfulList := false
 	c.watchCache.SetOnReplace(func() {
-		successfulList = true
-		c.ready.set(true)
+		c.ready.setReady()
 		klog.V(1).Infof("cacher (%v): initialized", c.groupResource.String())
 		metrics.WatchCacheInitializations.WithLabelValues(c.groupResource.String()).Inc()
 	})
+	var err error
 	defer func() {
-		if successfulList {
-			c.ready.set(false)
-		}
+		c.ready.setError(err)
 	}()
 
 	c.terminateAllWatchers()
-	// Note that since onReplace may be not called due to errors, we explicitly
-	// need to retry it on errors under lock.
-	// Also note that startCaching is called in a loop, so there's no need
-	// to have another loop here.
-	if err := c.reflector.ListAndWatch(stopChannel); err != nil {
+	err = c.reflector.ListAndWatch(stopChannel)
+	if err != nil {
 		klog.Errorf("cacher (%v): unexpected ListAndWatch error: %v; reinitializing...", c.groupResource.String(), err)
 	}
 }
@@ -506,11 +494,11 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 
 	var readyGeneration int
 	if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
-		var ok bool
+		var err error
 		var downtime time.Duration
-		readyGeneration, downtime, ok = c.ready.checkAndReadGeneration()
-		if !ok {
-			return nil, errors.NewTooManyRequests("storage is (re)initializing", calculateRetryAfterForUnreadyCache(downtime))
+		readyGeneration, downtime, err = c.ready.checkAndReadGeneration()
+		if err != nil {
+			return nil, errors.NewTooManyRequests(err.Error(), calculateRetryAfterForUnreadyCache(downtime))
 		}
 	} else {
 		readyGeneration, err = c.ready.waitAndReadGeneration(ctx)
@@ -631,7 +619,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 		c.Lock()
 		defer c.Unlock()
 
-		if generation, _, ok := c.ready.checkAndReadGeneration(); generation != readyGeneration || !ok {
+		if generation, _, err := c.ready.checkAndReadGeneration(); generation != readyGeneration || err != nil {
 			// We went unready or are already on a different generation.
 			// Avoid registering and starting the watch as it will have to be
 			// terminated immediately anyway.
@@ -749,10 +737,10 @@ func (c *Cacher) GetList(ctx context.Context, key string, opts storage.ListOptio
 	defer span.End(500 * time.Millisecond)
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
-		if downtime, ok := c.ready.check(); !ok {
+		if downtime, err := c.ready.check(); err != nil {
 			// If Cacher is not initialized, reject List requests
 			// as described in https://kep.k8s.io/4568
-			return errors.NewTooManyRequests("storage is (re)initializing", calculateRetryAfterForUnreadyCache(downtime))
+			return errors.NewTooManyRequests(err.Error(), calculateRetryAfterForUnreadyCache(downtime))
 		}
 	} else {
 		if err := c.ready.wait(ctx); err != nil {
@@ -1304,8 +1292,8 @@ func (c *Cacher) setInitialEventsEndBookmarkIfRequested(cacheInterval *watchCach
 }
 
 func (c *Cacher) Ready() bool {
-	_, ok := c.ready.check()
-	return ok
+	_, err := c.ready.check()
+	return err == nil
 }
 
 // errWatcher implements watch.Interface to return a single error

staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher_whitebox_test.go

@@ -3213,7 +3213,7 @@ func TestRetryAfterForUnreadyCache(t *testing.T) {
 		t.Fatalf("Unexpected error waiting for the cache to be ready")
 	}
 
-	cacher.ready.set(false)
+	cacher.ready.setError(nil)
 	clock.Step(14 * time.Second)
 
 	opts := storage.ListOptions{

staging/src/k8s.io/apiserver/pkg/storage/cacher/ready.go

@@ -41,7 +41,8 @@ const (
 //	|                           ^
 //	└---------------------------┘
 type ready struct {
-	state       status        // represent the state of the variable
+	state       status // represent the state of the variable
+	lastErr     error
 	generation  int           // represent the number of times we have transtioned to ready
 	lock        sync.RWMutex  // protect the state and generation variables
 	restartLock sync.Mutex    // protect the transition from ready to pending where the channel is recreated
@@ -87,49 +88,69 @@ func (r *ready) waitAndReadGeneration(ctx context.Context) (int, error) {
 		}
 
 		r.lock.RLock()
-		switch r.state {
-		case Pending:
+		if r.state == Pending {
 			// since we allow to switch between the states Pending and Ready
 			// if there is a quick transition from Pending -> Ready -> Pending
 			// a process that was waiting can get unblocked and see a Pending
 			// state again. If the state is Pending we have to wait again to
 			// avoid an inconsistent state on the system, with some processes not
 			// waiting despite the state moved back to Pending.
 			r.lock.RUnlock()
-		case Ready:
-			generation := r.generation
-			r.lock.RUnlock()
-			return generation, nil
-		case Stopped:
-			r.lock.RUnlock()
-			return 0, fmt.Errorf("apiserver cacher is stopped")
-		default:
-			r.lock.RUnlock()
-			return 0, fmt.Errorf("unexpected apiserver cache state: %v", r.state)
+			continue
 		}
+		generation, err := r.readGenerationLocked()
+		r.lock.RUnlock()
+		return generation, err
 	}
 }
 
 // check returns the time elapsed since the state was last changed and the current value.
-func (r *ready) check() (time.Duration, bool) {
-	_, elapsed, ok := r.checkAndReadGeneration()
-	return elapsed, ok
+func (r *ready) check() (time.Duration, error) {
+	_, elapsed, err := r.checkAndReadGeneration()
+	return elapsed, err
 }
 
 // checkAndReadGeneration returns the current generation, the time elapsed since the state was last changed and the current value.
-func (r *ready) checkAndReadGeneration() (int, time.Duration, bool) {
+func (r *ready) checkAndReadGeneration() (int, time.Duration, error) {
 	r.lock.RLock()
 	defer r.lock.RUnlock()
-	return r.generation, r.clock.Since(r.lastStateChangeTime), r.state == Ready
+	generation, err := r.readGenerationLocked()
+	return generation, r.clock.Since(r.lastStateChangeTime), err
+}
+
+func (r *ready) readGenerationLocked() (int, error) {
+	switch r.state {
+	case Pending:
+		if r.lastErr == nil {
+			return 0, fmt.Errorf("storage is (re)initializing")
+		} else {
+			return 0, fmt.Errorf("storage is (re)initializing: %w", r.lastErr)
+		}
+	case Ready:
+		return r.generation, nil
+	case Stopped:
+		return 0, fmt.Errorf("apiserver cacher is stopped")
+	default:
+		return 0, fmt.Errorf("unexpected apiserver cache state: %v", r.state)
+	}
+}
+
+func (r *ready) setReady() {
+	r.set(true, nil)
+}
+
+func (r *ready) setError(err error) {
+	r.set(false, err)
 }
 
 // set the state to Pending (false) or Ready (true), it does not have effect if the state is Stopped.
-func (r *ready) set(ok bool) {
+func (r *ready) set(ok bool, err error) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 	if r.state == Stopped {
 		return
 	}
+	r.lastErr = err
 	if ok && r.state == Pending {
 		r.state = Ready
 		r.generation++