kubernets|kubernetes源码剖析之client-go(二) Informer机制
kubernetes源码剖析之client-go(一) Informer机制 ? Kubernetes通过informer机制，实现在不依赖任何中间件的情况下保证消息的实时性、可靠性、顺序性。其他Kubernetes组件通过client-go的informer机制与Api Server进行通信。Informer的核心组件包括：
- Reflector
?用于监控(Watch)指定Kubernetes资源 - DeltaFIFO
? Delta的先进先出队列，Reflector为生产者，Controller为消费者 - Indexer
?自带索引功能的本地存储，用于存储资源对象
文章图片
代码示例
package mainimport (
"log"
"time"v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/clientcmd"
)func main() {config, err := clientcmd.BuildConfigFromFlags("", "D:\\coding\\config")
if err != nil {
panic(err)
}// Imformer通过clientset与Api Server通信
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
panic(err)
}// 创建stopCH对象，用于进程退出前通知Imformer提前退出
stopCh := make(chan struct{})
defer close(stopCh)// 实例化SharedInformer对象，参数clientset用于与Api Server交互， time.Minute设定resync周期，0为禁用resync
// 通过map共享Informer( informers map[reflect.Type]cache.SharedIndexInformer )，避免同一资源的Informer被重复实例化
sharedInformers := informers.NewSharedInformerFactory(clientset, time.Minute)
// 获取Pod资源的informer对象
// 每一个K8S资源都会实现Informer机制，每个Informer实现都会提供Informer和Lister方法
informer := sharedInformers.Core().V1().Pods().Informer()// 添加资源的回调方法
informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
// 创建资源时的回调方法
AddFunc: func(obj interface{}) {
mObj := obj.(v1.Object)
log.Printf("New Pod Added to Stroe: %s", mObj.GetName())},
// 更新资源时的回调方法
UpdateFunc: func(oldObj, newObj interface{}) {
oObj := oldObj.(v1.Object)
nObj := newObj.(v1.Object)
log.Printf("%s Pod Updated to %s", oObj.GetName(), nObj.GetName())
},
// 删除资源时的回调方法
DeleteFunc: func(obj interface{}) {
mObj := obj.(v1.Object)
log.Printf("Pod Deleted from Stroe : %s", mObj.GetName())
},
})informer.Run(stopCh)
}
Reflector ?Informer对Kubernetes的Api Server资源进行监控（Watch）操作。其中最核心的功能是Reflector，Reflector用于监控指定的Kubernetes资源，当监控的资源发生变化时，触发相应的变更事件。并将其资源对象存放到本地缓冲DeltaFIFO中。
?通过NewReflector方法实例化Reflector对象，方法必须传入ListerWatcher数据接口对象。 ListerWatcher拥有List和Watch方法，用于获取和监控资源列表，只要实现了List和Watch方法的对象都可以成为ListerWatcher。
#源码路径 vender\k8s.io\client-go\tools\cache\reflector.go// NewReflector creates a new Reflector object which will keep the given store up to
// date with the server's contents for the given resource. Reflector promises to
// only put things in the store that have the type of expectedType, unless expectedType
// is nil. If resyncPeriod is non-zero, then lists will be executed after every
// resyncPeriod, so that you can use reflectors to periodically process everything as
// well as incrementally processing the things that change.
func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod)
}? Reflector通过Run函数启动监控进程，并处理监控的事件。其中最主要的是ListAndWatch函数，它负责List和Watch指定的Kubernetes Api Server资源。
#源码路径 vender\k8s.io\client-go\tools\cache\reflector.go
// Run starts a watch and handles watch events. Will restart the watch if it is closed.
// Run will exit when stopCh is closed.
func (r *Reflector) Run(stopCh <-chan struct{}) {
klog.V(3).Infof("Starting reflector %v (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
wait.Until(func() {
if err := r.ListAndWatch(stopCh);
err != nil {
utilruntime.HandleError(err)
}
}, r.period, stopCh)
}ListAndWatch函数
? ListAndWatch第一次运行时，通过List获取资源下的所有对象和版本信息，后续通过版本进行watch
#源码路径 vender\k8s.io\client-go\tools\cache\reflector.go
// ListAndWatch first lists all items and get the resource version at the moment of call,
// and then use the resource version to watch.
// It returns error if ListAndWatch didn't even try to initialize watch.
func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
klog.V(3).Infof("Listing and watching %v from %s", r.expectedTypeName, r.name)
var resourceVersion string// Explicitly set "0" as resource version - it's fine for the List()
// to be served from cache and potentially be delayed relative to
// etcd contents. Reflector framework will catch up via Watch() eventually.
// 初始化时将版本置为0
options := metav1.ListOptions{ResourceVersion: "0"}if err := func() error {
...
// Attempt to gather list in chunks, if supported by listerWatcher, if not, the first
// list request will return the full response.
pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
// 获取资源下的所有对象的数据，当传入的opts的ResourceVersion为0时，返回 全量数据。非0时返回增量数据。
return r.listerWatcher.List(opts)
}))
if r.WatchListPageSize != 0 {
pager.PageSize = r.WatchListPageSize
}
// Pager falls back to full list if paginated list calls fail due to an "Expired" error.
list, err = pager.List(context.Background(), options)
close(listCh)
}()
......listMetaInterface, err := meta.ListAccessor(list)
if err != nil {
return fmt.Errorf("%s: Unable to understand list result %#v: %v", r.name, list, err)
}
// 获取资源的版本号信息
resourceVersion = listMetaInterface.GetResourceVersion()
initTrace.Step("Resource version extracted")
// 将资源数据装成资源对象列表
items, err := meta.ExtractList(list)
if err != nil {
return fmt.Errorf("%s: Unable to understand list result %#v (%v)", r.name, list, err)
}
initTrace.Step("Objects extracted")
// 将资源对象列表中的资源对象和资源版本存储只DeltaFIFO中
if err := r.syncWith(items, resourceVersion);
err != nil {
return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
}
initTrace.Step("SyncWith done")
// 设置最新的资源版本号
r.setLastSyncResourceVersion(resourceVersion)
initTrace.Step("Resource version updated")
return nil
}();
err != nil {
return err
}resyncerrc := make(chan error, 1)
cancelCh := make(chan struct{})
defer close(cancelCh)
go func() {
resyncCh, cleanup := r.resyncChan()
defer func() {
cleanup() // Call the last one written into cleanup
}()
for {
select {
case <-resyncCh:
case <-stopCh:
return
case <-cancelCh:
return
}
if r.ShouldResync == nil || r.ShouldResync() {
klog.V(4).Infof("%s: forcing resync", r.name)
if err := r.store.Resync();
err != nil {
resyncerrc <- err
return
}
}
cleanup()
resyncCh, cleanup = r.resyncChan()
}
}()for {
// give the stopCh a chance to stop the loop, even in case of continue statements further down on errors
select {
case <-stopCh:
return nil
default:
}timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
// 监控参数，会传入资源版本号信息
options = metav1.ListOptions{
ResourceVersion: resourceVersion,
// We want to avoid situations of hanging watchers. Stop any wachers that do not
// receive any events within the timeout window.
TimeoutSeconds: &timeoutSeconds,
// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
// Reflector doesn't assume bookmarks are returned at all (if the server do not support
// watch bookmarks, it will ignore this field).
AllowWatchBookmarks: true,
}// start the clock before sending the request, since some proxies won't flush headers until after the first watch event is sent
start := r.clock.Now()
// 调用Watch方法，监控资源对象
w, err := r.listerWatcher.Watch(options)
if err != nil {
switch err {
case io.EOF:
// watch closed normally
case io.ErrUnexpectedEOF:
klog.V(1).Infof("%s: Watch for %v closed with unexpected EOF: %v", r.name, r.expectedTypeName, err)
default:
utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.expectedTypeName, err))
}
// If this is "connection refused" error, it means that most likely apiserver is not responsive.
// It doesn't make sense to re-list all objects because most likely we will be able to restart
// watch where we ended.
// If that's the case wait and resend watch request.
if utilnet.IsConnectionRefused(err) {
time.Sleep(time.Second)
continue
}
return nil
}// 处理资源的变更事件
if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh);
err != nil {
if err != errorStopRequested {
switch {
case apierrs.IsResourceExpired(err):
klog.V(4).Infof("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err)
default:
klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err)
}
}
return nil
}
}
}DeltaFIFO ? DeltaFIFO可以分开理解为FIFO和Delta。 FIFO是一个先进先出队列，拥有队列的基本操作方法。Delta是资源对象存储，可以报错资源对象的操作类型。DeltaFIFO队列中，Reflector是生长泽，controller是消费者。DeltaFIFO结构如下：
#源码路径 vendor\k8s.io\client-go\tools\cache\delta_fifo.go
type DeltaFIFO struct {
// lock/cond protects access to 'items' and 'queue'.
lock sync.RWMutex
cond sync.Cond// We depend on the property that items in the set are in
// the queue and vice versa, and that all Deltas in this
// map have at least one Delta.
items map[string]Deltas
queue []string...
}生产者方法
【kubernets|kubernetes源码剖析之client-go(二) Informer机制】?DeltaFIFO队列中的资源对象在Added、Updated、Delete等事件被调用时都调用了queueActionLocked方法，它是DeltaFIFO实现的关键。
#源码路径 vendor\k8s.io\client-go\tools\cache\delta_fifo.go## Add、Update方法
// Add inserts an item, and puts it in the queue. The item is only enqueued
// if it doesn't already exist in the set.
func (f *DeltaFIFO) Add(obj interface{}) error {
// 执行前先进行加锁
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
return f.queueActionLocked(Added, obj)
}// Update is just like Add, but makes an Updated Delta.
func (f *DeltaFIFO) Update(obj interface{}) error {
f.lock.Lock()
defer f.lock.Unlock()
f.populated = true
return f.queueActionLocked(Updated, obj)
}## queueActionLockedf方法
// queueActionLocked appends to the delta list for the object.
// Caller must lock first.
func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
// 计算出资源对象的key
id, err := f.KeyOf(obj)
if err != nil {
return KeyError{obj, err}
}
// 将actionType和资源对象构造成dELTA,添加到items中，并通过dedupDeltas中去重。
newDeltas := append(f.items[id], Delta{actionType, obj})
newDeltas = dedupDeltas(newDeltas)if len(newDeltas) > 0 {
if _, exists := f.items[id];
!exists {
f.queue = append(f.queue, id)
}
f.items[id] = newDeltas
// 通过cond.Broadcast通知所有消费者接触阻塞。
f.cond.Broadcast()
} else {
// We need to remove this from our map (extra items in the queue are
// ignored if they are not in the map).
delete(f.items, id)
}
return nil
}消费者
#源码路径 vendor\k8s.io\client-go\tools\cache\delta_fifo.go// Pop blocks until an item is added to the queue, and then returns it.If
// multiple items are ready, they are returned in the order in which they were
// added/updated. The item is removed from the queue (and the store) before it
// is returned, so if you don't successfully process it, you need to add it back
// with AddIfNotPresent().
// process function is called under lock, so it is safe update data structures
// in it that need to be in sync with the queue (e.g. knownKeys). The PopProcessFunc
// may return an instance of ErrRequeue with a nested error to indicate the current
// item should be requeued (equivalent to calling AddIfNotPresent under the lock).
//
// Pop returns a 'Deltas', which has a complete list of all the things
// that happened to the object (deltas) while it was sitting in the queue.
func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
f.lock.Lock()
defer f.lock.Unlock()
for {
for len(f.queue) == 0 {
// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
// When Close() is called, the f.closed is set and the condition is broadcasted.
// Which causes this loop to continue and return from the Pop().
if f.IsClosed() {
return nil, ErrFIFOClosed
}
// 当队列中没有数据时，通过f.cond.wait阻塞等待数据。只有接收到cond.Broadcast时才说明有数据被添加，接触当前阻塞状态。
f.cond.Wait()
}
// 如果队列不为空，去除队列头部的数据。
id := f.queue[0]
f.queue = f.queue[1:]if f.initialPopulationCount > 0 {
f.initialPopulationCount--
}
item, ok := f.items[id]
if !ok {
// Item may have been deleted subsequently.
continue
}
delete(f.items, id)
// 将数据传入process回调函数，由上层消费者进行处理
err := process(item)
if e, ok := err.(ErrRequeue);
ok {
// 如果回调出错，则将数据重新添加回队列中
f.addIfNotPresent(id, item)
err = e.Err
}
// Don't need to copyDeltas here, because we're transferring
// ownership to the caller.
return item, err
}
}推荐阅读
- Android事件传递源码分析
- federation--kubernetes集群联邦的实现
- Quartz|Quartz 源码解析(四) —— QuartzScheduler和Listener事件监听
- [源码解析]|[源码解析] NVIDIA HugeCTR，GPU版本参数服务器---(3)
- ffmpeg源码分析01（结构体）
- Java程序员阅读源码的小技巧，原来大牛都是这样读的，赶紧看看！
- Vue源码分析—响应式原理(二)
- SwiftUI|SwiftUI iOS 瀑布流组件之仿CollectionView不规则图文混合(教程含源码)
- java|java b2b2c shop 多用户商城系统源码- config 修改配置
- Spring源码解析_属性赋值