Controller实现ReplicaSetController的示例分析

func NewControllerInitializers() map[string]InitFunc {

   controllers := map[string]InitFunc{}

   ....

   controllers["replicaset"] = startReplicaSetController

   ....

   return controllers

}

func startReplicaSetController(ctx ControllerContext) (bool, error) {

   if !ctx.AvailableResources[schema.GroupVersionResource{Group: "extensions", Version: "v1beta1", Resource: "replicasets"}] {

      return false, nil

   }

   // 获取监听ReplicaSets/Pods的Informer

   go replicaset.NewReplicaSetController(

      ctx.InformerFactory.Extensions().V1beta1().ReplicaSets(),

      ctx.InformerFactory.Core().V1().Pods(),

      ctx.ClientBuilder.ClientOrDie("replicaset-controller"),

      replicaset.BurstReplicas,

   ).Run(int(ctx.Options.ConcurrentRSSyncs), ctx.Stop)

   return true, nil

}

// NewReplicaSetController configures a replica set controller with the specified event recorder

func NewReplicaSetController(rsInformer extensionsinformers.ReplicaSetInformer, podInformer coreinformers.PodInformer, kubeClient clientset.Interface, burstReplicas int) *ReplicaSetController {

   ....

   // 初始化controller

   rsc := &ReplicaSetController{

      kubeClient: kubeClient,

      podControl: controller.RealPodControl{

         KubeClient: kubeClient,

         Recorder:   eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "replicaset-controller"}),

      },

      burstReplicas: burstReplicas,

      expectations:  controller.NewUIDTrackingControllerExpectations(controller.NewControllerExpectations()),

      queue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "replicaset"),

   }

   // 在replica set informer上注册回调handler

   rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{

      AddFunc:    rsc.enqueueReplicaSet,

      UpdateFunc: rsc.updateRS,

      // This will enter the sync loop and no-op, because the replica set has been deleted from the store.

      // Note that deleting a replica set immediately after scaling it to 0 will not work. The recommended

      // way of achieving this is by performing a `stop` operation on the replica set.

      DeleteFunc: rsc.enqueueReplicaSet,

   })

   rsc.rsLister = rsInformer.Lister()

   rsc.rsListerSynced = rsInformer.Informer().HasSynced

   // 在pod informer上注册回调

   podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{

      AddFunc: rsc.addPod,

      // This invokes the ReplicaSet for every pod change, eg: host assignment. Though this might seem like

      // overkill the most frequent pod update is status, and the associated ReplicaSet will only list from

      // local storage, so it should be ok.

      UpdateFunc: rsc.updatePod,

      DeleteFunc: rsc.deletePod,

   })

   rsc.podLister = podInformer.Lister()

   rsc.podListerSynced = podInformer.Informer().HasSynced

   // syncReplicaSet整个controller实际业务逻辑的入口，会被回调触发

   rsc.syncHandler = rsc.syncReplicaSet

   return rsc

}