上一篇看到 scheduler、kubelet 都是靠 List-Watch 盯着 apiserver。这一篇往下钻一层:client-go 把 Watch 事件包了两层——Informer 把 Reflector 的原始事件流转成一份本地缓存加一串"发生了什么"的通知,workqueue 再把这些通知压缩成一串待处理的字符串 key。等真正跑到 controller 的业务代码时,你手里拿到的从来不是"发生了什么事件",只是一个"去看看这个东西现在什么样"的提醒。
这一篇继续沿用上一篇的方法:没有本机真实集群实验,而是直接读 client-go v1.37.0 的真实源码,把 workqueue 的去重逻辑原样转写成 Python,用真实断言验证每一条行为。
Informer 不是一个东西,是三个部件接在一起。
上一篇的 Reflector 负责 List-Watch;Informer 在它前面加了一份线程安全的本地缓存(Indexer)——所有对象的最新状态都镜像在这里,业务代码读数据永远读这份本地缓存,不直接问 apiserver。
Informer 的事件回调不会直接跑业务逻辑,只做一件事:把对象的 namespace/name 塞进一个去重队列。真正的处理在另一组 worker goroutine 里,按自己的节奏从队列里取。
worker 拿到的只是一个 key,要用它重新去本地缓存查一遍对象当前的样子,再决定要不要做什么——这是"幂等"这个要求的直接来源。
Informer 的事件处理函数不会把整个对象塞进队列,而是先把它压缩成一个字符串。
staging/src/k8s.io/client-go/tools/cache/store.go · kubernetes @ v1.37.0, L144 // MetaNamespaceKeyFunc is a convenient default KeyFunc which knows how to make // keys for API objects which implement meta.Interface. // The key uses the format <namespace>/<name> unless <namespace> is empty, then // it's just <name>. func MetaNamespaceKeyFunc(obj interface{}) (string, error) { ... }
一个 Pod 对象,不管它当时的 phase、containerStatuses 有多复杂,进队列前都被压成一句 "default/web-1" 这样的字符串。这个设计直接决定了后面所有行为:队列里存的是"谁需要被看一眼",不是"发生了什么"。
同一个 key 短时间内被 Add 很多次,不会被处理很多次——靠两个集合(dirty / processing)配合实现。
staging/src/k8s.io/client-go/util/workqueue/queue.go · kubernetes @ v1.37.0, L196 // dirty defines all of the items that need to be processed. dirty sets.Set[t] // Things that are currently being processed are in the processing set. // These things may be simultaneously in the dirty set. When we finish // processing something and remove it from this set, we'll check if // it's in the dirty set, and if so, add it to the queue. processing sets.Set[t]
staging/src/k8s.io/client-go/util/workqueue/queue.go · kubernetes @ v1.37.0, L233 if q.dirty.Has(item) { // the same item is added again before it is processed, call the Touch // function if the queue cares about it (for e.g, reset its priority) if !q.processing.Has(item) { q.queue.Touch(item) } return }
逻辑分两半:一个 key 已经在排队等待(在 dirty 里但还没被取走),再 Add 一次直接跳过,不会排两条;一个 key 正在被处理(在 processing 里)的时候又被 Add,不会打断当前这次处理,但会在 Done() 的时候被发现"还是脏的",自动重新排回队尾——保证这次新变化不会被漏掉,但也不会让同一个 key 在队列里堆出好几份。
周期性地,每个对象都会被当成"更新"重新丢回队列一次——即使它压根没变过。
staging/src/k8s.io/client-go/tools/cache/shared_informer.go · kubernetes @ v1.37.0, L156 // AddEventHandlerWithResyncPeriod adds an event handler to the // shared informer with the requested resync period; ... The resync operation // consists of delivering to the handler an update notification // for every object in the informer's local cache; it does not add // any interactions with the authoritative storage.
staging/src/k8s.io/client-go/tools/cache/delta_fifo.go · kubernetes @ v1.37.0, L701 // Resync adds, with a Sync type of Delta, every object listed by // `f.knownObjects` whose key is not already queued for processing. func (f *DeltaFIFO) Resync() error { ... }
两个关键点:第一,resync 的数据来源是本地缓存(f.knownObjects),完全不碰 apiserver,不是又去问了一遍集群现状;第二,它给缓存里每一个对象都造了一次"更新"通知。这不是为了发现"漏掉的变化"——Watch 本身(靠 resourceVersion)不会漏事件——而是给 reconcile 逻辑里可能存在的 bug 一次自我修复的机会:哪怕某次 reconcile 因为代码问题什么都没做成,resync 也会definitely再给它一次重跑的机会。
同一个 key default/web-1:一次创建、两次更新(一次排队时到达,一次处理中到达)、一次 resync。跟着队列的 queue / dirty / processing 三个集合一步步看。
事件数据由 Python 脚本按上面两段真实源码逐行转写(Add/Get/Done 的 dirty/processing 集合操作跟真实实现一致),8 个事件里独立统计过"4 次会触发 Add 的事件"和"3 次真正的 Get()"这两个数字,并断言处理次数必须少于原始事件数——这正是去重生效的证据。
既然同一个 key 可能被重复处理(排队去重的巧合、resync 的周期重跑、worker 重启后的重试),reconcile 函数就不能假设"这是我第一次看到这个变化"。
def reconcile(desired_replicas, observed_state):
# 只认"现在的差距",不认"这次事件具体是什么" —— 每次最多补一步
if observed_state["replicas"] < desired_replicas:
observed_state["replicas"] += 1
return True # 这次做了改动
elif observed_state["replicas"] > desired_replicas:
observed_state["replicas"] -= 1
return True
return False # 已经收敛,什么都不用做
| 第几次调用 | replicas | 是否做了改动 |
|---|---|---|
| 1 | 1 | 是 |
| 2 | 2 | 是 |
| 3 | 3(desired) | 是 |
| 4 | 3 | 否(no-op) |
| 5 | 3 | 否(no-op) |
| 6 | 3 | 否(no-op) |
不管这个 reconcile 是被上面演示里的哪一次 Get() 触发的——创建事件、合并后的更新、还是 resync——只要状态已经收敛到期望值,继续调用它就是纯粹的 no-op,不会因为"重复处理"而产生任何副作用。这正是幂等的定义:reconcile(reconcile(x)) == reconcile(x)。真实的 K8s controller(比如 ReplicaSet controller)遵循的是同一个模式,只是每一步补的是"创建/删除一个 Pod",不是简单的加减计数。