DO NOT MERGE: add instrumenting to chase very short watches

Author: deads2k

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

@@ -450,6 +450,8 @@ func NewCacherFromConfig(config Config) (*Cacher, error) {
 }
 
 func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
+	fmt.Printf("#### 2a groupResource=%v startCaching\n", c.groupResource)
+
 	// 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
@@ -465,6 +467,7 @@ func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
 	})
 	defer func() {
 		if successfulList {
+			fmt.Printf("#### 2b groupResource=%v setting to false\n", c.groupResource)
 			c.ready.set(false)
 		}
 	}()
@@ -477,6 +480,7 @@ func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
 	if err := c.reflector.ListAndWatch(stopChannel); err != nil {
 		klog.Errorf("cacher (%v): unexpected ListAndWatch error: %v; reinitializing...", c.groupResource.String(), err)
 	}
+	fmt.Printf("#### 2e groupResource=%v exiting\n", c.groupResource)
 }
 
 // Versioner implements storage.Interface.
@@ -536,17 +540,20 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 
 	var readyGeneration int
 	if utilfeature.DefaultFeatureGate.Enabled(features.ResilientWatchCacheInitialization) {
+		fmt.Printf("#### 1a groupResource=%v\n", c.groupResource)
 		var ok bool
 		readyGeneration, ok = c.ready.checkAndReadGeneration()
 		if !ok {
 			return nil, errors.NewTooManyRequests("storage is (re)initializing", 1)
 		}
 	} else {
+		fmt.Printf("#### 1b groupResource=%v\n", c.groupResource)
 		readyGeneration, err = c.ready.waitAndReadGeneration(ctx)
 		if err != nil {
 			return nil, errors.NewServiceUnavailable(err.Error())
 		}
 	}
+	fmt.Printf("#### 1c groupResource=%v, originalReadyGeneration=%d\n", c.groupResource, readyGeneration)
 
 	// determine the namespace and name scope of the watch, first from the request, secondarily from the field selector
 	scope := namespacedName{}
@@ -659,6 +666,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 		defer c.Unlock()
 
 		if generation, ok := c.ready.checkAndReadGeneration(); generation != readyGeneration || !ok {
+			fmt.Printf("#### 1o groupResource=%v currentReadyGeneration=%d, originalReadyGeneration=%d, ok=%v\n", c.groupResource, generation, readyGeneration, ok)
 			// 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.
@@ -680,6 +688,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 	}()
 
 	if !addedWatcher {
+		fmt.Printf("#### 1x groupResource=%v returning the immediate closer thing\n", c.groupResource)
 		// Watcher isn't really started at this point, so it's safe to just drop it.
 		//
 		// We're simulating the immediate watch termination, which boils down to simply

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

@@ -117,6 +117,7 @@ func (r *ready) checkAndReadGeneration() (int, bool) {
 func (r *ready) set(ok bool) {
 	r.lock.Lock()
 	defer r.lock.Unlock()
+
 	if r.state == Stopped {
 		return
 	}