1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
|
secretInformer := kubecoreinformers.NewSecretInformer()-->
NewFilteredSecretInformer()-->
NewSharedIndexInformer(&cache.ListWatch{}, &corev1.Secret{}, resyncPeriod, indexers)-->
sharedIndexInformer := &sharedIndexInformer{
processor: &sharedProcessor{clock: realClock},
indexer: NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers),
listerWatcher: lw,
objectType: exampleObject,
resyncCheckPeriod: defaultEventHandlerResyncPeriod,
defaultEventHandlerResyncPeriod: defaultEventHandlerResyncPeriod,
cacheMutationDetector: NewCacheMutationDetector(fmt.Sprintf("%T", exampleObject)),
clock: realClock,
}
secretInformer.AddEventHandler()-->
AddEventHandlerWithResyncPeriod()-->
listener := newProcessListener(handler, resyncPeriod, determineResyncPeriod(resyncPeriod, s.resyncCheckPeriod), s.clock.Now(), initialBufferSize)-->
ret := &processorListener{
nextCh: make(chan interface{}),
addCh: make(chan interface{}),
handler: handler,
pendingNotifications: *buffer.NewRingGrowing(bufferSize),
requestedResyncPeriod: requestedResyncPeriod,
resyncPeriod: resyncPeriod,
}
s.processor.addListener(listener)
listener.run()-->
for next := range p.nextCh {
p.handler.OnUpdate(notification.oldObj, notification.newObj)
p.handler.OnAdd(notification.newObj)
p.handler.OnDelete(notification.oldObj)
}
listener.pop()-->
for {
select {
case nextCh <- notification:
notification, ok = p.pendingNotifications.ReadOne()
case notificationToAdd, ok := <-p.addCh:
p.pendingNotifications.WriteOne(notificationToAdd)
}
for _, item := range s.indexer.List() {
listener.add(addNotification{newObj: item})-->
p.addCh <- notification
}
go secretInformer.Run(ctx.Stop)
fifo := NewDeltaFIFOWithOptions()
cfg := &Config{
Queue: fifo,
ListerWatcher: s.listerWatcher,
ObjectType: s.objectType,
FullResyncPeriod: s.resyncCheckPeriod,
RetryOnError: false,
ShouldResync: s.processor.shouldResync,
Process: s.HandleDeltas,
}
wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
wg.StartWithChannel(processorStopCh, s.processor.run)
s.controller = New(cfg)
s.controller.Run(stopCh)-->
r := NewReflector(
c.config.ListerWatcher,
c.config.ObjectType,
c.config.Queue,
c.config.FullResyncPeriod,
)
wg.StartWithChannel(stopCh, r.Run)-->
r.ListAndWatch(stopCh)-->
list := pager.List(context.Background(), options) (1)
items, err := meta.ExtractList(list)
r.syncWith(items, resourceVersion)-->
r.store.Replace(found, resourceVersion) (2)
r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh)-->
r.store.Update(event.Object)
c.processLoop-->
c.config.Queue.Pop(PopProcessFunc(c.config.Process))
for _, d := range obj.(Deltas) {
s.processor.distribute(updateNotification)
s.processor.distribute(addNotification)
s.processor.distribute(deleteNotification)
}
|