字节跳动员工人数变化,怀化百度关键词优化公司,网站换程序301,乡村旅游网站开发k8s informer的re-list与resync
1、informer的list-watch机制 client-go中的reflector模块首先会list apiserver获取某个资源的全量信息#xff0c;然后根据list到的resourceversion来watch资源的增量信息。且希望使用client-go编写的控制器组件在与apiserver发生连接异常时然后根据list到的resourceversion来watch资源的增量信息。且希望使用client-go编写的控制器组件在与apiserver发生连接异常时尽量的re-watch资源而不是re-list 2、re-list的场景
场景一very short watch
reflector与api建立watch连接但apiserver关闭了连接则会重新re-list
这意味着 apiserver 接受了监视请求但立即终止了连接如果您偶尔看到它则表明存在暂时性错误并不值得警惕。如果您反复看到它则意味着 apiserver或 etcd有问题。 I0728 11:32:06.170821 67483 streamwatcher.go:114] Unexpected EOF during watch stream event decoding: unexpected EOF I0728 11:32:06.171062 67483 reflector.go:391] k8s.io/client-go/informers/factory.go:134: Watch close - *v1.Deployment total 0 items received W0728 11:32:06.187394 67483 reflector.go:302] k8s.io/client-go/informers/factory.go:134: watch of *v1.Deployment ended with: very short watch: k8s.io/client-go/informers/factory.go:134: Unexpected watch close - watch lasted less than a second and no items received 场景二401 Gone
为什么跟etcd不会一直记录历史版本有关参考bookmark机制
reflector与api建立watch连接但是出现watch的相关事件丢失时etcd不会一直记录历史版本api返回401 Gonereflector提示too old resource version并重新re-list I0728 14:40:58.807670 71423 reflector.go:300] k8s.io/client-go/informers/factory.go:134: watch of *v1.Deployment ended with: too old resource version: 332167941 (332223202) I0728 14:40:59.808153 71423 reflector.go:159] Listing and watching *v1.Deployment from k8s.io/client-go/informers/factory.go:134 I0728 14:41:00.300695 71423 reflector.go:312] reflector list resourceVersion: 332226582 3、resync场景
// k8s.io/client-go/tools/cache/delta_fifo.go
// 重新同步一次 Indexer 缓存数据到 Delta FIFO 队列中
func (f *DeltaFIFO) Resync() error {f.lock.Lock()defer f.lock.Unlock()if f.knownObjects nil {return nil}// 遍历 indexer 中的 key传入 syncKeyLocked 中处理keys : f.knownObjects.ListKeys()for _, k : range keys {if err : f.syncKeyLocked(k); err ! nil {return err}}return nil
}func (f *DeltaFIFO) syncKeyLocked(key string) error {obj, exists, err : f.knownObjects.GetByKey(key)if err ! nil {klog.Errorf(Unexpected error %v during lookup of key %v, unable to queue object for sync, err, key)return nil} else if !exists {klog.Infof(Key %v does not exist in known objects store, unable to queue object for sync, key)return nil}// 如果发现 FIFO 队列中已经有相同 key 的 event 进来了说明该资源对象有了新的 event// 在 Indexer 中旧的缓存应该失效因此不做 Resync 处理直接返回 nilid, err : f.KeyOf(obj)if err ! nil {return KeyError{obj, err}}if len(f.items[id]) 0 {return nil}// 重新放入 FIFO 队列中if err : f.queueActionLocked(Sync, obj); err ! nil {return fmt.Errorf(couldnt queue object: %v, err)}return nil
}为什么需要 Resync 机制呢因为在处理 SharedInformer 事件回调时可能存在处理失败的情况定时的 Resync 让这些处理失败的事件有了重新处理的机会。
那么经过 Resync 重新放入 Delta FIFO 队列的事件和直接从 apiserver 中 watch 得到的事件处理起来有什么不一样呢
// k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {s.blockDeltas.Lock()defer s.blockDeltas.Unlock()// from oldest to newestfor _, d : range obj.(Deltas) {// 判断事件类型看事件是通过新增、更新、替换、删除还是 Resync 重新同步产生的switch d.Type {case Sync, Replaced, Added, Updated:s.cacheMutationDetector.AddObject(d.Object)if old, exists, err : s.indexer.Get(d.Object); err nil exists {if err : s.indexer.Update(d.Object); err ! nil {return err}isSync : falseswitch {case d.Type Sync:// 如果是通过 Resync 重新同步得到的事件则做个标记isSync truecase d.Type Replaced:...}// 如果是通过 Resync 重新同步得到的事件则触发 onUpdate 回调s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)} else {if err : s.indexer.Add(d.Object); err ! nil {return err}s.processor.distribute(addNotification{newObj: d.Object}, false)}case Deleted:if err : s.indexer.Delete(d.Object); err ! nil {return err}s.processor.distribute(deleteNotification{oldObj: d.Object}, false)}}return nil
}从上面对 Delta FIFO 的队列处理源码可看出如果是从 Resync 重新同步到 Delta FIFO 队列的事件会分发到 updateNotification 中触发 onUpdate 的回调 Resync 机制的引入定时将 Indexer 缓存事件重新同步到 Delta FIFO 队列中在处理 SharedInformer 事件回调时让处理失败的事件得到重新处理。并且通过入队前判断 FIFO 队列中是否已经有了更新版本的 event来决定是否丢弃 Indexer 缓存不进行 Resync 入队。在处理 Delta FIFO 队列中的 Resync 的事件数据时触发 onUpdate 回调来让事件重新处理。 4、Reflector.lastSyncResourceVersion 是哪个资源的 resourceVersion 一个resourceVersion怎么对应多种资源呢其实是一个informer对应一个Reflector一个informer本来就是对应一种资源的然后每一类resourceVersion是不断递增的比如informer watch了pod那么informer对应的Reflector的resourceVersion是对应k8s etcd里pod这一类资源的resourceVersion。因此是一类资源使用一个resourceVersion list-watch example https://codeburst.io/kubernetes-watches-by-example-bc1edfb2f83 5、注意点 1、resync不是re-listresync不需要访问apiserver 2、resync 是重放 informer 中的 obj 到 DeltaFIFO 队列中触发 handler 再次处理 obj。 目的是防止有些 handler 处理失败了而缺乏重试的机会。特别是需要修改外部系统的状态的时候需要做一些补偿的时候。 比如说根据 networkpolicy刷新 node 上的 iptables。 iptables 有可能会被其他进程或者管理员意外修改有 resync 的话才有机会定期修正。 这也说明回调函数的实现需要保证幂等性。对于 OnUpdate 函数而言有可能会拿到完全一样的两个 Obj实现 OnUpdate 时要考虑到。 3、re-list 是指 reflector 重新调用 kube-apiserver 全量同步所有 obj。但目前(v1.20)没有显式配置 re-list 周期的参数。 list 的时机一般是在程序第一次启动或者 watch 有错误才会 re-list。 4、resync 是一个水平触发的模式 水平触发是只要处于某个状态就会一直通知。比如在这里对象已经在缓存里会触发不止一次回调函数。 5、process 函数怎么区分从 DeltaFIFO 里拿到的 obj 是新的还是重放的呢 根据 obj 的 keynamespace/name从 index 里拿到旧的 obj和新出队的 obj 比较 resource revision这两个 resource revision 如果一样就是重放的如果不一样就是从 kube-apiserver 拿到的新的。因为 resource revision 只有在 etcd 才能更新。 index 作为客户端缓存这个值是不变的。 6、sharedInformer 如何实现 Resync 把需要 resyncresync 的周期到了 的 listeners 数组复制一份到 syncingListeners在 distribute 中 会调用 syncingListeners 的 add 函数触发 syncingListeners 上的回调函数
6、resync要注意的问题 1、如何配置resync的周期 func NewSharedIndexInformer(lw ListerWatcher, exampleObject runtime.Object, defaultEventHandlerResyncPeriod time.Duration, indexers Indexers) 第三个参数 defaultEventHandlerResyncPeriod 就指定多久 resync 一次。如果为0则不 resync。 AddEventHandlerWithResyncPeriod也可以给单独的 handler 定义 resync period否则默认和 informer 的是一样的。 2、配置resync周期间隔太小会有什么问题 此时会以比较高的频率促使事件重新入队进行reconcile造成controller的压力过大 3、resync用于解决什么问题resync 多久一次比较合适或者需不需要 resync 根据具体业务场景来根据外部状态是不是稳定的、是否需要做这个补偿来决定的举例假设controller是一个LB controller 当watch到了service创建然后调用LB api去创建一个对应的LB然后如果此时这个对应的LB由于某种bug被删除了此时service就不通了 那么此时状态不一致了集群里有这个serviceLB那边没有对应的LB并且由于bug被删除了而不是删除service而触发LB删除的此时service是没有变化的 也就不会出发reconcile了。假设我们reconcile里有逻辑是判断如果service没有对应的LB就创建那么此时reconcile不会被出发那也就没有被执行了。 此时如果有resync定时将indexer里的对象也就是缓存的对象来一次update事件的入队进行后续出队触发reconcile那我们就会发现service对应的LB没了 进而进行创建。也就是本质上resync是防止业务层的bug。且resync将indexer的对象重入队里面的service不是所有service而是创建了LB的service因为我们 只会watch我们关心的资源前提是代码里添加的对象写的是有LB字段的service 因为这些操作都是异步的 合理的sync可以提高事件消费的容错性。Resync 机制的引入定时将 Indexer 缓存事件重新同步到 Delta FIFO 队列中在处理 SharedInformer 事件回调时让处理失败的事件得到重新处理。并且通过入队前判断 FIFO 队列中是否已经有了更新版本的 event来决定是否丢弃 Indexer 缓存不进行 Resync 入队。在处理 Delta FIFO 队列中的 Resync 的事件数据时触发 onUpdate 回调来让事件重新处理。
7、这个issue讨论里面有Programming Kubernetes相关讨论内容
https://github.com/cloudnativeto/sig-kubernetes/issues/11
