apiserver的list-watch代码解读
list-watch，作为k8s系统中统一的异步消息传递方式，对系统的性能、数据一致性起到关键性的作用。今天我想从代码这边探究一下list-watch的实现方式。并看是否能在后面的工作中优化这个过程。
0. list-watch的需求
上图是一个典型的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操作需要做这么几件事：
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)
- 那么就看看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)
- 响应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 :=
