网站框架代码,遂溪 网站,游戏门户网站建设,中国农村建设网站[阅读指南] 这是该系列第二篇 基于kubernetes 1.27 stage版本 为了方便阅读#xff0c;后续所有代码均省略了错误处理及与关注逻辑无关的部分。 文章目录 Reflector是什么整体结构工作流程list拉取数据缓存resync操作watch监听操作 总结 Reflector是什么
reflector在informer… [阅读指南] 这是该系列第二篇 基于kubernetes 1.27 stage版本 为了方便阅读后续所有代码均省略了错误处理及与关注逻辑无关的部分。 文章目录 Reflector是什么整体结构工作流程list拉取数据缓存resync操作watch监听操作 总结 Reflector是什么
reflector在informer中就像是一个对外的窗口它与api-server建立连接监听和获取来自api-server的资源变化信息并把这些信息放进deltaFIFO中交给下一个环节处理。
整体结构
与api-server进行交互通过list获取指定的全量资源watch监听指定的资源变化事件并将这些事件放入delta FIFO队列中。 结构与交互如下图
// 省略了部分字段只留下我们关注的
type Reflector struct {// name identifies this reflector. By default it will be a file:line if possible.name string// reflector对象需要监控的资源类型比如上一节workqueue中的v1.Pod{}expectedType reflect.Type// deltaFIFO 队列存储对象store Store// 实现list/watchlisterWatcher ListerWatcher// 上次更新的资源版本号用来判断当前的node的资源状况lastSyncResourceVersion string......
}工作流程
reflecter主函数比较简单循环同步运行ListAndWatch直到收到stop信号。
func (r *Reflector) Run(stopCh -chan struct{}) {wait.BackoffUntil(func() {if err : r.ListAndWatch(stopCh); err ! nil {r.watchErrorHandler(r, err)}}, r.backoffManager, true, stopCh)
}ListAndWatch主要做了这几件事
通过stream或者chunk方式拉取全量list数据开启一个协程进行缓存resync操作。循环执行watch监听操作
func (r *Reflector) ListAndWatch(stopCh -chan struct{}) error {...fallbackToList : !r.UseWatchList// stream式同步if r.UseWatchList {w, err r.watchList(stopCh)...if err ! nil {...fallbackToList truew nil}}// chunk式同步if fallbackToList {err r.list(stopCh)if err ! nil {return err}}...go r.startResync(stopCh, cancelCh, resyncerrc)return r.watch(w, stopCh, resyncerrc)
}接下来咱一步步来看。
list拉取数据
ListAndWatch拉取全量数据时出现了两种数据拉取的方式list /watch和stream list /watch。 stream list是 kubernetes 1.27 引入的新方案通过 ENABLE_CLIENT_GO_WATCH_LIST_ALPHA 变量可以启用stream list默认会使用原有的list/watch。后续会单独开一篇介绍stream list方案详情可以通过KEP-3157了解 前者在初始化时list拉取全量数据通过watch更新增量变化。 后者可以通过watch 请求的方式获取list数据从而减轻大规模集群初始化list数据时的资源消耗。 在建立watch连接时携带如下两个参数即可告知服务器使用streaming list进行一致性读取。 sendInitialEventstrue resourceVersionMatchNotOlderThan
常规的list流程借用这个博主画的时序图来看下。
缓存resync操作
resync负责定期将本地的缓存重新加入deltaFIFO队列确保本地缓存与controller的数据一致性。
国内太多博客没了解清楚就介绍这一部分是与api-server交互进行relist。实际上resync完全没有涉及到服务端的部分他就是一个本地缓存的同步机制。与服务端的交互使用list/watch已经完全可以确保资源一致性了基本不怎么需要进行relist操作并且对于节点非常多的大集群来说list非常消耗资源何况是定期relist呢。
关于resync机制的介绍不在这里展开详细看下一篇笔记。
watch监听操作
watch的实现非常巧妙它利用了http的chunk编码传输机制建立长连接来实现动态的数据监听可以了解分块传输编码。 同样借用一张时序图来看下watch的流程
reflector通过Watcher监听api-server端的数据delta事件并将这些事件放入deltaFIFO中统一处理。
// 在这里向服务端发起watch请求并接收和处理资源变更事件
func (r *Reflector) watch(w watch.Interface, stopCh -chan struct{}, resyncerrc chan error) error {...for {...// w nil表示使用常规的list/watch方式streaming 方式会创建特殊的watcherif w nil {timeoutSeconds : int64(minWatchTimeout.Seconds() * (rand.Float64() 1.0))options : metav1.ListOptions{// 上次同步的资源版本也就是本地的资源版本。以此来获取增量的数据ResourceVersion: r.LastSyncResourceVersion(),// watch 超时时间长时间没有接受任务事件的watcher会被关掉避免长时间挂起。TimeoutSeconds: timeoutSeconds,// watch书签避免watch重启时请求api-server导致的消耗。AllowWatchBookmarks: true,}// 创建一个watch对象监听api-server的资源变更事件将接收到的事件丢进resultChan中w, err r.listerWatcher.Watch(options)...}// 将resultChan中的取出放入FIFO 队列err watchHandler(start, w, r.store, r.expectedType, r.expectedGVK, r.name, r.typeDescription, r.setLastSyncResourceVersion, nil, r.clock, resyncerrc, stopCh)// 失败重试逻辑...}
}建立连接的逻辑在这一行 w, err r.listerWatcher.Watch(options)
还是用上一篇workqueue来看看这个Watch实例的实现。 从Watch函数一路往上追溯可以看到先是与server建立了http连接再通过watch标记建立了watch连接创建stream watcher对象并拉起一个协程去处理监听到的事件信息。
此后所有监听的delta事件都会经过receive协程进入到resultChan中。
// reflector调用的watch函数
func (lw *ListWatch) Watch(options metav1.ListOptions) (watch.Interface, error) {return lw.WatchFunc(options)
}// watchFunc函数的定义
func NewFilteredListWatchFromClient(c Getter, resource string, namespace string, optionsModifier func(options *metav1.ListOptions)) *ListWatch {...watchFunc : func(options metav1.ListOptions) (watch.Interface, error) {options.Watch true // 向服务端请求chunk连接optionsModifier(options)return c.Get().Namespace(namespace).Resource(resource).VersionedParams(options, metav1.ParameterCodec).Watch(context.TODO()) // 这里调用了getter的watch函数// getter是controller初始化时建立的http客户端 clientset.CoreV1().RESTClient()}return ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
}func (r *Request) Watch(ctx context.Context) (watch.Interface, error) {... url : r.URL().String()for {req, err : r.newHTTPRequest(ctx)resp, err : client.Do(req)if err nil resp.StatusCode http.StatusOK {return r.newStreamWatcher(resp)}...}
}func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher {sw : StreamWatcher{source: d,reporter: r,result: make(chan Event),done: make(chan struct{}),}go sw.receive() // 处理消息事件的协程return sw
}// 解析接收到的事件并放到resultChan中等待后续处理。
func (sw *StreamWatcher) receive() {for {// 解析数据action, obj, err : sw.source.Decode()select {case -sw.done:return// 将事件发送到resultChancase sw.result - Event{Type: action,Object: obj,}:}}
}进入resultChan的事件由watchHandler取出再分类添加到FIFO队列中。
func watchHandler(start time.Time,w watch.Interface, // watch实例store Store, // 存储对象 比如delta FIFO queue...
) error {...
loop:for {select {case -stopCh:return errorStopRequestedcase err : -errc:return err// 从ResultChan中取出变更事件并放进队列中比如delta FIFO队列中case event, ok : -w.ResultChan():// 省略了一些资源过滤和错误处理...// 解析监听到的事件数据meta, err : meta.Accessor(event.Object)if err ! nil {utilruntime.HandleError(fmt.Errorf(%s: unable to understand watch event %#v, name, event))continue}// 解析资源事件的版本resourceVersion : meta.GetResourceVersion()switch event.Type {case watch.Added: err : store.Add(event.Object) // 往队列添加add delta事件... // err handlecase watch.Modified: err : store.Update(event.Object) // 往队列添加update delta事件... // err handlecase watch.Deleted: err : store.Delete(event.Object) // 往队列添加delete delta事件在此之前会判断事件对应的资源对象是否存在... // err handlecase watch.Bookmark:...default:... // err handle}// 更新resourceVersion版本号下一轮watch就不会再收到重复的更新事件setLastSyncResourceVersion(resourceVersion)if rvu, ok : store.(ResourceVersionUpdater); ok {rvu.UpdateResourceVersion(resourceVersion)}...}}...return nil
}总结
用一个图来回顾下reflector各个模块的关系
