Kubernetes client-go 源碼分析 - ListWatcher

概述ListWatch 對象的創建GetterListWatchList() & Watch()

概述

源碼版本信息

• Project: kubernetes
• Branch: master
• Last commit id: d25d741c
• Date: 2021-09-26

ListWatcher 是 Reflector 的一個主要能力提供者，今天我們具體看下 ListWatcher 是如何實現 List() 和 Watch() 過程的。這里我們只跟到 RESTClient 到調用層，不深入 RESTClient 本身的實現；后面有機會再單獨結合 apiserver、etcd 等整體串在一起講 k8s 里的 list-watch 機制底層原理。

ListWatch 對象的創建

ListWatcher 對應的新建實例函數如下：

• client-go/tools/cache/listwatch.go:70

1// 這里 Getter 類型的 c 對應一個 RESTClient
2func NewListWatchFromClient(c Getter, resource string, namespace string, fieldSelector fields.Selector) *ListWatch {
3   optionsModifier := func(options *metav1.ListOptions) {
4      options.FieldSelector = fieldSelector.String() // 序列化成字符串
5   }
6   // 調用下面這個 NewFilteredListWatchFromClient() 函數
7   return NewFilteredListWatchFromClient(c, resource, namespace, optionsModifier)
8}

主要邏輯在下面，list 和 watch 能力都是通過 RESTClient 提供：

 1func NewFilteredListWatchFromClient(c Getter, resource string, namespace string, optionsModifier func(options *metav1.ListOptions)) *ListWatch {
 2   // list 某個 namespace 下的某個 resource
 3   listFunc := func(options metav1.ListOptions) (runtime.Object, error) {
 4      optionsModifier(&options)
 5      return c.Get(). // RESTClient.Get() -> *request.Request
 6         Namespace(namespace).
 7         Resource(resource).
 8         VersionedParams(&options, metav1.ParameterCodec).
 9         Do(context.TODO()).
10         Get()
11   }
12   // watch 某個 namespace 下的某個 resource
13   watchFunc := func(options metav1.ListOptions) (watch.Interface, error) {
14      options.Watch = true
15      optionsModifier(&options)
16      return c.Get().
17         Namespace(namespace).
18         Resource(resource).
19         VersionedParams(&options, metav1.ParameterCodec).
20         Watch(context.TODO())
21   }
22   return &ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
23}

Getter

上面有一個 Getter 接口，看下定義：

• client-go/tools/cache/listwatch.go:65

1type Getter interface {
2   Get() *restclient.Request
3}

這里需要一個能夠獲得 *restclient.Request 的方式

我們實際使用的時候，會用 rest.Interface 接口類型的實例，這是一個相對底層的工具，封裝的是 Kubernetes REST apis 相應動作：

• client-go/rest/client.go:41

 1type Interface interface {
 2   GetRateLimiter() flowcontrol.RateLimiter
 3   Verb(verb string) *Request
 4   Post() *Request
 5   Put() *Request
 6   Patch(pt types.PatchType) *Request
 7   Get() *Request
 8   Delete() *Request
 9   APIVersion() schema.GroupVersion
10}

對應實現是：

• client-go/rest/client.go:81

 1type RESTClient struct {
 2   base *url.URL
 3   versionedAPIPath string
 4   content ClientContentConfig
 5   createBackoffMgr func() BackoffManager
 6   rateLimiter flowcontrol.RateLimiter
 7   warningHandler WarningHandler
 8   Client *http.Client
 9}
10

Getter 接口的 Get() 方法返回的是一個 *restclient.Request 類型，Request 的用法我們直接看 ListWatch 的 New 函數里已經看到是怎么玩的了。

至於這里的 RESTClient 和我們代碼里常用的 Clientset 的關系，這里先簡單舉個例子介紹一下：我們在用 clientset 去 Get 一個指定名字的 DaemonSet 的時候，調用過程類似這樣：

1r.AppsV1().DaemonSets("default").Get(ctx, "test-ds", getOpt)

這里的 Get 其實就是利用了 RESTClient 提供的能力，方法實現對應如下：

 1func (c *daemonSets) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.DaemonSet, err error) {
 2    result = &v1beta1.DaemonSet{}
 3  err = c.client.Get(). // 其實就是 RESTClient.Get()，返回的是 *rest.Request 對象
 4        Namespace(c.ns).
 5        Resource("daemonsets").
 6        Name(name).
 7        VersionedParams(&options, scheme.ParameterCodec).
 8        Do(ctx).
 9        Into(result)
10    return
11}

ListWatch

上面 NewFilteredListWatchFromClient() 函數里實現了 ListFunc 和 WatchFunc 屬性的初始化，我們接着看下 ListWatch 結構體定義：

• client-go/tools/cache/listwatch.go:57

1type ListWatch struct {
2   ListFunc  ListFunc
3   WatchFunc WatchFunc
4   // DisableChunking requests no chunking for this list watcher.
5   DisableChunking bool
6}

實現的接口叫做 ListWatcher

1type ListerWatcher interface {
2    Lister
3    Watcher
4}

這里的 Lister 是

1type Lister interface {
2   // List 的返回值應該是一個 list 類型對象，也就是里面有 Items 字段，里面的 ResourceVersion 可以用來 watch
3   List(options metav1.ListOptions) (runtime.Object, error)
4}

這里的 Watcher 是

1type Watcher interface {
2   // 從指定的資源版本開始 watch
3   Watch(options metav1.ListOptions) (watch.Interface, error)
4}

List() & Watch()

最后 ListWatch 對象的 List() 和 Watch() 的實現就沒有太多新內容了：

• client-go/tools/cache/listwatch.go:103

1func (lw *ListWatch) List(options metav1.ListOptions) (runtime.Object, error) {   // ListWatch 在 Reflector 中使用，在 Reflector 中已經有了分頁邏輯，所以這里不能再添加分頁相關代碼   return lw.ListFunc(options)}func (lw *ListWatch) Watch(options metav1.ListOptions) (watch.Interface, error) {   return lw.WatchFunc(options)}

（轉載請保留本文原始鏈接 https://www.danielhu.cn）