Kubernetes(k8s)代码解读apiserver之listwatch篇

云计算 2023-07-10 张二河 手机阅读

apiserver的list-watch代码解读

list-watch，作为k8s系统中统一的异步消息传递方式，对系统的性能、数据一致性起到关键性的作用。今天我想从代码这边探究一下list-watch的实现方式。并看是否能在后面的工作中优化这个过程。

0. list-watch的需求

20170315101923

上图是一个典型的Pod创建过程，在这个过程中，每次当kubectl创建了ReplicaSet对象后，controller-manager都是通过list-watch这种方式得到了最新的ReplicaSet对象，并执行自己的逻辑来创建Pod对象。其他的几个组件，Scheduler/Kubelet也是一样，通过list-watch得知变化并进行处理。这是组件的处理端代码：

go
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操作需要做这么几件事：

  • 由组件向apiserver而不是etcd发起watch请求，在组件启动时就进行订阅，告诉apiserver需要知道什么数据发生变化。Watch是一个典型的发布-订阅模式。
  • 组件向apiserver发起的watch请求是可以带条件的，例如，scheduler想要watch的是所有未被调度的Pod，也就是满足Pod.destNode=””的Pod来进行调度操作；而kubelet只关心自己节点上的Pod列表。apiserver向etcd发起的watch是没有条件的，只能知道某个数据发生了变化或创建、删除，但不能过滤具体的值。也就是说对象数据的条件过滤必须在apiserver端而不是etcd端完成。
  • 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)  
  • 一个rest.Storage对象会被转换为watcher和lister对象
  • 提供list和watch服务的入口是同一个，在API接口中是通过 GET /pods?watch=true 这种方式来区分是list还是watch
  • 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) 
  • 每次有一个watch的url请求过来，都会调用rw.Watch()创建一个watcher，好吧这里的名字和上面那一层的名字重复了，但我们可以区分开，然后使用serveWatch()来处理这个请求。watcher的生命周期是每个http请求的，这一点非常重要。
  • 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 :=