Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

Commit 86d098d

Browse files
author
Zhang Jun
committed
update
1 parent 168fa18 commit 86d098d

File tree

2 files changed

+292
-1
lines changed

2 files changed

+292
-1
lines changed

‎client-go/4.sharedInformer.md‎

Lines changed: 261 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,14 @@
44

55
- [SharedInformer](#sharedinformer)
66
- [processorListener](#processorlistener)
7+
- [add() 方法](#add-方法)
8+
- [pop() 方法](#pop-方法)
9+
- [run() 方法](#run-方法)
710
- [sharedProcessor](#sharedprocessor)
11+
- [addListener() 和 addListenerLocked() 方法](#addlistener-和-addlistenerlocked-方法)
12+
- [distribute() 方法](#distribute-方法)
13+
- [run() 方法](#run-方法-1)
14+
- [shouldResync() 方法](#shouldresync-方法)
815
- [SharedInformer 和 SharedIndexInformer](#sharedinformer-和-sharedindexinformer)
916
- [实现 SharedIndexInformer 接口的 sharedIndexInformer 类型](#实现-sharedindexinformer-接口的-sharedindexinformer-类型)
1017
- [Run() 方法](#run-方法)
@@ -32,7 +39,258 @@ SharedInformer 提供一个共享的对象缓存,并且可以将缓存中对
3239
SharedInformer 和 SharedIndexInformer 一般和 workqueue 同时使用,具体参考:[8.customize-controller.md](8.customize-controller.md)
3340

3441
## processorListener
42+
43+
processorListener 封装了监听器处理函数 ResourceEventHandler 以及 RingGrowing 类型的循环队列。
44+
45+
它通过 addCh Channel 接收对象,保存到循环队列 pendingNotifications 中缓存(队列大),然后 pop 到 nextCh,最后 run() 方法获得该对象,调用监听器函数。
46+
47+
``` go
48+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
49+
type processorListener struct {
50+
// nextCh 保存从 pendingNotifications.ReadOne() 读取的对象;
51+
nextCh chan interface{}
52+
// addCh 用于接收 add() 方法发送的对象,pop 方法读取它后 endingNotifications.WriteOne(notificationToAdd) 该对象;
53+
addCh chan interface{}
54+
55+
// 用户实际配置的回调函数
56+
handler ResourceEventHandler
57+
58+
// 循环队列,默认缓存 1024 个对象,从而提供了事件缓冲的能力
59+
pendingNotifications buffer.RingGrowing
60+
61+
// 创建 listner 时,用户指定的 Resync 周期
62+
requestedResyncPeriod time.Duration
63+
64+
// 该 listener 实际使用的 Resync 周期,一般是所有 listner 周期的最小值,所以可能与 requestedResyncPeriod 不同
65+
resyncPeriod time.Duration
66+
67+
// nextResync is the earliest time the listener should get a full resync
68+
nextResync time.Time
69+
// resyncLock guards access to resyncPeriod and nextResync
70+
resyncLock sync.Mutex
71+
}
72+
```
73+
74+
### add() 方法
75+
76+
add() 方法将通知对象写入到 p.addCh Channel 中,如果 pendingNotifications 未满(默认 1024)则可以直接写入,而**不需要等待** ResourceEventHandler 处理完毕。
77+
78+
``` go
79+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
80+
func (p *processorListener) add(notification interface{}) {
81+
p.addCh <- notification
82+
}
83+
```
84+
85+
### pop() 方法
86+
87+
pop() 方法是 processorListener 的核心方法,它实现了:
88+
89+
1. 从 p.addCh Channel 读取数据,存入循环队列 p.pendingNotifications;
90+
2. 从 循环队列 p.pendingNotifications 取数据,写入 p.nextCh,供后续 run() 方法读取;
91+
92+
run() 从 p.nextCh 读取对象,然后调用用户的处理函数,执行时间可能较长,而 pop() 方法通过 channel selector 机制以及循环队列,巧妙地实现了**异步非阻塞**的写入和读取对象。
93+
94+
95+
``` go
96+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
97+
func (p *processorListener) pop() {
98+
defer utilruntime.HandleCrash()
99+
defer close(p.nextCh) // Tell .run() to stop
100+
101+
var nextCh chan<- interface{}
102+
var notification interface{}
103+
for {
104+
select {
105+
// 先将对象写入 p.nextCh,然后从循环对内中读取一个对象,下一次执行另一个 case 时写入 p.nextCh
106+
case nextCh <- notification:
107+
// Notification dispatched
108+
var ok bool
109+
notification, ok = p.pendingNotifications.ReadOne()
110+
if !ok { // Nothing to pop
111+
nextCh = nil // Disable this select case
112+
}
113+
// 从 p.addCh 读取一个待加入的对象,然后看是否有待通知的对象,如果有则设置好通知返回,否则写入循环队列
114+
case notificationToAdd, ok := <-p.addCh:
115+
if !ok {
116+
return
117+
}
118+
if notification == nil { // No notification to pop (and pendingNotifications is empty)
119+
// Optimize the case - skip adding to pendingNotifications
120+
notification = notificationToAdd
121+
nextCh = p.nextCh
122+
} else { // There is already a notification waiting to be dispatched
123+
p.pendingNotifications.WriteOne(notificationToAdd)
124+
}
125+
}
126+
}
127+
}
128+
```
129+
130+
### run() 方法
131+
132+
run() 方法从 p.nextCh channel 中获取通知对象,然后根据对象的类型调用注册的监听器函数。
133+
134+
``` go
135+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
136+
func (p *processorListener) run() {
137+
// this call blocks until the channel is closed. When a panic happens during the notification
138+
// we will catch it, **the offending item will be skipped!**, and after a short delay (one second)
139+
// the next notification will be attempted. This is usually better than the alternative of never
140+
// delivering again.
141+
stopCh := make(chan struct{})
142+
wait.Until(func() {
143+
// this gives us a few quick retries before a long pause and then a few more quick retries
144+
err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
145+
for next := range p.nextCh {
146+
switch notification := next.(type) {
147+
case updateNotification:
148+
p.handler.OnUpdate(notification.oldObj, notification.newObj)
149+
case addNotification:
150+
p.handler.OnAdd(notification.newObj)
151+
case deleteNotification:
152+
p.handler.OnDelete(notification.oldObj)
153+
default:
154+
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %#v", next))
155+
}
156+
}
157+
// the only way to get here is if the p.nextCh is empty and closed
158+
return true, nil
159+
})
160+
161+
// the only way to get here is if the p.nextCh is empty and closed
162+
if err == nil {
163+
close(stopCh)
164+
}
165+
}, 1*time.Minute, stopCh)
166+
}
167+
```
168+
35169
## sharedProcessor
170+
171+
sharedProcessor 类型封装了多个 processorListener,用于表示多组用户注册的通知函数。
172+
173+
``` go
174+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
175+
type sharedProcessor struct {
176+
// 标记 processor 是否启动
177+
listenersStarted bool
178+
listenersLock sync.RWMutex
179+
listeners []*processorListener
180+
syncingListeners []*processorListener
181+
clock clock.Clock
182+
wg wait.Group
183+
}
184+
```
185+
186+
### addListener() 和 addListenerLocked() 方法
187+
188+
addListener() 方法用于向 Processor 添加新的、封装了用户处理函数的 listener,如果 process 的 run() 方法已经在运行,则启动 listener。
189+
190+
``` go
191+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
192+
func (p *sharedProcessor) addListener(listener *processorListener) {
193+
p.listenersLock.Lock()
194+
defer p.listenersLock.Unlock()
195+
196+
p.addListenerLocked(listener)
197+
if p.listenersStarted {
198+
p.wg.Start(listener.run)
199+
p.wg.Start(listener.pop)
200+
}
201+
}
202+
203+
func (p *sharedProcessor) addListenerLocked(listener *processorListener) {
204+
p.listeners = append(p.listeners, listener)
205+
p.syncingListeners = append(p.syncingListeners, listener)
206+
}
207+
```
208+
209+
### distribute() 方法
210+
211+
遍历 listeners,调用他们的 add() 方法添加对象。如果 obj 的事件类型是 Sync,则 sync 为 true。
212+
213+
``` go
214+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
215+
func (p *sharedProcessor) distribute(obj interface{}, sync bool) {
216+
p.listenersLock.RLock()
217+
defer p.listenersLock.RUnlock()
218+
219+
if sync {
220+
for _, listener := range p.syncingListeners {
221+
listener.add(obj)
222+
}
223+
} else {
224+
for _, listener := range p.listeners {
225+
listener.add(obj)
226+
}
227+
}
228+
}
229+
```
230+
231+
### run() 方法
232+
233+
运行已经注册的 listeners。
234+
235+
``` go
236+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
237+
func (p *sharedProcessor) run(stopCh <-chan struct{}) {
238+
func() {
239+
p.listenersLock.RLock()
240+
defer p.listenersLock.RUnlock()
241+
for _, listener := range p.listeners {
242+
p.wg.Start(listener.run)
243+
p.wg.Start(listener.pop)
244+
}
245+
p.listenersStarted = true
246+
}()
247+
<-stopCh
248+
p.listenersLock.RLock()
249+
defer p.listenersLock.RUnlock()
250+
for _, listener := range p.listeners {
251+
close(listener.addCh) // Tell .pop() to stop. .pop() will tell .run() to stop
252+
}
253+
p.wg.Wait() // Wait for all .pop() and .run() to stop
254+
}
255+
```
256+
257+
### shouldResync() 方法
258+
259+
根据已经注册的所有 listerns,判断所有需要 syncing 的 listeners。
260+
261+
``` go
262+
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
263+
func (p *sharedProcessor) shouldResync() bool {
264+
p.listenersLock.Lock()
265+
defer p.listenersLock.Unlock()
266+
267+
p.syncingListeners = []*processorListener{}
268+
269+
resyncNeeded := false
270+
now := p.clock.Now()
271+
for _, listener := range p.listeners {
272+
// need to loop through all the listeners to see if they need to resync so we can prepare any
273+
// listeners that are going to be resyncing.
274+
if listener.shouldResync(now) {
275+
resyncNeeded = true
276+
p.syncingListeners = append(p.syncingListeners, listener)
277+
listener.determineNextResync(now)
278+
}
279+
}
280+
return resyncNeeded
281+
}
282+
283+
func (p *sharedProcessor) resyncCheckPeriodChanged(resyncCheckPeriod time.Duration) {
284+
p.listenersLock.RLock()
285+
defer p.listenersLock.RUnlock()
286+
287+
for _, listener := range p.listeners {
288+
resyncPeriod := determineResyncPeriod(listener.requestedResyncPeriod, resyncCheckPeriod)
289+
listener.setResyncPeriod(resyncPeriod)
290+
}
291+
}
292+
```
293+
36294
## SharedInformer 和 SharedIndexInformer
37295

38296
``` go
@@ -113,7 +371,9 @@ type sharedIndexInformer struct {
113371
// 从 apiserver List/Watch 对象的 Controller
114372
controller Controller
115373

374+
// 可以注册多组监听器的共享 processor
116375
processor *sharedProcessor
376+
117377
cacheMutationDetector CacheMutationDetector
118378

119379
// 创建 controller 使用的 ListerWatcher 和对象类型
@@ -222,6 +482,7 @@ func (s *sharedIndexInformer) HasSynced() bool {
222482
``` go
223483
// 来源于:k8s.io/client-go/tools/cache/shared_informer.go
224484
func (s *sharedIndexInformer) AddEventHandler(handler ResourceEventHandler) {
485+
// 参考后文对 AddEventHandlerWithResyncPeriod 的分析
225486
s.AddEventHandlerWithResyncPeriod(handler, s.defaultEventHandlerResyncPeriod)
226487
}
227488
```

‎client-go/开发笔记.md‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,23 @@
1+
# 目录
2+
3+
<!-- TOC -->
4+
5+
- [目录](#目录)
6+
- [开发自定义 Controller 时,对资源类型的删除回调函数,为何一般不用 workqueue 来处理?](#开发自定义-controller-时对资源类型的删除回调函数为何一般不用-workqueue-来处理)
7+
- [创建 Index 和 DeltaFIFO 的 KeyFunc 为何不同?](#创建-index-和-deltafifo-的-keyfunc-为何不同)
8+
- [自定义 Controller 的 OnDelete Handler 为何要判断对象类型是否是 DeletedFinalStateUnknownn 类型?](#自定义-controller-的-ondelete-handler-为何要判断对象类型是否是-deletedfinalstateunknownn-类型)
9+
- [为何要在调用了 InformerFactory()的 Informer() 方法后,才调用它的 Run() 方法?](#为何要在调用了-informerfactory的-informer-方法后才调用它的-run-方法)
10+
- [为何要等所有 Informer 的 WaitSync 都返回后,再执行 workqueue 的 worker goroutine?](#为何要等所有-informer-的-waitsync-都返回后再执行-workqueue-的-worker-goroutine)
11+
- [如果 pkg/apis/<group>/<versin>/register.go 中没有将自定义 CRD 类型添加到 scheme.AddKnownTypes 中,则创建 Core Group 中的资源对象时出错](#如果-pkgapisgroupversinregistergo-中没有将自定义-crd-类型添加到-schemeaddknowntypes-中则创建-core-group-中的资源对象时出错)
12+
- [启用 client-go 中的日志](#启用-client-go-中的日志)
13+
- [添加自定义类型的步骤](#添加自定义类型的步骤)
14+
- [CRD Kind 冲突 Bug](#crd-kind-冲突-bug)
15+
- [自定义资源对象的名称不能和 K8S 已有的重名](#自定义资源对象的名称不能和-k8s-已有的重名)
16+
- [给对象打 Path 的方法](#给对象打-path-的方法)
17+
- [K8S Job 的 .spec.selector 和 .spec.template 是不能更新的](#k8s-job-的-specselector-和-spectemplate-是不能更新的)
18+
19+
<!-- /TOC -->
20+
121
# 开发自定义 Controller 时,对资源类型的删除回调函数,为何一般不用 workqueue 来处理?
222

323
因为 Informer 内部使用的 controller 会先将对象从 ClientState Indexer 缓存中删除,再调用用户配置的回调函数。
@@ -371,4 +391,14 @@ if err != nil {
371391
log.Fatalln(err)
372392
}
373393
fmt.Println(string(fb))
374-
```
394+
```
395+
396+
# K8S Job 的 .spec.selector 和 .spec.template 是不能更新的
397+
398+
如果更新 K8S Job 的 .spec.template 如 image,就会出错:
399+
400+
``` text
401+
Job.batch "aol-test" is invalid: spec.template: Invalid value: xxx: field is immutable
402+
```
403+
404+
各 K8S 资源类型不可更新的字段,可以参考:https://github.com/pulumi/pulumi-kubernetes/blob/master/pkg/provider/diff.go

0 commit comments

Comments
(0)

AltStyle によって変換されたページ (->オリジナル) /