watch监听 | 南风未起｜云原生/Go技术分享博客

# 1.简介

# 1.1.定义

List-Watch是kubernetes统一的异步消息处理机制，保证消息的实时性、可靠性及顺序性。kube-apiserver作为集群入口，提供资源监听能力。
注意

list和watch本质是一种特殊的get请求，前者基于分页递归查询，后者基于长连接向watcher推送事件

# 1.2.实现

List-Watch的入口是同一个，利用API接口的watch=true标记方式区分list还是watch，接口会在registerResourceHandlers注册。

func ListResource(...) http.HandlerFunc {
	return func(w http.ResponseWriter, req *http.Request) {
		...

        // 获取请求namespace
		namespace, err := scope.Namer.Namespace(req)
		...
        
        // 获取请求name
		_, name, err := scope.Namer.Name(req)
		...

        // 响应类型
		outputMediaType, _, err := negotiation.NegotiateOutputMediaType(req, scope.Serializer, scope)
		...
        
        // 请求参数
		metainternalversionscheme.ParameterCodec.DecodeParameters(req.URL.Query(), scope.MetaGroupVersion,&opts)
        ...

		// fieldSelector转换
		if opts.FieldSelector != nil {
            // 请求的字段名转换为内部字段名
			fn := func(label, value string) (newLabel, newValue string, err error) {
				return scope.Convertor.ConvertFieldLabel(scope.Kind, label, value)
			}
			opts.FieldSelector, err = opts.FieldSelector.Transform(fn)
            ...
		}

        // 请求到具体name
		if hasName {
			nameSelector := fields.OneTermEqualSelector("metadata.name", name)
			// 校验selector条件
			if opts.FieldSelector != nil && !opts.FieldSelector.Empty() {
				selectedName, ok := opts.FieldSelector.RequiresExactMatch("metadata.name")
				if !ok || name != selectedName {
					return
				}
			} else {
				opts.FieldSelector = nameSelector
			}
		}

        // 开启watch监听
		if opts.Watch || forceWatch {
			...
            
            // 激活底层watch流
			watcher, err := rw.Watch(ctx, &opts)
			...

			// 响应watch流			
            serveWatch(watcher, scope, outputMediaType, req, w, timeout)
			return
		}
		...
        
        // 执行List存储查询
		result, err := r.List(ctx, &opts)
		...
        
        // 序列化数据及响应
		transformResponseObject(ctx, scope, req, w, http.StatusOK, outputMediaType, result)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

注意

watch的url请求会调用rw.Watch()创建一个watcher，利用serveWatch()处理这个请求，watcher的生命周期是每个http请求的

# 1.3.serveWatch

serveWatch()会实例化一个watchServer结构体，基于watchServer.ServeHTTP()处理watcher channel监听到的数据及推送到客户端。

// serveWatch will serve a watch response.
func serveWatch(...) {
	defer watcher.Stop()
    
    // 获取请求参数(table转换/字段裁剪)
    options, err := optionsForTransform(mediaTypeOptions, req)
	...

    // 内容协商
	serializer, err := negotiation.NegotiateOutputMediaTypeStream(req, scope.Serializer, scope)
	...
    
    // 事件分帧对象
	framer := serializer.StreamSerializer.Framer
    // 事件序列化对象
	streamSerializer := serializer.StreamSerializer.Serializer
    // 事件编码对象
	encoder := scope.Serializer.EncoderForVersion(streamSerializer, scope.Kind.GroupVersion())
    // 文本类型
	useTextFraming := serializer.EncodesAsText
	...
    
	// 设置watch媒体类型
	mediaType := serializer.MediaType
	if mediaType != runtime.ContentTypeJSON {
		mediaType += ";stream=watch"
	}
	...
    
	// watch流对象相关序列化及GVK
	contentKind, contentSerializer, transform := targetEncodingForTransform(scope, mediaTypeOptions, req)
	// 对象编码调整
	if transform {
		info, ok := runtime.SerializerInfoForMediaType(contentSerializer.SupportedMediaTypes(), serializer.MediaType)
		...
		embeddedEncoder = contentSerializer.EncoderForVersion(info.Serializer, contentKind.GroupVersion())
    // 沿用旧的对象编码
	} else {
		embeddedEncoder = scope.Serializer.EncoderForVersion(serializer.Serializer, contentKind.GroupVersion())
	}
	...
    
    // watchServer构造
	server := &WatchServer{
		Watching: watcher,
		Scope:    scope,

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

		Fixup: func(obj runtime.Object) runtime.Object { ... },

		TimeoutFactory:       &realTimeoutFactory{timeout},
		ServerShuttingDownCh: serverShuttingDownCh,
	}

    // 处理watcher事件
	server.ServeHTTP(w, req)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

注意

serveWatch()本质是将watch.Interface转换为遵循HTTP内容协商、支持transform和framing的流式watch相应

# 1.4.serveHTTP

server.ServeHTTP()核心就是取出watcher channel中的event对象，持续不断的编码及响应到http response，进而推送到客户端。

// ServeHTTP serves a series of encoded events via HTTP with Transfer-Encoding: chunked
// or over a websocket connection.
func (s *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
    // 资源类型
	kind := s.Scope.Kind

    // websocket watch处理(特殊请求)
	if wsstream.IsWebSocketRequest(req) {
		w.Header().Set("Content-Type", s.MediaType)
		websocket.Handler(s.HandleWS).ServeHTTP(w, req)
		return
	}

    // watch是长连接,需要不断flush数据,需要服务器支持chunked flush
	flusher, ok := w.(http.Flusher)
	...

    // 构建frame writer(watch流不是普通的json输出,利用frame边界分隔传输)
	framer := s.Framer.NewFrameWriter(w)
	...

    // 构建streaming encoder(编码事件)
    e = streaming.NewEncoder(framer, s.Encoder)

	// 设置超时
	timeoutCh, cleanup := s.TimeoutFactory.TimeoutCh()
	defer cleanup()

	// 写入HTTP Header,开始chunked流
	w.Header().Set("Content-Type", s.MediaType)
	w.Header().Set("Transfer-Encoding", "chunked")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()
    ...
    
    // 这里其实就是watcher.resultChan
    ch := s.Watching.ResultChan()
	...
    
    // 检查嵌入对象的encoder对象是否实现Allocator(对象编码复用内存,避免频繁GC)
	if encoder, supportsAllocator := s.EmbeddedEncoder.(runtime.EncoderWithAllocator); supportsAllocator {
		// 由AllocatorPool获取一个Allocator
        if memoryAllocator == nil {
			memoryAllocator = runtime.AllocatorPool.Get().(*runtime.Allocator)
			defer runtime.AllocatorPool.Put(memoryAllocator)
		}
        // 基于Allocator包装编码对象(序列化期间复用buf内存)
		embeddedEncodeFn = func(obj runtime.Object, w io.Writer) error {
			return encoder.EncodeWithAllocator(obj, w, memoryAllocator)
		}
	}

	for {
		select {
		case <-s.ServerShuttingDownCh:
			return
		case <-done:
			return
		case <-timeoutCh:
			return
        // 消费watcher数据
		case event, ok := <-ch:
			...

            // 事件对象转换(版本转换/字段裁剪...)
			obj := s.Fixup(event.Object)
            // 对象编码到buf
			embeddedEncodeFn(obj, buf)
            ...

			// 对象编码字节写入event
			unknown.Raw = buf.Bytes()
			event.Object = &unknown
			...

			// 内部事件转为外部事件
			*internalEvent = metav1.InternalEvent(event)
			metav1.Convert_v1_InternalEvent_To_v1_WatchEvent(internalEvent, outEvent, nil)
			...
            
            // 编码输出watchEvent
			e.Encode(outEvent)
            ...
            
            // 没有更多事件就刷新响应
			if len(ch) == 0 {
				flusher.Flush()
			}

            // buf重置
			buf.Reset()
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

注意

serveHTTP()会不停获取watcher channel事件就行编码写出，内部基于allocator和buf进行编码及响应的内存复用

# 2.Cacher

# 2.1.定义

kube-apiserver对etcd的list-watch进行了一层抽象，基于一个大Cache减轻etcd压力，所有对象的最新状态和事件都会存放在Cacher。

// Cacher is responsible for serving WATCH and LIST requests for a given
// resource from its internal cache and updating its cache in the background
// based on the underlying storage contents.
type Cacher struct {
	...
	incoming chan watchCacheEvent								// incoming事件管道,分发给所有的watcher
	...

    storage storage.Interface									// 底层rawStorage对象(交互etcd)

	objectType reflect.Type										// 对象类型,各资源一个Cacher
	...
    
	watchCache *watchCache										// watchCache滑动窗口,维护kind的最新资源和事件数组
	
    reflector  *cache.Reflector									// list-watch实现基础,将事件同步到watchCache
	...
    
	watchersBuffer []*cacheWatcher								// 维护与apiserver建立watch连接的客户端
	...
}

// watchCache implements a Store interface.
type watchCache struct {
	...
	cond *sync.Cond											// list接口基于这个cond同步最新RV,间隔3s会boradcast一次

	capacity int											// 滑动窗口的最大值

	upperBoundCapacity int									// 高峰时自动扩容上限

	lowerBoundCapacity int									// 阶段性下降到基本负荷的下限

	keyFunc func(runtime.Object) (string, error)

	getAttrsFunc func(runtime.Object) (labels.Set, fields.Set, error)

	cache      []*watchCacheEvent							// 事件循环队列
	...
    
	store cache.Indexer										// 缓存etcd最新数据

	resourceVersion uint64									// 当前watchCache最新的RV
	...

	onReplace func()

	eventHandler func(*watchCacheEvent)
	...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

注意

1.Cacher本身实现了storage interface，主要为list/watch服务，其它大部分操作委派给底层的rawStorage

2.watchCache基于cache.Store缓存最新数据，基于[]*watchCacheEvent队列缓存事件，基于cacher.reflector更新

# 2.2.restStore

路由分析提到apiserver基于rest对象注册路由，rest包装底层的storage实现，这个实现就是Cacher，由Cacher包装rawStorage。

// New returns a new instance of CustomResourceDefinitions from the given config.
func (c completedConfig) New(delegation genericapiserver.DelegationTarget) (*CustomResourceDefinitions, error) {
	...
    
    // 创建一个rest对象,内嵌store
    crdStorage, err := customresourcedefinition.NewREST(Scheme, c.GenericConfig.RESTOptionsGetter)
    ...

	return s, nil
}

// NewREST returns a RESTStorage object that will work against API services.
func NewREST(scheme *runtime.Scheme, optsGetter generic.RESTOptionsGetter) (*REST, error) {
	...

    // 初始化rest store对象
	store := &genericregistry.Store{...}
    
	options := &generic.StoreOptions{RESTOptions: optsGetter, AttrFunc: GetAttrs}
	// 补全store对象
    store.CompleteWithOptions(options)
    ...
    
	return &REST{store}, nil
}

// CompleteWithOptions updates the store with the provided options and defaults common fields.
func (e *Store) CompleteWithOptions(options *generic.StoreOptions) error {
	...
    
    // optsGetter就是StorageFactoryRestOptionsFactory
	opts, err := options.RESTOptions.GetRESTOptions(e.DefaultQualifiedResource)
	...

	if e.Storage.Storage == nil {
		e.Storage.Codec = opts.StorageConfig.Codec
		...
        
        // 基于Decorator创建storage对象
		e.Storage.Storage, e.DestroyFunc, err = opts.Decorator(
			opts.StorageConfig, prefix, keyFunc, e.NewFunc, e.NewListFunc, attrFunc,
			options.TriggerFunc, options.Indexers,
		)
		...
        
		e.StorageVersioner = opts.StorageConfig.EncodeVersioner
		...
	}

	return nil
}

func (f *StorageFactoryRestOptionsFactory) GetRESTOptions(resource schema.GroupResource) (RESTOptions, error) {
	storageConfig, err := f.StorageFactory.NewConfig(resource)
	...

	ret := generic.RESTOptions{
		StorageConfig:             storageConfig,
		Decorator:                 generic.UndecoratedStorage,					// 修饰器,后续的etcd client创建
		DeleteCollectionWorkers:   f.Options.DeleteCollectionWorkers,
		EnableGarbageCollection:   f.Options.EnableGarbageCollection,
		ResourcePrefix:            f.StorageFactory.ResourcePrefix(resource),
		CountMetricPollPeriod:     f.Options.StorageConfig.CountMetricPollPeriod,
		StorageObjectCountTracker: f.Options.StorageConfig.StorageObjectCountTracker,
	}
    
    // 默认开启
	if f.Options.EnableWatchCache {
        ...
        
        // Decorator调整为带cacher实现
        ret.Decorator = genericregistry.StorageWithCacher()
	}

	return ret, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

注意

StorageWithCacher()就是用来初始化Cacher对象作为带缓存的storage interface实现，真正的rawStorage作为Cacher的属性

# 2.3.storeCacher

kube-apiserver底层用的是Cacher实现，Cacher内部封装storage及维护etcd的list-watch结果，支持外部注册watcher进行订阅。

// Creates a cacher based given storageConfig.
func StorageWithCacher() generic.StorageDecorator {
	return func(...) (storage.Interface, factory.DestroyFunc, error) {
		// 生成一个rawStorage,带缓存的storage是此基础上进行封装的
		s, d, err := generic.NewRawStorage(storageConfig, newFunc)
		...

        // cacher配置(包装storage)
		cacherConfig := cacherstorage.Config{
			Storage:        s,
			Versioner:      storage.APIObjectVersioner{},
			GroupResource:  storageConfig.GroupResource,
			ResourcePrefix: resourcePrefix,
			KeyFunc:        keyFunc,							// 获取对象对应的name/namespace索引
			NewFunc:        newFunc,							// 获取接收结果的对象
			NewListFunc:    newListFunc,						// 获取接收list结果的对象
			GetAttrsFunc:   getAttrsFunc,						// 获取对象的label/annotations属性
			IndexerFuncs:   triggerFuncs,
			Indexers:       indexers,
			Codec:          storageConfig.Codec,
		}
        // 构造cacher(cacheStorage)
		cacher, err := cacherstorage.NewCacherFromConfig(cacherConfig)
		...
        
        // 终止回调
		destroyFunc := func() {
			once.Do(func() {
                // 终止cacher同步
				cacher.Stop()
                // 终止storage的版本压缩和内存监控
				d()
			})
		}

		return cacher, destroyFunc, nil
	}
}

// NewCacherFromConfig creates a new Cacher responsible for servicing WATCH and LIST requests from
// its internal cache and updating its cache in the background based on the given configuration.
func NewCacherFromConfig(config Config) (*Cacher, error) {
	...
    
	cacher := &Cacher{
		resourcePrefix: config.ResourcePrefix,
		...
		storage:        config.Storage,
		...
		newFunc:        config.NewFunc,
		newListFunc:    config.NewListFunc,
		indexedTrigger: indexedTrigger,
		...
		incoming:              make(chan watchCacheEvent, 100),
		...
	}
	...

    // watchCache构造(滑动窗口+store存储)
	watchCache := newWatchCache(...)
    // listerWatcher初始化
	listerWatcher := NewCacherListerWatcher(config.Storage, config.ResourcePrefix, config.NewListFunc)
	...
    
    // reflector构造(全量List+watch监听)
	reflector := cache.NewNamedReflector(reflectorName, listerWatcher, obj, watchCache, 0)
	...
    
    // cacher使用普通的List-Watch交互etcd,不使用watchCache模拟提供的
	reflector.UseWatchList = false

	cacher.watchCache = watchCache
	cacher.reflector = reflector

    // 启动事件分发器
	go cacher.dispatchEvents()

	cacher.stopWg.Add(1)
	go func() {
		...
        
        // 尝试启动(给1s缓冲期)
		wait.Until(
			func() {
                // cacher未终止
				if !cacher.isStopped() {
                    // 执行List-watch同步
					cacher.startCaching(stopCh)
				}
			}, time.Second, stopCh)
	}()

	return cacher, nil
}

注意

Cacher初始化会构造watchCache和Reflector，启动dispatchEvents事件分发协程和startCaching()同步协程

# 3.同步

# 3.1.startCaching

startCaching()调用reflector的ListAndWatch方法，获取etcd最新数据及更新到watchCache，供dispatchEvents分发协程处理。

func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
	...
    
    // replace回调,更新cacher.ready=true,可以开始watch
	c.watchCache.SetOnReplace(func() {
		successfulList = true
		c.ready.set(true)
		...
	})
    
	defer func() {
		if successfulList {
			c.ready.set(false)
		}
	}()

    // 先终止所有watcher
	c.terminateAllWatchers()
	...
	
    // 执行同步
    c.reflector.ListAndWatch(stopChannel)
    ...
}

// ListAndWatch first lists all items and get the resource version at the moment of call,
// and then use the resource version to watch.
func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
	...

	if r.UseWatchList {
        // 基于watch模拟List及更新watchCache
		w, err = r.watchList(stopCh)
		...
        
		if err != nil {
			...
            // 回退使用etcd List
			fallbackToList = true
			// Ensure that we won't accidentally pass some garbage down the watch.
			w = nil
		}
	}

    // cacher同步/回退
	if fallbackToList {
        // 执行rawStorage List及更新到watchCache
		r.list(stopCh)
		...
	}

	...
    
    // 开始同步
	go r.startResync(stopCh, cancelCh, resyncerrc)
    
    // 进入watch循环
	return r.watch(w, stopCh, resyncerrc)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

注意

startCaching()支持基于watch模拟List同步初始快照及更新watchCache，还支持基于List分页更新，同步完成启用watch持续监听

# 3.2.watchList

reflector.watchList()会执行List同步及更新watchCache数据，同步后会执行一次watch获取watchChan，进行第一次KV监听。

// watchList establishes a stream to get a consistent snapshot of data from the server.
func (r *Reflector) watchList(stopCh <-chan struct{}) (watch.Interface, error) {
	...

	// 开启watch快照循环
	for {
		...

        // 直接执行watch,首次加载的快照全部作为create事件推到watch管道
		w, err = r.listerWatcher.Watch(options)
		...
        
		// 读取watch事件处理
		err = watchHandler(...)
		...
        
        // 初始快照完成就退出
		if *bookmarkReceived {
			break
		}
	}
	...
    
    // 替换store缓存
	r.store.Replace(temporaryStore.List(), resourceVersion)
    ...

	return w, nil
}

// Implements cache.ListerWatcher interface.
func (lw *cacherListerWatcher) Watch(options metav1.ListOptions) (watch.Interface, error) {
	...
    
    // 执行rawStorage.Watch
	return lw.storage.Watch(context.TODO(), lw.resourcePrefix, opts)
}

// Watch implements storage.Interface.Watch.
func (s *store) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) {
	...
    
    // 构造监听索引
	preparedKey, err := s.prepareKey(key)
	...
    
    // 监听RV版本
	rev, err := s.versioner.ParseResourceVersion(opts.ResourceVersion)
	...
    
    // 执行watch
    return s.watcher.Watch(...)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

注意

reflector.watchList()较为特殊，主要解决首次执行watch将List的历史数据作为事件分发给外部处理

# 3.3.watchChan

s.watcher.Watch()是rawStorage初始化注册的watcher，用于构造wc对象及同步初始快照，同步的快照以事件转到wc对象供外部处理。

// Watch watches on a key and returns a watch.Interface that transfers relevant notifications.
func (w *watcher) Watch(...) (watch.Interface, error) {
	...
    
    // 初始化wc对象,用于转发事件
	wc := w.createWatchChan(ctx, key, rev, recursive, progressNotify, transformer, pred)
    
    // 启动watch循环
	go wc.run()

	// 发送watch初始化完成信号
	utilflowcontrol.WatchInitialized(ctx)

	return wc, nil
}

func (wc *watchChan) run() {
	...
    
    // watch etcd数据及放入管道
	go wc.startWatching(watchClosedCh)
	...
    
    // 分发事件
	go wc.processEvent(&resultChanWG)

	select {
	case err := <-wc.errChan:
	...
	case <-watchClosedCh:
	case <-wc.ctx.Done(): // user cancel
	}
	...
}

// watch on given key and send events to process.
func (wc *watchChan) startWatching(watchClosedCh chan struct{}) {
    // rev未初始化,进行初始同步
	if wc.initialRev == 0 {
        // Range数据及生成事件
		wc.sync()
        ...
	}
    ...
    
    // 由最新的rev开始watch etcd数据
	wch := wc.watcher.client.Watch(wc.ctx, wc.key, opts...)
	for wres := range wch {
		...
        
        // 心跳事件
		if wres.IsProgressNotify() {
            // 转发心跳事件
			wc.sendEvent(progressNotifyEvent(wres.Header.GetRevision()))
			...
			continue
		}

        // 真正的资源事件
		for _, e := range wres.Events {
			...
            // 解析
			parsedEvent, err := parseEvent(e)
			...
            // 转发事件
			wc.sendEvent(parsedEvent)
		}
	}
	...
}

// processEvent processes events from etcd watcher and sends results to resultChan.
func (wc *watchChan) processEvent(wg *sync.WaitGroup) {
	...

	for {
		select {
        // 获取转发的事件
		case e := <-wc.incomingEventChan:
			res := wc.transform(e)
			...
			
            // 放入resultChan供外部获取
			select {
			case wc.resultChan <- *res:
			case <-wc.ctx.Done():
				return
			}
		case <-wc.ctx.Done():
			return
		}
	}
}

注意

watchChan间接提供了watch能力，内部执行首次快照同步或watch etcd，基于快照或watch结果生成event及转发到外部

# 3.4.watchHandler

watchHandler用于消费wc对象监听及推到resultChan的事件，针对不同的事件类型执行watchCache同步，此外将事件抛给上层的Cacher。

// watchHandler watches w and sets setLastSyncResourceVersion
func watchHandler(...) error {
...

loop:
	for {
		select {
		...
        
        // 获取wc.resultChan推送的结果
		case event, ok := <-w.ResultChan():
			...
            
            // 获取事件对象
			meta, err := meta.Accessor(event.Object)
			...
            
            // 获取事件RV
			resourceVersion := meta.GetResourceVersion()
            
			switch event.Type {
			case watch.Added:
                // watchCache分发事件及Add对象
				store.Add(event.Object)
				...
			case watch.Modified:
                // watchCache分发事件及Update对象
				store.Update(event.Object)
				...
			case watch.Deleted:
				// watchCache分发事件及Delete对象
				store.Delete(event.Object)
				...
			case watch.Bookmark:
				// 快照同步完成事件
				if _, ok := meta.GetAnnotations()["k8s.io/initial-events-end"]; ok {
					if exitOnInitialEventsEndBookmark != nil {
						*exitOnInitialEventsEndBookmark = true
					}
				}
			...
			}
			...
		}
	}
	...
    
	return nil
}

// processEvent is safe as long as there is at most one call to it in flight
// at any point in time.
func (w *watchCache) processEvent(event watch.Event, rv uint64, hook func(*storeElement) error) error {
	...

    // 获取事件对象唯一键
	key, err := w.keyFunc(event.Object)
	...
    
    // 初始化当前对象
	elem := &storeElement{Key: key, Object: event.Object}
	elem.Labels, elem.Fields, err = w.getAttrsFunc(event.Object)
	...

    // 初始化wcEvent
	wcEvent := &watchCacheEvent{
		Type:            event.Type,
		Object:          elem.Object,
		ObjLabels:       elem.Labels,
		ObjFields:       elem.Fields,
		Key:             key,
		ResourceVersion: resourceVersion,
		RecordTime:      w.clock.Now(),
	}

	if err := func() error {
		...

        // 由watchCache获取缓存的最新对象(前一个版本)
		previous, exists, err := w.store.Get(elem)
		...
        
        // 存在,加到wcEvent作为旧对象
		if exists {
			previousElem := previous.(*storeElement)
			wcEvent.PrevObject = previousElem.Object
			wcEvent.PrevObjLabels = previousElem.Labels
			wcEvent.PrevObjFields = previousElem.Fields
		}

        // 更新事件对象
		w.updateCache(wcEvent)
        // 重置最新RV
		w.resourceVersion = resourceVersion
		defer w.cond.Broadcast()

        // 执行watchCache.store更新回调(新建/更新/删除...)
		return updateFunc(elem)
	}(); err != nil {
		return err
	}

    // 执行事件分发回调,一般是cacher.dispatchEvent()
	if w.eventHandler != nil {
		w.eventHandler(wcEvent)
	}
	return nil
}

// w.eventHandler回调
func (c *Cacher) processEvent(event *watchCacheEvent) {
	...
    
	c.incoming <- *event
}

注意

watchHandler会消费wc.resultChan的最新事件，更新watchCache的事件缓存队列及资源缓存，将event包装后分发给上层的cacher

# 3.5.sync&watch

startResync()循环处理cache.Store同步，内部基于定时器控制频率，这里其实是空转，watch()用于循环监听及分发事件到watchCache。

// startResync periodically calls r.store.Resync() method.
func (r *Reflector) startResync(stopCh <-chan struct{}, cancelCh <-chan struct{}, resyncerrc chan error) {
	...
    
	for {
		select {
        // 定时器
		case <-resyncCh:
		...
		}
        
        // 这里看起来是空转,因为watchCache.Resync()实现为空
		if r.ShouldResync == nil || r.ShouldResync() {
			r.store.Resync()
            ...
		}
        // 清理及重置定时器
		cleanup()
		resyncCh, cleanup = r.resyncChan()
	}
}

func (w *watchCache) Resync() error {
	// Nothing to do
	return nil
}

// watch simply starts a watch request with the server.
func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc chan error) error {
	...

	for {
		...

        // watchChan为空
		if w == nil {
			...
            
            // 初始化watchChan(基于已同步的最新RV进行Watch)
			w, err = r.listerWatcher.Watch(options)
			...
		}

        // 转发事件及更新watchCache
		watchHandler(...)
        
        // 终止当前watch,等待下一轮新建(捕获最新)
		w.Stop()
		w = nil
		...
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

注意

watchList/list进行首次同步及监听，同步完成会执行reflector.Watch()循环监听，每轮监听基于最新RV重置wc分发到watchCache

# 4.分发

# 4.1.dispatchEvents

cacher.dispatchEvents()是cacher初始化构造的另一个协程，主要用于循环处理cacher事件，分发cacher的事件到watcher订阅者。

func (c *Cacher) dispatchEvents() {
	...
    
	for {
		select {
        // 由cacher.incoming管道获取事件
		case event, ok := <-c.incoming:
			...
            
            // 只分发普通event
			if event.Type != watch.Bookmark {
                // 分发到watcher
				c.dispatchEvent(&event)
			}
            
            // bootmark心跳仅更新最新RV
            lastProcessedResourceVersion = event.ResourceVersion
        // 定时bootmark心跳事件
		case <-bookmarkTimer.C():
			...
            
            // 未处理过任何普通事件,仅清理过期的watcher
			if lastProcessedResourceVersion == 0 {
				// pop expired watchers in case there has been no update
				c.bookmarkWatchers.popExpiredWatchers()
				continue
			}
            
            // 构造bootmark事件,通知watcher最新的RV及触发watcher侧的过期终止
			bookmarkEvent := &watchCacheEvent{
				Type:            watch.Bookmark,
				Object:          c.newFunc(),
				ResourceVersion: lastProcessedResourceVersion,
			}
			c.versioner.UpdateObject(bookmarkEvent.Object, bookmarkEvent.ResourceVersion)
            ...
            
            // 分发到watcher
			c.dispatchEvent(bookmarkEvent)
		case <-c.stopCh:
			return
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

注意

cacher.dispatchEvents()将cacher收到的普通事件推送到watcher，同时会定期生成bootmark心跳，将最新RV以事件推到watcher

# 4.2.dispatchEvent

cacher.dispatchEvent()用于投递事件到watcher，针对bootmark事件阻塞失败不会重试，针对普通事件会给一定容忍期，尝试再次投递。

func (c *Cacher) dispatchEvent(event *watchCacheEvent) {
    ...
    
    // bootmark事件
	if event.Type == watch.Bookmark {
        // 转给待分发watcher
		for _, watcher := range c.watchersBuffer {
			watcher.nonblockingAdd(event)
		}
	} else {
		...
        
        // 转给待分发watcher
		for _, watcher := range c.watchersBuffer {
			if !watcher.nonblockingAdd(event) {
                // 分发阻塞,watcher加入blockedWatchers
				c.blockedWatchers = append(c.blockedWatchers, watcher)
			}
		}

        // 处理阻塞的watcher
		if len(c.blockedWatchers) > 0 {
			...
            
            // 转给待分发watcher
			for _, watcher := range c.blockedWatchers {
				if !watcher.add(event, timer) {
					// fired, clean the timer by set it to nil.
					timer = nil
				}
			}
			...
		}
	}
}

func (c *cacheWatcher) nonblockingAdd(event *watchCacheEvent) bool {
	// 旧的bootmark事件丢弃
	if event.Type == watch.Bookmark && event.ResourceVersion < c.bookmarkAfterResourceVersion {
		return false
	}
	select {
    // 推送到watcher
	case c.input <- event:
        // 更新期望的bootmark版本
		c.markBookmarkAfterRvAsReceived(event)
		return true
	default:
		return false
	}
}

func (c *cacheWatcher) add(event *watchCacheEvent, timer *time.Timer) bool {
	// 先尝试非阻塞发送
	if c.nonblockingAdd(event) {
		return true
	}
	...

	// 否则阻塞发送,最大尝试timeout时间
	select {
	case c.input <- event:
		return true
	case <-timer.C:
		closeFunc()
		return false
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

注意

cacher.dispatchEvent()负责将event分发给watcher订阅者，bootmark事件不会进行重试，普通事件会给一定容忍期

# 5.订阅

# 2.1.rw.watch

rw.Watch()根据ListOptions构造一个过滤器(predicate)，执行WatchPredicate()获取watcher对象监听及接收事件。

func (r *REST) Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) {
	return r.store.Watch(ctx, options)
}

// Watch makes a matcher for the given label and field, and calls WatchPredicate.
func (e *Store) Watch(ctx context.Context, options *metainternalversion.ListOptions) (watch.Interface, error) {
	...
    
    // 基于label和field创建Predicate,用于目标对象断言
	predicate := e.PredicateFunc(label, field)

    // resourceVersion构造(watch起点)
	resourceVersion := ""
	if options != nil {
		resourceVersion = options.ResourceVersion
        // 允许客户端接收BOOKMARK事件,用于保持同步但不包含对象变化
		predicate.AllowWatchBookmarks = options.AllowWatchBookmarks
	}
    
    // 调用底层cacher注册watcher
	return e.WatchPredicate(ctx, predicate, resourceVersion, options.SendInitialEvents)
}

// WatchPredicate starts a watch for the items that matches.
func (e *Store) WatchPredicate(...) (watch.Interface, error) {
    ...

	// 上下文加入namespace限制
	if requestNamespace, _ := genericapirequest.NamespaceFrom(ctx); len(requestNamespace) == 0 {
        // 基于label/field限制匹配命名空间
		if selectorNamespace, ok := p.MatchesSingleNamespace(); ok {
            // 验证namespace合法性
			if len(validation.ValidateNamespaceName(selectorNamespace, false)) == 0 {
				ctx = genericapirequest.WithNamespace(ctx, selectorNamespace)
			}
		}
	}

    // watch的根key
	key := e.KeyRootFunc(ctx)
    // 请求限制的name
	if name, ok := p.MatchesSingle(); ok {
        // 生成对象的完整key
		key = e.KeyFunc(ctx, name)
        ...
        // watch单key不需要递归监听
        storageOpts.Recursive = false
	}

    // 获取storage watcher对象
	w, err := e.Storage.Watch(ctx, key, storageOpts)
	...
    
    // 修饰Watcher
	if e.Decorator != nil {
		return newDecoratedWatcher(ctx, w, e.Decorator), nil
	}
	return w, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

注意

rw.Watch() 会构造watch条件，调用cacher.Watch()注册watcher对象，外层可能包装Decorator装饰

# 2.2.watcher

kube-apiserver底层存储用的是Cacher，e.Storage.Watch()会获取watcher对象，watcher就是cacher的cacheWatcher对象。

func (s *DryRunnableStorage) Watch(...) (watch.Interface, error) {
	return s.Storage.Watch(ctx, key, opts)
}

// Watch implements storage.Interface.
func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions) (watch.Interface, error) {
	...

    // 初始化cacheWatcher
	watcher := newCacheWatcher(...)
	...
    
    // 获取RV后的所有事件
    cacheInterval, err = c.watchCache.getAllEventsSinceLocked(startWatchRV)
    ...

	func() {
		...

		// 注册watcher注销回调
		watcher.forget = forgetWatcher(c, watcher, c.watcherIdx, scope, triggerValue, triggerSupported)
		// 设置watcher的bootmark起始RV
		watcher.setBookmarkAfterResourceVersion(bookmarkAfterResourceVersionFn())
        // watcher注册到cacher的watchers接收普通事件
		c.watchers.addWatcher(watcher, c.watcherIdx, scope, triggerValue, triggerSupported)
		addedWatcher = true

		// watcher注册到cacher的bootmarkWatcher接收bootmark事件
		if watcher.allowWatchBookmarks {
			c.bookmarkWatchers.addWatcher(watcher)
		}
		c.watcherIdx++
	}()

    // 未注册,返回一个立即关闭的watcher,模拟client刚连上就终止的语义
	if !addedWatcher {
		return newImmediateCloseWatcher(), nil
	}

    // 启用事件分发
	go watcher.processInterval(ctx, cacheInterval, startWatchRV)
	return watcher, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

注意

Watch()本质上构造一个cacheWatcher，将cacheWatcher注册到Cacher订阅事件，基于watcher.processInterval()分发

# 2.3.process

watcher.processInterval()负责同步incoming管道的事件到resultChan，serveWatch()就会从watcher.resultChan消费及响应。

func (c *cacheWatcher) processInterval(ctx context.Context, cacheInterval *watchCacheInterval, rv uint64) {
	,,,
    
	for {
        // 获取cacheInterval的事件
		event, err := cacheInterval.Next()
        ...
        
        // 全部消费退出
		if event == nil {
			break
		}
        
        // 将event转到resultChan
		c.sendWatchCacheEvent(event)

		// 更新RV
		if event.ResourceVersion > resourceVersion {
			resourceVersion = event.ResourceVersion
		}
	}
	...

    // 初始事件处理完毕,开始正式消费增量事件
	c.process(ctx, resourceVersion)
}

func (c *cacheWatcher) process(ctx context.Context, resourceVersion uint64) {
	...

	for {
		select {
        // 持续消费watcher收到的增量事件
		case event, ok := <-c.input:
			...
            
            // 发送到watcher.resultChan
            c.sendWatchCacheEvent(event)
		case <-ctx.Done():
			return
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

注意

这里有一点比较疑惑，为什么cacher的事件推到watcher inputChan不直接进行消费，还要基于异步协程由input-->result

# 6.总结

注意

由整个流程可以看出，watch请求总会关联一个cacheWatcher订阅Cacher事件，Cacher事件又来源于watchCache和Reflector