kubernetes 实现 list-watch 的底层原理

 1 //  startWatching does:
 2 // - get current objects if initialRev=0; set initialRev to current rev
 3 // - watch on given key and send events to process.
 4 func (wc *watchChan) startWatching(watchClosedCh chan struct{}) {
 5   if wc.initialRev == 0 {
 6     if err := wc.sync(); err != nil {
 7       klog.Errorf("failed to sync with latest state: %v", err)
 8       wc.sendError(err)
 9       return
10     }
11   }
12   opts := []clientv3.OpOption{clientv3.WithRev(wc.initialRev + 1), clientv3.WithPrevKV()}
13   if wc.recursive {
14     opts = append(opts, clientv3.WithPrefix())
15   }
16   if wc.progressNotify {
17     opts = append(opts, clientv3.WithProgressNotify())
18   }
19   wch := wc.watcher.client.Watch(wc.ctx, wc.key, opts...)
20   for wres := range wch {
21     if wres.Err() != nil {
22       err := wres.Err()
23       // If there is an error on server (e.g. compaction), the channel will return it before closed.
24       logWatchChannelErr(err)
25       wc.sendError(err)
26       return
27     }
28     if wres.IsProgressNotify() {
29       wc.sendEvent(progressNotifyEvent(wres.Header.GetRevision()))
30       metrics.RecordEtcdBookmark(wc.watcher.objectType)
31       continue
32     }
33 
34     for _, e := range wres.Events {
35       parsedEvent, err := parseEvent(e)
36       if err != nil {
37         logWatchChannelErr(err)
38         wc.sendError(err)
39         return
40       }
41       wc.sendEvent(parsedEvent)
42     }
43   }
44   // When we come to this point, it's only possible that client side ends the watch.
45   // e.g. cancel the context, close the client.
46   // If this watch chan is broken and context isn't cancelled, other goroutines will still hang.
47   // We should notify the main thread that this goroutine has exited.
48   close(watchClosedCh)
49 }

 1 func (wc *watchChan) sendEvent(e *event) {
 2   if len(wc.incomingEventChan) == incomingBufSize {
 3     klog.V(3).InfoS("Fast watcher, slow processing. Probably caused by slow decoding, user not receiving fast, or other processing logic", "incomingEvents", incomingBufSize, "objectType", wc.watcher.objectType)
 4   }
 5   select {
 6   case wc.incomingEventChan <- e:
 7   case <-wc.ctx.Done():
 8   }
 9 }
10 
11 // processEvent processes events from etcd watcher and sends results to resultChan.
12 func (wc *watchChan) processEvent(wg *sync.WaitGroup) {
13   defer wg.Done()
14 
15   for {
16     select {
17     case e := <-wc.incomingEventChan:
18       res := wc.transform(e)
19       if res == nil {
20         continue
21       }
22       if len(wc.resultChan) == outgoingBufSize {
23         klog.V(3).InfoS("Fast watcher, slow processing. Probably caused by slow dispatching events to watchers", "outgoingEvents", outgoingBufSize, "objectType", wc.watcher.objectType)
24       }
25       // If user couldn't receive results fast enough, we also block incoming events from watcher.
26       // Because storing events in local will cause more memory usage.
27       // The worst case would be closing the fast watcher.
28       select {
29       case wc.resultChan <- *res:
30       case <-wc.ctx.Done():
31         return
32       }
33     case <-wc.ctx.Done():
34       return
35     }
36   }
37 }

 1 // watchChan implements watch.Interface.
 2 type watchChan struct {
 3   // 省略其他属性
 4   watcher           *watcher
 5   incomingEventChan chan *event
 6   resultChan        chan watch.Event
 7   errChan           chan error
 8 }
 9 
10 func (wc *watchChan) ResultChan() <-chan watch.Event {
11   return wc.resultChan
12 }

watchChan.resultChan -> Cacher.incoming 
  -> cacheWatcher.input -> cacheWatcher.result

 1 // vendor/k8s.io/apiserver/pkg/storage/cacher/cacher.go:749
 2 func (c *Cacher) processEvent(event *watchCacheEvent) {
 3   if curLen := int64(len(c.incoming)); c.incomingHWM.Update(curLen) {
 4     // Monitor if this gets backed up, and how much.
 5     klog.V(1).Infof("cacher (%v): %v objects queued in incoming channel.", c.objectType.String(), curLen)
 6   }
 7   c.incoming <- *event
 8 }
 9 
10 // 中间经过 dispatchEvents
11 // vendor/k8s.io/apiserver/pkg/storage/cacher/cacher.go:757
12 
13 // vendor/k8s.io/apiserver/pkg/storage/cacher/cacher.go:1204
14 func (c *cacheWatcher) nonblockingAdd(event *watchCacheEvent) bool {
15   select {
16   case c.input <- event:
17     return true
18   default:
19     return false
20   }
21 }
22 
23 // vendor/k8s.io/apiserver/pkg/storage/cacher/cacher.go:1415
24 func (c *cacheWatcher) process(ctx context.Context, resourceVersion uint64) {
25   // At this point we already start processing incoming watch events.
26   // However, the init event can still be processed because their serialization
27   // and sending to the client happens asynchrnously.
28   // TODO: As describe in the KEP, we would like to estimate that by delaying
29   //   the initialization signal proportionally to the number of events to
30   //   process, but we're leaving this to the tuning phase.
31   utilflowcontrol.WatchInitialized(ctx)
32 
33   for {
34     select {
35     case event, ok := <-c.input:
36       if !ok {
37         return
38       }
39       // only send events newer than resourceVersion
40       if event.ResourceVersion > resourceVersion {
41         c.sendWatchCacheEvent(event)
42       }
43     case <-ctx.Done():
44       return
45     }
46   }
47 }
48 
49 // vendor/k8s.io/apiserver/pkg/storage/cacher/cacher.go:1319
50 // NOTE: sendWatchCacheEvent is assumed to not modify <event> !!!
51 func (c *cacheWatcher) sendWatchCacheEvent(event *watchCacheEvent) {
52   watchEvent := c.convertToWatchEvent(event)
53   if watchEvent == nil {
54     // Watcher is not interested in that object.
55     return
56   }
57 
58   // We need to ensure that if we put event X to the c.result, all
59   // previous events were already put into it before, no matter whether
60   // c.done is close or not.
61   // Thus we cannot simply select from c.done and c.result and this
62   // would give us non-determinism.
63   // At the same time, we don't want to block infinitely on putting
64   // to c.result, when c.done is already closed.
65   //
66   // This ensures that with c.done already close, we at most once go
67   // into the next select after this. With that, no matter which
68   // statement we choose there, we will deliver only consecutive
69   // events.
70   select {
71   case <-c.done:
72     return
73   default:
74   }
75 
76   select {
77   case c.result <- *watchEvent:
78   case <-c.done:
79   }
80 }
81 
82 // Implements watch.Interface.
83 func (c *cacheWatcher) ResultChan() <-chan watch.Event {
84   return c.result
85 }

  1 // vendor/k8s.io/apiserver/pkg/endpoints/handlers/watch.go:164
  2 // ServeHTTP serves a series of encoded events via HTTP with Transfer-Encoding: chunked
  3 // or over a websocket connection.
  4 func (s *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
  5   kind := s.Scope.Kind
  6   metrics.RegisteredWatchers.WithContext(req.Context()).WithLabelValues(kind.Group, kind.Version, kind.Kind).Inc()
  7   defer metrics.RegisteredWatchers.WithContext(req.Context()).WithLabelValues(kind.Group, kind.Version, kind.Kind).Dec()
  8 
  9   if wsstream.IsWebSocketRequest(req) {
 10     w.Header().Set("Content-Type", s.MediaType)
 11     websocket.Handler(s.HandleWS).ServeHTTP(w, req)
 12     return
 13   }
 14 
 15   flusher, ok := w.(http.Flusher)
 16   if !ok {
 17     err := fmt.Errorf("unable to start watch - can't get http.Flusher: %#v", w)
 18     utilruntime.HandleError(err)
 19     s.Scope.err(errors.NewInternalError(err), w, req)
 20     return
 21   }
 22 
 23   framer := s.Framer.NewFrameWriter(w)
 24   if framer == nil {
 25     // programmer error
 26     err := fmt.Errorf("no stream framing support is available for media type %q", s.MediaType)
 27     utilruntime.HandleError(err)
 28     s.Scope.err(errors.NewBadRequest(err.Error()), w, req)
 29     return
 30   }
 31 
 32   var e streaming.Encoder
 33   var memoryAllocator runtime.MemoryAllocator
 34 
 35   if encoder, supportsAllocator := s.Encoder.(runtime.EncoderWithAllocator); supportsAllocator {
 36     memoryAllocator = runtime.AllocatorPool.Get().(*runtime.Allocator)
 37     defer runtime.AllocatorPool.Put(memoryAllocator)
 38     e = streaming.NewEncoderWithAllocator(framer, encoder, memoryAllocator)
 39   } else {
 40     e = streaming.NewEncoder(framer, s.Encoder)
 41   }
 42 
 43   // ensure the connection times out
 44   timeoutCh, cleanup := s.TimeoutFactory.TimeoutCh()
 45   defer cleanup()
 46 
 47   // begin the stream
 48   w.Header().Set("Content-Type", s.MediaType)
 49   w.Header().Set("Transfer-Encoding", "chunked")
 50   w.WriteHeader(http.StatusOK)
 51   flusher.Flush()
 52 
 53   var unknown runtime.Unknown
 54   internalEvent := &metav1.InternalEvent{}
 55   outEvent := &metav1.WatchEvent{}
 56   buf := &bytes.Buffer{}
 57   ch := s.Watching.ResultChan()
 58   done := req.Context().Done()
 59 
 60   embeddedEncodeFn := s.EmbeddedEncoder.Encode
 61   if encoder, supportsAllocator := s.EmbeddedEncoder.(runtime.EncoderWithAllocator); supportsAllocator {
 62     if memoryAllocator == nil {
 63       // don't put the allocator inside the embeddedEncodeFn as that would allocate memory on every call.
 64       // instead, we allocate the buffer for the entire watch session and release it when we close the connection.
 65       memoryAllocator = runtime.AllocatorPool.Get().(*runtime.Allocator)
 66       defer runtime.AllocatorPool.Put(memoryAllocator)
 67     }
 68     embeddedEncodeFn = func(obj runtime.Object, w io.Writer) error {
 69       return encoder.EncodeWithAllocator(obj, w, memoryAllocator)
 70     }
 71   }
 72 
 73   for {
 74     select {
 75     case <-done:
 76       return
 77     case <-timeoutCh:
 78       return
 79     case event, ok := <-ch:
 80       if !ok {
 81         // End of results.
 82         return
 83       }
 84       metrics.WatchEvents.WithContext(req.Context()).WithLabelValues(kind.Group, kind.Version, kind.Kind).Inc()
 85 
 86       obj := s.Fixup(event.Object)
 87       if err := embeddedEncodeFn(obj, buf); err != nil {
 88         // unexpected error
 89         utilruntime.HandleError(fmt.Errorf("unable to encode watch object %T: %v", obj, err))
 90         return
 91       }
 92 
 93       // ContentType is not required here because we are defaulting to the serializer
 94       // type
 95       unknown.Raw = buf.Bytes()
 96       event.Object = &unknown
 97       metrics.WatchEventsSizes.WithContext(req.Context()).WithLabelValues(kind.Group, kind.Version, kind.Kind).Observe(float64(len(unknown.Raw)))
 98 
 99       *outEvent = metav1.WatchEvent{}
100 
101       // create the external type directly and encode it.  Clients will only recognize the serialization we provide.
102       // The internal event is being reused, not reallocated so its just a few extra assignments to do it this way
103       // and we get the benefit of using conversion functions which already have to stay in sync
104       *internalEvent = metav1.InternalEvent(event)
105       err := metav1.Convert_v1_InternalEvent_To_v1_WatchEvent(internalEvent, outEvent, nil)
106       if err != nil {
107         utilruntime.HandleError(fmt.Errorf("unable to convert watch object: %v", err))
108         // client disconnect.
109         return
110       }
111       if err := e.Encode(outEvent); err != nil {
112         utilruntime.HandleError(fmt.Errorf("unable to encode watch object %T: %v (%#v)", outEvent, err, e))
113         // client disconnect.
114         return
115       }
116       if len(ch) == 0 {
117         flusher.Flush()
118       }
119 
120       buf.Reset()
121     }
122   }
123 }

// s.Watching.ResultChan() 即 cacheWatcher.result
 ch := s.Watching.ResultChan()