K8s apiserver watch 机制浅析

type Cacher struct {
   // incoming 事件管道, 会被分发给所有的watchers
   incoming chan watchCacheEvent
   
   //storage 的底层实现
   storage storage.Interface
   // 对象类型
   objectType reflect.Type


   // watchCache 滑动窗口，维护了当前kind的所有的资源，和一个基于滑动窗口的最近的事件数组
   watchCache *watchCache
   
   // reflector list并watch etcd 并将事件和资源存到watchCache中
   reflector  *cache.Reflector
   
   // watchersBuffer 代表着所有client-go客户端跟apiserver的连接
   watchersBuffer []*cacheWatcher
   ....
}

func NewCacherFromConfig(config Config) (*Cacher, error) {
      ...
   cacher := &Cacher{
      ...
      incoming:              make(chan watchCacheEvent, 100),
      ...
   }
      ...
    watchCache := newWatchCache(
      config.KeyFunc, cacher.processEvent, config.GetAttrsFunc, config.Versioner, config.Indexers, config.Clock, objType)
   listerWatcher := NewCacherListerWatcher(config.Storage, config.ResourcePrefix, config.NewListFunc)
   reflectorName := "storage/cacher.go:" + config.ResourcePrefix


   reflector := cache.NewNamedReflector(reflectorName, listerWatcher, obj, watchCache, 0)
   // Configure reflector's pager to for an appropriate pagination chunk size for fetching data from
   // storage. The pager falls back to full list if paginated list calls fail due to an "Expired" error.
   reflector.WatchListPageSize = storageWatchListPageSize


   cacher.watchCache = watchCache
   cacher.reflector = reflector


   go cacher.dispatchEvents() // 1


   cacher.stopWg.Add(1)
   go func() {
      defer cacher.stopWg.Done()
      defer cacher.terminateAllWatchers()
      wait.Until(
         func() {
            if !cacher.isStopped() {
               cacher.startCaching(stopCh)  // 2
            }
         }, time.Second, stopCh,
      )
   }()


   return cacher, nil
}

func (c *Cacher) dispatchEvents() {
   ...
   for {
      select {
      case event, ok := <-c.incoming:
         if !ok {
            return
         }
         if event.Type != watch.Bookmark {
         // 从incoming通道中获取事件，并发送给交给dispatchEvent方法处理
            c.dispatchEvent(&event)
         }
         lastProcessedResourceVersion = event.ResourceVersion
         metrics.EventsCounter.WithLabelValues(c.objectType.String()).Inc()
      ...
      case <-c.stopCh:
         return
      }
   }
}

func (c *Cacher) dispatchEvent(event *watchCacheEvent) {
   c.startDispatching(event)
   defer c.finishDispatching()
   if event.Type == watch.Bookmark {
      for _, watcher := range c.watchersBuffer {
         watcher.nonblockingAdd(event)
      }
   } else {
      wcEvent := *event
      setCachingObjects(&wcEvent, c.versioner)
      event = &wcEvent




      c.blockedWatchers = c.blockedWatchers[:0]
      // watchersBuffer 是一个数组，维护着所有client-go跟apiserver的watch连接，产生的cacheWatcher。
      for _, watcher := range c.watchersBuffer {
         if !watcher.nonblockingAdd(event) {
            c.blockedWatchers = append(c.blockedWatchers, watcher)
         }
      }
      ...
   }
}

func (c *cacheWatcher) nonblockingAdd(event *watchCacheEvent) bool {
   select {
   case c.input <- event:
      return true
   default:
      return false
   }
}

type cacheWatcher struct {
   input     chan *watchCacheEvent
   result    chan watch.Event
   done      chan struct{}
   filter    filterWithAttrsFunc
   stopped   bool
   forget    func()
   versioner storage.Versioner
   // The watcher will be closed by server after the deadline,
   // save it here to send bookmark events before that.
   deadline            time.Time
   allowWatchBookmarks bool
   // Object type of the cache watcher interests
   objectType reflect.Type




   // human readable identifier that helps assigning cacheWatcher
   // instance with request
   identifier string
}

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.objectType.String(), err)
   }
}

func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) error {
   found := make([]interface{}, 0, len(items))
   for _, item := range items {
      found = append(found, item)
   }
   return r.store.Replace(found, resourceVersion)
}

func watchHandler(start time.Time,
   w watch.Interface,
   store Store,
   expectedType reflect.Type,
   expectedGVK *schema.GroupVersionKind,
   name string,
   expectedTypeName string,
   setLastSyncResourceVersion func(string),
   clock clock.Clock,
   errc chan error,
   stopCh <-chan struct{},
) error {
   eventCount := 0




   // Stopping the watcher should be idempotent and if we return from this function there's no way
   // we're coming back in with the same watch interface.
   defer w.Stop()




loop:
   for {
      select {
      case <-stopCh:
         return errorStopRequested
      case err := <-errc:
         return err
      case event, ok := <-w.ResultChan():
         if !ok {
            break loop
         }
         if event.Type == watch.Error {
            return apierrors.FromObject(event.Object)
         }
         if expectedType != nil {
            if e, a := expectedType, reflect.TypeOf(event.Object); e != a {
               utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", name, e, a))
               continue
            }
         }
         if expectedGVK != nil {
            if e, a := *expectedGVK, event.Object.GetObjectKind().GroupVersionKind(); e != a {
               utilruntime.HandleError(fmt.Errorf("%s: expected gvk %v, but watch event object had gvk %v", name, e, a))
               continue
            }
         }
         meta, err := meta.Accessor(event.Object)
         if err != nil {
            utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", name, event))
            continue
         }
         resourceVersion := meta.GetResourceVersion()
         switch event.Type {
         case watch.Added:
            err := store.Add(event.Object)
            if err != nil {
               utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", name, event.Object, err))
            }
         ...  
         setLastSyncResourceVersion(resourceVersion)
         if rvu, ok := store.(ResourceVersionUpdater); ok {
            rvu.UpdateResourceVersion(resourceVersion)
         }
         eventCount++
      }
   }
}

func (w *watchCache) Replace(objs []interface{}, resourceVersion string) error {
   ... 
   if err := w.store.Replace(toReplace, resourceVersion); err != nil {
      return err
   }
   ...
}

func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, updateFunc func(*storeElement) error) error {
   ...
   if err := func() error {
      // TODO: We should consider moving this lock below after the watchCacheEvent
      // is created. In such situation, the only problematic scenario is Replace(
      // happening after getting object from store and before acquiring a lock.
      // Maybe introduce another lock for this purpose.
      w.Lock()
      defer w.Unlock()
      w.updateCache(wcEvent)
      w.resourceVersion = resourceVersion
      defer w.cond.Broadcast()
    ...
      return updateFunc(elem)
   }(); err != nil {
      return err
   }
   if w.eventHandler != nil {
     w.eventHandler(wcEvent)
   }
}

func (w *watchCache) updateCache(event *watchCacheEvent) {
   w.resizeCacheLocked(event.RecordTime)
   if w.isCacheFullLocked() {
      // Cache is full - remove the oldest element.
      w.startIndex++
   }
   w.cache[w.endIndex%w.capacity] = event
   w.endIndex++
}

func (c *Cacher) processEvent(event *watchCacheEvent) {
   if curLen := int64(len(c.incoming)); c.incomingHWM.Update(curLen) {
      // Monitor if this gets backed up, and how much.
      klog.V(1).Infof("cacher (%v): %v objects queued in incoming channel.", c.objectType.String(), curLen)
   }
   c.incoming <- *event
}

func ListResource(r rest.Lister, rw rest.Watcher, scope *RequestScope, forceWatch bool, minRequestTimeout time.Duration) http.HandlerFunc {
   return func(w http.ResponseWriter, req *http.Request) {
      ...
      if opts.Watch || forceWatch {
         if rw == nil {
         ...
         watcher, err := rw.Watch(ctx, &opts)
         ...
         metrics.RecordLongRunning(req, requestInfo, metrics.APIServerComponent, func() {
            serveWatch(watcher, scope, outputMediaType, req, w, timeout)
         })
         return
      }
      ...
   }
}

func serveWatch(watcher watch.Interface, scope *RequestScope, mediaTypeOptions negotiation.MediaTypeOptions, req *http.Request, w http.ResponseWriter, timeout time.Duration) {
   ...
   server := &WatchServer{
      Watching: watcher,
      Scope:    scope,




      UseTextFraming:  useTextFraming,
      MediaType:       mediaType,
      Framer:          framer,
      Encoder:         encoder,
      EmbeddedEncoder: embeddedEncoder,




      Fixup: func(obj runtime.Object) runtime.Object {
         result, err := transformObject(ctx, obj, options, mediaTypeOptions, scope, req)
         if err != nil {
            utilruntime.HandleError(fmt.Errorf("failed to transform object %v: %v", reflect.TypeOf(obj), err))
            return obj
         }
         // When we are transformed to a table, use the table options as the state for whether we
         // should print headers - on watch, we only want to print table headers on the first object
         // and omit them on subsequent events.
         if tableOptions, ok := options.(*metav1.TableOptions); ok {
            tableOptions.NoHeaders = true
         }
         return result
      },




      TimeoutFactory: &realTimeoutFactory{timeout},
   }




   server.ServeHTTP(w, req)
}

func (s *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
   kind := s.Scope.Kind




   if wsstream.IsWebSocketRequest(req) {
      w.Header().Set("Content-Type", s.MediaType)
      websocket.Handler(s.HandleWS).ServeHTTP(w, req)
      return
   }




   flusher, ok := w.(http.Flusher)
   if !ok {
      err := fmt.Errorf("unable to start watch - can't get http.Flusher: %#v", w)
      utilruntime.HandleError(err)
      s.Scope.err(errors.NewInternalError(err), w, req)
      return
   }




   framer := s.Framer.NewFrameWriter(w)
   if framer == nil {
      // programmer error
      err := fmt.Errorf("no stream framing support is available for media type %q", s.MediaType)
      utilruntime.HandleError(err)
      s.Scope.err(errors.NewBadRequest(err.Error()), w, req)
      return
   }




   var e streaming.Encoder
   var memoryAllocator runtime.MemoryAllocator




   if encoder, supportsAllocator := s.Encoder.(runtime.EncoderWithAllocator); supportsAllocator {
      memoryAllocator = runtime.AllocatorPool.Get().(*runtime.Allocator)
      defer runtime.AllocatorPool.Put(memoryAllocator)
      e = streaming.NewEncoderWithAllocator(framer, encoder, memoryAllocator)
   } else {
      e = streaming.NewEncoder(framer, s.Encoder)
   }




   // ensure the connection times out
   timeoutCh, cleanup := s.TimeoutFactory.TimeoutCh()
   defer cleanup()




   // begin the stream
   w.Header().Set("Content-Type", s.MediaType)
   w.Header().Set("Transfer-Encoding", "chunked")
   w.WriteHeader(http.StatusOK)
   flusher.Flush()




   var unknown runtime.Unknown
   internalEvent := &metav1.InternalEvent{}
   outEvent := &metav1.WatchEvent{}
   buf := &bytes.Buffer{}
   ch := s.Watching.ResultChan()
   done := req.Context().Done()




   embeddedEncodeFn := s.EmbeddedEncoder.Encode
   if encoder, supportsAllocator := s.EmbeddedEncoder.(runtime.EncoderWithAllocator); supportsAllocator {
      if memoryAllocator == nil {
         // don't put the allocator inside the embeddedEncodeFn as that would allocate memory on every call.
         // instead, we allocate the buffer for the entire watch session and release it when we close the connection.
         memoryAllocator = runtime.AllocatorPool.Get().(*runtime.Allocator)
         defer runtime.AllocatorPool.Put(memoryAllocator)
      }
      embeddedEncodeFn = func(obj runtime.Object, w io.Writer) error {
         return encoder.EncodeWithAllocator(obj, w, memoryAllocator)
      }
   }




   for {
      select {
      case <-done:
         return
      case <-timeoutCh:
         return
      case event, ok := <-ch:
         if !ok {
            // End of results.
            return
         }
         metrics.WatchEvents.WithContext(req.Context()).WithLabelValues(kind.Group, kind.Version, kind.Kind).Inc()




         obj := s.Fixup(event.Object)
         if err := embeddedEncodeFn(obj, buf); err != nil {
            // unexpected error
            utilruntime.HandleError(fmt.Errorf("unable to encode watch object %T: %v", obj, err))
            return
         }




         // ContentType is not required here because we are defaulting to the serializer
         // type
         unknown.Raw = buf.Bytes()
         event.Object = &unknown
         metrics.WatchEventsSizes.WithContext(req.Context()).WithLabelValues(kind.Group, kind.Version, kind.Kind).Observe(float64(len(unknown.Raw)))




         *outEvent = metav1.WatchEvent{}




         // create the external type directly and encode it.  Clients will only recognize the serialization we provide.
         // The internal event is being reused, not reallocated so its just a few extra assignments to do it this way
         // and we get the benefit of using conversion functions which already have to stay in sync
         *internalEvent = metav1.InternalEvent(event)
         err := metav1.Convert_v1_InternalEvent_To_v1_WatchEvent(internalEvent, outEvent, nil)
         if err != nil {
            utilruntime.HandleError(fmt.Errorf("unable to convert watch object: %v", err))
            // client disconnect.
            return
         }
         if err := e.Encode(outEvent); err != nil {
            utilruntime.HandleError(fmt.Errorf("unable to encode watch object %T: %v (%#v)", outEvent, err, e))
            // client disconnect.
            return
         }
         if len(ch) == 0 {
            flusher.Flush()
         }




         buf.Reset()
      }
   }
}

func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) {
   ...
   watcher := newCacheWatcher(chanSize, filterWithAttrsFunction(key, pred), emptyFunc, c.versioner, deadline, pred.AllowWatchBookmarks, c.objectType, identifier)
   ...
   cacheInterval, err := c.watchCache.getAllEventsSinceLocked(watchRV)
   if err != nil {
      // To match the uncached watch implementation, once we have passed authn/authz/admission,
      // and successfully parsed a resource version, other errors must fail with a watch event of type ERROR,
      // rather than a directly returned error.
      return newErrWatcher(err), nil
   }




   func() {
      c.Lock()
      defer c.Unlock()
      // Update watcher.forget function once we can compute it.
      watcher.forget = forgetWatcher(c, c.watcherIdx, triggerValue, triggerSupported)
      c.watchers.addWatcher(watcher, c.watcherIdx, triggerValue, triggerSupported)




      // Add it to the queue only when the client support watch bookmarks.
      if watcher.allowWatchBookmarks {
         c.bookmarkWatchers.addWatcher(watcher)
      }
      c.watcherIdx++
   }()




   go watcher.processInterval(ctx, cacheInterval, watchRV)
   return watcher, nil
}

func (c *cacheWatcher) processInterval(ctx context.Context, cacheInterval *watchCacheInterval, resourceVersion uint64) {
   ...
   initEventCount := 0
   /* 首次watch 获取cacheInterval 的事件并发送到resultChan*/
   for {
      event, err := cacheInterval.Next()
      if err != nil {
         klog.Warningf("couldn't retrieve watch event to serve: %#v", err)
         return
      }
      if event == nil {
         break
      }
      c.sendWatchCacheEvent(event)
      resourceVersion = event.ResourceVersion
      initEventCount++
   }
   /* 后续建立watch连接之后，将input通道中的事件发送到resultChan*/
   c.process(ctx, resourceVersion)
}