kubernetes代碼閱讀-apiserver之list-watch篇

apiserver的list-watch代碼解讀

list-watch，作為k8s系統中統一的異步消息傳遞方式，對系統的性能、資料一緻性 起到關鍵性的作用。今天我想從代碼這邊探究一下list-watch的實作方式。并看是否能在後面的工作中優化這個過程。

0. list-watch的需求

上圖是一個典型的Pod建立過程，在這個過程中，每次當kubectl建立了ReplicaSet對象後，controller-manager都是通過list-watch這種方式得到了最新的ReplicaSet對象，并執行自己的邏輯來建立Pod對象。其他的幾個元件，Scheduler/Kubelet也是一樣，通過list-watch得知變化并進行處理。這是元件的處理端代碼：

c.NodeLister.Store, c.nodePopulator = framework.NewInformer(
 c.createNodeLW(), ...(1)
		&api.Node{}, ...(2)
		0, ...(3)
		framework.ResourceEventHandlerFuncs{ ...(4)
			AddFunc: c.addNodeToCache, ...(5)
			UpdateFunc: c.updateNodeInCache,
			DeleteFunc: c.deleteNodeFromCache,
		},
)
           

其中(1)是list-watch函數，(4)(5)則是相應事件觸發操作的入口。

list-watch操作需要做這麼幾件事：

1. 由元件向apiserver而不是etcd發起watch請求，在元件啟動時就進行訂閱，告訴apiserver需要知道什麼資料發生變化。Watch是一個典型的釋出-訂閱模式。
2. 元件向apiserver發起的watch請求是可以帶條件的，例如，scheduler想要watch的是所有未被排程的Pod，也就是滿足Pod.destNode=""的Pod來進行排程操作；而kubelet隻關心自己節點上的Pod清單。apiserver向etcd發起的watch是沒有條件的，隻能知道某個資料發生了變化或建立、删除，但不能過濾具體的值。也就是說對象資料的條件過濾必須在apiserver端而不是etcd端完成。
3. list是watch失敗，資料太過陳舊後的彌補手段，這方面詳見 基于list-watch的Kubernetes異步事件處理架構詳解-用戶端部分 。list本身是一個簡單的清單操作，和其它apiserver的增删改操作一樣，不再多描述細節。

1. watch的API處理

既然watch本身是一個apiserver提供的http restful的API，那麼就按照API的方式去閱讀它的代碼，按照

apiserver的基礎功能實作

一文所描述，我們來看它的代碼，

• 關鍵的處理API注冊代碼

  pkg/apiserver/api_installer.go

func (a *APIInstaller) registerResourceHandlers(path string, storage rest.Storage,...

...
 lister, isLister := storage.(rest.Lister)
	watcher, isWatcher := storage.(rest.Watcher) ...(1)
... 
		case "LIST": // List all resources of a kind. ...(2)
			doc := "list objects of kind " + kind
			if hasSubresource {
				doc = "list " + subresource + " of objects of kind " + kind
			}
			handler := metrics.InstrumentRouteFunc(action.Verb, resource, ListResource(lister, watcher, reqScope, false, a.minRequestTimeout)) ...(3)
           

1. 一個

   rest.Storage

   對象會被轉換為

   watcher

   和

   lister

   對象
2. 提供list和watch服務的入口是同一個，在API接口中是通過

   GET /pods?watch=true

   這種方式來區分是list還是watch
3. API處理函數是由

   lister

   watcher

   經過

   ListResource()

   合體後完成的。

• 那麼就看看

  ListResource()

  的具體實作吧，

  /pkg/apiserver/resthandler.go

func ListResource(r rest.Lister, rw rest.Watcher,... {
...
		if (opts.Watch || forceWatch) && rw != nil {
			watcher, err := rw.Watch(ctx, &opts) ...(1)
 ....
			serveWatch(watcher, scope, req, res, timeout)
			return
		}
		result, err := r.List(ctx, &opts) ...(2) 
		write(http.StatusOK, scope.Kind.GroupVersion(), scope.Serializer, result, w, req.Request)
           

1. 每次有一個watch的url請求過來，都會調用

   rw.Watch()

   建立一個

   watcher

   ，好吧這裡的名字和上面那一層的名字重複了，但我們可以區分開，然後使用

   serveWatch()

   來處理這個請求。watcher的生命周期是每個http請求的，這一點非常重要。
2. list在這裡是另外一個分支，和watch分别處理，可以忽略。

• 響應http請求的過程

  serveWatch()

  的代碼在

  /pkg/apiserver/watch.go

  裡面

func serveWatch(watcher watch.Interface... {
	server.ServeHTTP(res.ResponseWriter, req.Request)
}

func (s *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	for {
		select {
		case event, ok := <-s.watching.ResultChan():

			obj := event.Object

			if err := s.embeddedEncoder.EncodeToStream(obj, buf); 
...
}
           

這段的操作基本毫無技術含量，就是從

watcher

的結果channel中讀取一個event對象，然後持續不斷的編碼寫入到http response的流當中。

• 這是整個過程的圖形化描述：

是以，我們的問題就回到了

1. watcher

   這個對象，嚴格來說是

   watch.Interface

   的對象，位置在

   pkg/watch/watch.go

   中，是怎麼被建立出來的？
2. 這個

   watcher

   對象是怎麼從etcd中獲得變化的資料的？又是怎麼過濾條件的？

2. 在代碼迷宮中追尋watcher

回到上面的代碼追蹤過程來看，watcher(watch.Interface)對象是被Rest.Storage對象建立出來的。從上一篇

可以知道，所有的Rest.Storage分兩層，一層是每個對象自己的邏輯，另一層則是通過通用的操作來搞定，像watch這樣的操作應該是通用的，是以我們看這個源代碼

• /pkg/registry/generic/registry/store.go

func (e *Store) Watch(ctx api.Context, options *api.ListOptions) (watch.Interface, error) {
...
	return e.WatchPredicate(ctx, e.PredicateFunc(label, field), resourceVersion)
}

func (e *Store) WatchPredicate(ctx api.Context, m generic.Matcher, resourceVersion string) (watch.Interface, error) {

			return e.Storage.Watch(ctx, key, resourceVersion, filterFunc) ...(1)

	return e.Storage.WatchList(ctx, e.KeyRootFunc(ctx), resourceVersion, filterFunc)
}
           

果然，我們在(1)這裡找到了生成Watch的函數，但這個工作是由e.Storage來完成的，是以我們需要找一個具體的Storage的生成過程，以Pod為例子

• /pkg/registry/pod/etcd/etcd.go

func NewStorage(opts generic.RESTOptions, k client.ConnectionInfoGetter, proxyTransport http.RoundTripper) PodStorage {
	prefix := "/pods"

	storageInterface := opts.Decorator(
		opts.Storage, cachesize.GetWatchCacheSizeByResource(cachesize.Pods), &api.Pod{}, prefix, pod.Strategy, newListFunc) ...(1)

	store := &registry.Store{
 ...
		Storage: storageInterface, ...(2)
	}
	return PodStorage{
		Pod: &REST{store, proxyTransport}, ...(3)
           

這(1)就是Storage的生成現場，傳入的參數包括了一個緩存Pod的數量。(2)(3)是和上面代碼的連接配接點。那麼現在問題就轉化為追尋

Decorator

這個東西具體是怎麼生成的，需要重複剛才的過程，往上搜尋opts是怎麼搞進來的。

• /pkg/master/master.go - GetRESTOptionsOrDie()

• /pkg/genericapiserver/genericapiserver.go - StorageDecorator()

• /pkg/registry/generic/registry/storage_factory.go - StorageWithCacher()

• /pkg/storage/cacher.go

OK，這樣我們就來到正題，一個具體的watch緩存的實作了！

把上面這個過程用一幅圖表示：

3. watch緩存的具體實作

看代碼，首要看的是資料結構，以及考慮這個資料結構和需要解決的問題之間的關系。

3.1 Cacher(pkg/storage/cacher.go)

對于cacher這結構來說，我們從外看需求，可以知道這是一個Storage，用于提供某個類型的資料，例如Pod的增删改查請求，同時它又用于watch，用于在client端需要對某個key的變化感興趣時，建立一個watcher來源源不斷的提供新的資料給用戶端。

那麼cacher是怎麼滿足這些需求的呢？答案就在它的結構裡面：

type Cacher struct { // Underlying storage.Interface.
	storage Interface

	// "sliding window" of recent changes of objects and the current state.
	watchCache *watchCache
	reflector *cache.Reflector

	// Registered watchers.
	watcherIdx int
	watchers map[int]*cacheWatcher
}
           

略去裡面的鎖（在看代碼的時候一開始要忽略鎖的存在，鎖是後期為了避免破壞資料再加上去的，不影響資料流），略去裡面的一些非關鍵的成員，現在我們剩下這3段重要的成員，其中

• storage

  是連接配接etcd的，也就是背後的裸存儲

• watchCache

  并不僅僅是和注釋裡面說的那樣，是個滑動視窗，裡面存儲了所有資料+滑動視窗

• watchers

  這是為每個請求建立的struct，每個watch的client上來後都會被建立一個，是以這裡有個map

當然，這3個成員的作用是我看了所有代碼後，總結出來的，一開始讀代碼時不妨先在腦子裡面有個定位，然後在看下面的方法時不斷修正這個定位。那麼，接下來就看看具體的方法是怎麼讓資料在這些結構裡面流動的吧！

• 初始化方法

func NewCacherFromConfig(config CacherConfig) *Cacher { 
...
					cacher.startCaching(stopCh)
}

func (c *Cacher) startCaching(stopChannel <-chan struct{}) {
...
	if err := c.reflector.ListAndWatch(stopChannel); err != nil {
		glog.Errorf("unexpected ListAndWatch error: %v", err)
	}
}
           

其他的部分都是陳詞濫調，隻有

startCaching()

這段有點意思，這裡啟動一個go協程，最後啟動了

c.reflector.ListAndWatch()

這個方法，如果對k8s的基本有了解的話，這個其實就是一個把遠端資料源源不斷的同步到本地的方法，那麼資料落在什麼地方呢？往上看可以看到

reflector: cache.NewReflector(listerWatcher, config.Type, watchCache, 0),
           

也就是說從建立cacher的執行個體開始，就會從etcd中把所有Pod的資料同步到watchCache裡面來。這也就印證了watchCache是資料從etcd過來的第一站。

• 增删改方法

func (c *Cacher) Create(ctx context.Context, key string, obj, out runtime.Object, ttl uint64) error {
	return c.storage.Create(ctx, key, obj, out, ttl)
}
           

大部分方法都很無聊，就是短路到底層的storage直接執行。

• Watch方法

// Implements storage.Interface.
func (c *Cacher) Watch(ctx context.Context, key string, resourceVersion string, filter FilterFunc) (watch.Interface, error) {
	
	initEvents, err := c.watchCache.GetAllEventsSinceThreadUnsafe(watchRV)

	watcher := newCacheWatcher(watchRV, initEvents, filterFunction(key, c.keyFunc, filter), forgetWatcher(c, c.watcherIdx))
	c.watchers[c.watcherIdx] = watcher
	c.watcherIdx++
	return watcher, nil
}
           

這裡的邏輯就比較清晰，首先從watchCache中拿到從某個resourceVersion以來的所有資料——initEvents，然後用這個資料建立了一個watcher傳回出去為某個用戶端提供服務。

• List方法

// Implements storage.Interface.
func (c *Cacher) List(ctx context.Context, key string, resourceVersion string, filter FilterFunc, listObj runtime.Object) error {

	filterFunc := filterFunction(key, c.keyFunc, filter)

	objs, readResourceVersion, err := c.watchCache.WaitUntilFreshAndList(listRV)
	if err != nil {
		return fmt.Errorf("failed to wait for fresh list: %v", err)
	}
	for _, obj := range objs {
		if filterFunc(object) {
			listVal.Set(reflect.Append(listVal, reflect.ValueOf(object).Elem()))
		}
	}

}
           

從這段代碼中我們可以看出2件事，一是list的資料都是從watchCache中擷取的，二是擷取後通過filterFunc過濾了一遍然後傳回出去。

3.2 WatchCache(pkg/storage/watch_cache.go)

這個結構應該是緩存的核心結構，從上一層的代碼分析中我們已經知道了對這個結構的需求，包括存儲所有這個類型的資料，包括當有新的資料過來時把資料扔到

cacheWatcher

裡面去，總之，提供List和Watch兩大輸出。

type watchCache struct { // cache is used a cyclic buffer - its first element (with the smallest // resourceVersion) is defined by startIndex, its last element is defined // by endIndex (if cache is full it will be startIndex + capacity). // Both startIndex and endIndex can be greater than buffer capacity - // you should always apply modulo capacity to get an index in cache array.
	cache []watchCacheElement
	startIndex int
	endIndex int // store will effectively support LIST operation from the "end of cache // history" i.e. from the moment just after the newest cached watched event. // It is necessary to effectively allow clients to start watching at now.
	store cache.Store
}
           

這裡的關鍵資料結構依然是2個

• cache

  環形隊列，存儲有限個數的最新資料

• store

  底層實際上是個線程安全的hashMap，存儲全量資料

那麼繼續看看方法是怎麼運轉的吧~

func (w *watchCache) Update(obj interface{}) error {
	event := watch.Event{Type: watch.Modified, Object: object}
	f := func(obj runtime.Object) error { return w.store.Update(obj) }
	return w.processEvent(event, resourceVersion, f)
}


func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, updateFunc func(runtime.Object) error) error {

	previous, exists, err := w.store.Get(event.Object)
	watchCacheEvent := watchCacheEvent{event.Type, event.Object, prevObject, resourceVersion}
		w.onEvent(watchCacheEvent)
	w.updateCache(resourceVersion, watchCacheEvent)

}

// Assumes that lock is already held for write.
func (w *watchCache) updateCache(resourceVersion uint64, event watchCacheEvent) {
	w.cache[w.endIndex%w.capacity] = watchCacheElement{resourceVersion, event}
	w.endIndex++
}
           

所有的增删改方法做的事情都差不多，就是在

store

裡面存具體的資料，然後調用

processEvent()

去增加環形隊列裡面的資料，如果詳細看一下

onEvent

的操作，就會發現這個操作的本質是落在cacher.go裡面：

func (c *Cacher) processEvent(event watchCacheEvent) {
	for _, watcher := range c.watchers {
		watcher.add(event)
	}
}
           

往所有的watcher裡面挨個添加資料。總體來說，我們可以從上面的代碼中得出一個結論：

cache

裡面存儲的是Event，也就是有

prevObject

的，對于所有操作都會在

cache

裡面儲存，但對于store來說，隻存儲當下的資料，删了就删了，改了就改了。

• WaitUntilFreshAndList()

這裡本來應該讨論List()方法的，但在

cacher

裡面的

List()

實際上使用的是這個，是以我們看這個方法。

func (w *watchCache) WaitUntilFreshAndList(resourceVersion uint64) ([]interface{}, uint64, error) {
	startTime := w.clock.Now()
	go func() {
		w.cond.Broadcast()
	}()

	for w.resourceVersion < resourceVersion {
		w.cond.Wait()
	}
	return w.store.List(), w.resourceVersion, nil
}
           

這個方法比較繞，前面使用了一堆

cond

通知來和其他協程通信，最後還是調用了

store.List()

把資料傳回出去。後面來具體分析這裡的協調機制。

• GetAllEventsSinceThreadUnsafe()

這個方法在

cacher

的建立

cacheWatcher

裡面使用，把目前

store

裡面的所有資料都搞出來，然後把

store

裡面的資料都轉換為

AddEvent

，配上

cache

裡面的Event，全部傳回出去。

3.3 CacheWatcher(pkg/storage/cacher.go)

這個結構是每個watch的client都會擁有一個的，從上面的分析中我們也能得出這個結構的需求，就是從

watchCache

裡面搞一些資料，然後寫到用戶端那邊。

// cacherWatch implements watch.Interface
type cacheWatcher struct {
	sync.Mutex
	input chan watchCacheEvent
	result chan watch.Event
	filter FilterFunc
	stopped bool
	forget func(bool)
}
           

這段代碼比較簡單，就不去分析方法了，簡單說就是資料在增加的時候放到

input

這個channel裡面去，通過

filter

然後輸出到

result

這個channel裡面去。

4. 結語

這裡的代碼分析比較冗長，但從中可以得出看代碼的一般邏輯：

• 把資料結構和需求對比着看
• 碰到邏輯複雜的畫個圖來進行記憶
• 在分析的時候把想到的問題記錄下來，然後在後面專門去考慮

這裡我看完代碼後有這些問題：

• 這個cache機制是list-watch操作中最短的闆嗎？
• 在實際生産中，對這List和Wath的使用頻率和方式是怎麼樣的？顯然這兩者存在競争關系
• 目前的資料結構是否是最優的？還有更好的方式嗎？
• 需要一個單元測試來對性能進行測試，然後作為調優的基礎
• etcd v3的一些代碼對我們的機制有什麼影響？這個目錄在

  /pkg/storage/etcd3

本文轉自掘金-

kubernetes代碼閱讀-apiserver之list-watch篇