《一起读 kubernetes 源码》揭秘 k8s 关键机制 informer

for {
    实际状态 := 获取集群中对象 X 的实际状态(Actual State)
    期望状态 := 获取集群中对象 X 的期望状态(Expectation State)
    if 实际状态 == 期望状态{
        什么都不做
    } else {
        执行编排动作，将实际状态调整为期望状态
    }
}

func NewController(pods informers.PodInformer) *Controller {
    c := &Controller{
        pods: pods.Lister(),
        podsSynced: pods.Informer().HasSynced,
        queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "controller-name"),
    }

    pods.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            // ...
        },
        UpdateFunc: func(old interface{}, new interface{}) {
            // ...
        },
        DeleteFunc: func(obj interface{}) {
            // ...
        },
    },)

    return c
}

// staging/src/k8s.io/client-go/tools/cache/controller.go:380
func NewIndexerInformer(
	lw ListerWatcher,
	objType runtime.Object,
	resyncPeriod time.Duration,
	h ResourceEventHandler,
	indexers Indexers,
) (Indexer, Controller) {
	// This will hold the client state, as we know it.
	clientState := NewIndexer(DeletionHandlingMetaNamespaceKeyFunc, indexers)

	return clientState, newInformer(lw, objType, resyncPeriod, h, clientState, nil)
}

// staging/src/k8s.io/client-go/tools/cache/controller.go:483
func newInformer(
	lw ListerWatcher,
	objType runtime.Object,
	resyncPeriod time.Duration,
	h ResourceEventHandler,
	clientState Store,
	transformer TransformFunc,
) Controller {
	// This will hold incoming changes. Note how we pass clientState in as a
	// KeyLister, that way resync operations will result in the correct set
	// of update/delete deltas.
	fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
		KnownObjects:          clientState,
		EmitDeltaTypeReplaced: true,
		Transformer:           transformer,
	})

	cfg := &Config{
		Queue:            fifo,
		ListerWatcher:    lw,
		ObjectType:       objType,
		FullResyncPeriod: resyncPeriod,
		RetryOnError:     false,

		Process: func(obj interface{}, isInInitialList bool) error {
			if deltas, ok := obj.(Deltas); ok {
				return processDeltas(h, clientState, deltas, isInInitialList)
			}
			return errors.New("object given as Process argument is not Deltas")
		},
	}
	return New(cfg)
}

// staging/src/k8s.io/client-go/tools/cache/controller.go:436
func processDeltas(
    handler ResourceEventHandler,
    clientState Store,
    deltas Deltas,
    isInInitialList bool,
) error {
    for _, d := range deltas {
        obj := d.Object

        switch d.Type {
        case Sync, Replaced, Added, Updated:
            if old, exists, err := clientState.Get(obj); err == nil && exists {
                clientState.Update(obj)
                handler.OnUpdate(old, obj)
            } else {
                clientState.Add(obj)
                handler.OnAdd(obj, isInInitialList)
            }
        case Deleted:
            clientState.Delete(obj)
            handler.OnDelete(obj)
        }
    }
    return nil
}

// staging/src/k8s.io/client-go/tools/cache/controller.go:186
func (c *controller) processLoop() {
	for {
		obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
		if err != nil {
			if err == ErrFIFOClosed {
				return
			}
			if c.config.RetryOnError {
				// This is the safe way to re-enqueue.
				c.config.Queue.AddIfNotPresent(obj)
			}
		}
	}
}

// staging/src/k8s.io/client-go/tools/cache/controller.go:129
func (c *controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	go func() {
		<-stopCh
		c.config.Queue.Close()
	}()
	r := NewReflectorWithOptions(
		c.config.ListerWatcher,
		c.config.ObjectType,
		c.config.Queue,
		ReflectorOptions{
			ResyncPeriod:    c.config.FullResyncPeriod,
			TypeDescription: c.config.ObjectDescription,
			Clock:           c.clock,
		},
	)
	r.ShouldResync = c.config.ShouldResync
	r.WatchListPageSize = c.config.WatchListPageSize
	if c.config.WatchErrorHandler != nil {
		r.watchErrorHandler = c.config.WatchErrorHandler
	}

	c.reflectorMutex.Lock()
	c.reflector = r
	c.reflectorMutex.Unlock()

	var wg wait.Group

	wg.StartWithChannel(stopCh, r.Run)

	wait.Until(c.processLoop, time.Second, stopCh)
	wg.Wait()
}

// staging/src/k8s.io/client-go/tools/cache/reflector.go:743
resourceVersion := meta.GetResourceVersion()
switch event.Type {
case watch.Added:
    err := store.Add(event.Object)
case watch.Modified:
    err := store.Update(event.Object)
case watch.Deleted:
    err := store.Delete(event.Object)
case watch.Bookmark:
    if _, ok := meta.GetAnnotations()["k8s.io/initial-events-end"]; ok {
        if exitOnInitialEventsEndBookmark != nil {
            *exitOnInitialEventsEndBookmark = true
        }
    }
default:
    utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", name, event))
}