kubernetes的Controller-manager的工作过程

run := func(stop <-chan struct{}) {
	rootClientBuilder := controller.SimpleControllerClientBuilder{
		ClientConfig: kubeconfig,
	}
	var clientBuilder controller.ControllerClientBuilder
	if len(s.ServiceAccountKeyFile) > 0 && s.UseServiceAccountCredentials {
		clientBuilder = controller.SAControllerClientBuilder{
			ClientConfig:         restclient.AnonymousClientConfig(kubeconfig),
			CoreClient:           kubeClient.Core(),
			AuthenticationClient: kubeClient.Authentication(),
			Namespace:            "kube-system",
		}
	} else {
		clientBuilder = rootClientBuilder
	}
	
	err := StartControllers(newControllerInitializers(), s, rootClientBuilder, clientBuilder, stop)
	glog.Fatalf("error running controllers: %v", err)
	panic("unreachable")
}

func newControllerInitializers() map[string]InitFunc {
	controllers := map[string]InitFunc{}
	controllers["endpoint"] = startEndpointController
	controllers["replicationcontroller"] = startReplicationController
	controllers["podgc"] = startPodGCController
	controllers["resourcequota"] = startResourceQuotaController
	controllers["namespace"] = startNamespaceController
	controllers["serviceaccount"] = startServiceAccountController
	controllers["garbagecollector"] = startGarbageCollectorController
	controllers["daemonset"] = startDaemonSetController
	controllers["job"] = startJobController
	controllers["deployment"] = startDeploymentController
	controllers["replicaset"] = startReplicaSetController
	controllers["horizontalpodautoscaling"] = startHPAController
	controllers["disruption"] = startDisruptionController
	controllers["statefuleset"] = startStatefulSetController
	controllers["cronjob"] = startCronJobController
	controllers["certificatesigningrequests"] = startCSRController
	controllers["ttl"] = startTTLController
	controllers["bootstrapsigner"] = startBootstrapSignerController
	controllers["tokencleaner"] = startTokenCleanerController
	
	return controllers
}

func StartControllers(controllers map[string]InitFunc, s *options.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) error {
	versionedClient := rootClientBuilder.ClientOrDie("shared-informers")
	sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())
	...
	ctx := ControllerContext{
		ClientBuilder:      clientBuilder,
		InformerFactory:    sharedInformers,
		Options:            *s,
		AvailableResources: availableResources,
		Stop:               stop,
	}

func startEndpointController(ctx ControllerContext) (bool, error) {
	go endpointcontroller.NewEndpointController(
		ctx.InformerFactory.Core().V1().Pods(),
		ctx.InformerFactory.Core().V1().Services(),
		ctx.ClientBuilder.ClientOrDie("endpoint-controller"),
	).Run(int(ctx.Options.ConcurrentEndpointSyncs), ctx.Stop)
	return true, nil
}

sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())

func (f *sharedInformerFactory) Core() core.Interface {
	return core.New(f)
}

func New(f internalinterfaces.SharedInformerFactory) Interface {
	return &group{f}
}

type group struct {
	internalinterfaces.SharedInformerFactory
}

func (g *group) V1() v1.Interface {
	return v1.New(g.SharedInformerFactory)
}

func New(f internalinterfaces.SharedInformerFactory) Interface {
	return &version{f}
}

type Interface interface {
	ComponentStatuses() ComponentStatusInformer
	ConfigMaps() ConfigMapInformer
	Endpoints() EndpointsInformer
	Events() EventInformer
	LimitRanges() LimitRangeInformer
	Pods() PodInformer
	...

func (v *version) Pods() PodInformer {
	return &podInformer{factory: v.SharedInformerFactory}
}

type podInformer struct {
	factory internalinterfaces.SharedInformerFactory
}

--podInformer : struct
    [fields]
   -factory : internalinterfaces.SharedInformerFactory
    [methods]
   +Informer() : cache.SharedIndexInformer
   +Lister() : v1.PodLister

func (f *podInformer) Lister() v1.PodLister {
	return v1.NewPodLister(f.Informer().GetIndexer())
}

func (f *podInformer) Informer() cache.SharedIndexInformer {
	return f.factory.InformerFor(&api_v1.Pod{}, newPodInformer)
}

func (f *sharedInformerFactory) InformerFor(obj runtime.Object, newFunc internalinterfaces.NewInformerFunc) cache.SharedIndexInformer {
	f.lock.Lock()
	defer f.lock.Unlock()

	informerType := reflect.TypeOf(obj)
	informer, exists := f.informers[informerType]
	if exists {
		return informer
	}
	informer = newFunc(f.client, f.defaultResync)
	f.informers[informerType] = informer

	return informer
}

func newPodInformer(client clientset.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer {
	sharedIndexInformer := cache.NewSharedIndexInformer(
		&cache.ListWatch{
			ListFunc: func(options meta_v1.ListOptions) (runtime.Object, error) {
				return client.CoreV1().Pods(meta_v1.NamespaceAll).List(options)
			},
			WatchFunc: func(options meta_v1.ListOptions) (watch.Interface, error) {
				return client.CoreV1().Pods(meta_v1.NamespaceAll).Watch(options)
			},
		},
		&api_v1.Pod{},
		resyncPeriod,
		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
	)
	
	return sharedIndexInformer
}

func startEndpointController(ctx ControllerContext) (bool, error) {
	go endpointcontroller.NewEndpointController(
		ctx.InformerFactory.Core().V1().Pods(),
		ctx.InformerFactory.Core().V1().Services(),
		ctx.ClientBuilder.ClientOrDie("endpoint-controller"),
	).Run(int(ctx.Options.ConcurrentEndpointSyncs), ctx.Stop)
	return true, nil
}

func NewEndpointController(podInformer coreinformers.PodInformer, serviceInformer coreinformers.ServiceInformer, client clientset.Interface) *EndpointController {
	if client != nil && client.Core().RESTClient().GetRateLimiter() != nil {
		metrics.RegisterMetricAndTrackRateLimiterUsage("endpoint_controller", client.Core().RESTClient().GetRateLimiter())
	}
	...
	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    e.addPod,
		UpdateFunc: e.updatePod,
		DeleteFunc: e.deletePod,
	})
	e.podLister = podInformer.Lister()
	e.podsSynced = podInformer.Informer().HasSynced
	...

func (e *EndpointController) addPod(obj interface{}) {
	pod := obj.(*v1.Pod)
	services, err := e.getPodServiceMemberships(pod)
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("Unable to get pod %v/%v's service memberships: %v", pod.Namespace, pod.Name, err))
		return
	}
	for key := range services {
		e.queue.Add(key)
	}
}

func (e *EndpointController) Run(workers int, stopCh <-chan struct{}) {
	...
	for i := 0; i < workers; i++ {
		go wait.Until(e.worker, time.Second, stopCh)
	}
	...
}

func (e *EndpointController) worker() {
	for e.processNextWorkItem() {
	}
}

func (e *EndpointController) processNextWorkItem() bool {
	eKey, quit := e.queue.Get()
	if quit {
		return false
	}
	defer e.queue.Done(eKey)

	err := e.syncService(eKey.(string))
	...
}

func (e *EndpointController) syncService(key string) error {
...
if createEndpoints {
	_, err = e.client.Core().Endpoints(service.Namespace).Create(newEndpoints)
} else {
	_, err = e.client.Core().Endpoints(service.Namespace).Update(newEndpoints)
}