for {
    actualState := GetResourceActualState(rsvc)
    expectState := GetResourceExpectState(rsvc) // 来自yaml 文件
    if actualState == expectState {
        // do nothing
    } else {
        Reconcile(rsvc) // 编排逻辑，调谐的最终结果一般是对被控制对象的某种写操作，比如增/删/改 Pod
    }
}

kubeClient = kubernetes.NewForConfigOrDie(opt.Config)
// 基于GVK 操作资源，假设需要操作数十种不同资源时，我们需要为每一种资源实现各自的函数
podInformer = informers.NewSharedInformerFactory(pod.kubeClient, 0).Core().V1().Pods()
podLister = pod.podInformer.Lister()
podSynced = pod.podInformer.Informer().HasSynced
podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
	AddFunc:    AddPod,
	DeleteFunc: DeletePod,
	UpdateFunc: UpdatePod,
})

// 启动informer
podInformer.Informer().Run(stopCh)
cache.WaitForCacheSync(stopCh, podSynced)
// 此处没有使用workqueue，但一般都是会用workqueue 增强处理逻辑的

func main(){
	// 初始化 Manager，同时生成一个默认配置的 Cache
 	mgr, err = ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{ 
        Scheme:             scheme,
        Namespace:          namespace,
    })
	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
        setupLog.Error(err, "problem running manager")
        os.Exit(1)
    }
}

type ApplicationReconciler struct {
	client.Client
	Scheme *runtime.Scheme
    Log    logr.Logger
}
func (r *ApplicationReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
	// get application
	application := &xx.Application{}
	if err := r.client.Get(ctx, req.NamespacedName, application); err != nil {...}

    // your logic here

    return ctrl.Result{}, nil
}
// 将 Reconcile 添加到 manager 中，这样当 manager 启动时它就会被启动。
func (r *ApplicationReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&xx.Application{}).
        Complete(r)
}

func SetupWithManager(mgr ctrl.Manager, cfg *Config, 其它参数) error {
	r := &applicationReconciler{
		client:        mgr.GetClient(),
		cfg:           cfg,
		log:           ctrl.LoggerFrom(context.Background()).WithName(name),
		recorder:      mgr.GetEventRecorderFor(name),
	}
	return ctrl.NewControllerManagedBy(mgr).
        For(&xx.Application{}).
        Complete(r)
}

func main() {
  	scheme := runtime.NewScheme()
  	_ = corev1.AddToScheme(scheme)
	// 1. init Manager
  	mgr, _ := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
    	Scheme: scheme,
    	Port:   9443,
  	})
  	// 2. init Reconciler
  	application.SetupWithManager(mgr)
  	// 3. start Manager
  	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
  	}
}

func main() {
	...
	// manager -> controller -> reconciler 的对象层级结构
	// 主体分为两部分，一是配置manager，二是启动manager
    // 1. init Manager 
	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme:             scheme,		// 要将你监听的crd 加入到scheme 中
		Port:               9443,})
    // 2. init Reconciler（Controller）
	c := &controllers.ApplicationReconciler{
		Client: mgr.GetClient(),
		Scheme: mgr.GetScheme(),
	}
	if err = c.SetupWithManager(mgr); err != nil {...}
    // 3. start Manager
	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {...}
}

// kubebuilder 生成
func (r *ApplicationReconciler) SetupWithManager(mgr ctrl.Manager) error {
	c, err := controller.New(r.ControllerName(), mgr, controller.Options{
		Reconciler: r,
	})
	// 为Controller指定cr
	c.Watch(&source.Kind{Type: &appsv1alpha1.Application{}}, &handler.EnqueueRequestForObject{},
		predicate.Funcs{CreateFunc: r.onOwnerCreateFunc()},
	)
	return nil
}
func New(name string, mgr manager.Manager, options Options) (Controller, error) {
	...
	// Inject dependencies into Reconciler
	if err := mgr.SetFields(options.Reconciler); err != nil {...}
	c := &controller.Controller{    // Create controller with dependencies set
		Do:       options.Reconciler,
		Cache:    mgr.GetCache(),
		Config:   mgr.GetConfig(),
		Scheme:   mgr.GetScheme(),
		Client:   mgr.GetClient(),
        Name:     name,
	}
	return c, mgr.Add(c)    // Add the controller as a Manager components
}

// Manager 初始化共享的依赖关系，比如 Caches 和 Client，并将他们提供给 Runnables
type Manager interface {
 	// Add 将在组件上设置所需的依赖关系，并在调用 Start 时启动组件
  	// Add 将注入接口的依赖关系 - 比如 注入 inject.Client
  	// 根据 Runnable 是否实现了 LeaderElectionRunnable 接口判断Runnable 可以在非 LeaderElection 模式（始终运行）或 LeaderElection 模式（如果启用了 LeaderElection，则由 LeaderElection 管理）下运行
  	Add(Runnable) error
 	// SetFields 设置对象上的所有依赖关系，而该对象已经实现了 inject 接口
  	// 比如 inject.Client
 	SetFields(interface{}) error
 	// Start 启动所有已注册的控制器，并一直运行，直到停止通道关闭
  	// 如果使用了 LeaderElection，则必须在此返回后立即退出二进制，否则需要 Leader 选举的组件可能会在 Leader 锁丢失后继续运行
 	Start(<-chan struct{}) error
 	...
}

func (cm *controllerManager) Start(stop <-chan struct{}) error {
    // 启动metric 组件供Prometheus 拉取数据
    go cm.serveMetrics(cm.internalStop)
    // 启动健康检查探针
	go cm.serveHealthProbes(cm.internalStop)
	go cm.startNonLeaderElectionRunnables()
	if cm.resourceLock != nil {
		if err := cm.startLeaderElection(); err != nil{...}
	} else {
		go cm.startLeaderElectionRunnables()
	}
    ...
}
type controllerManager struct {
    ...
	// leaderElectionRunnables is the set of Controllers that the controllerManager injects deps into and Starts.
	// These Runnables are managed by lead election.
	leaderElectionRunnables []Runnable
}
// 启动cache/informer 及 Controller
func (cm *controllerManager) startLeaderElectionRunnables() {
    // 核心是启动Informer, waitForCache ==> cm.startCache = cm.cache.Start ==> InformersMap.Start ==> 
    // InformersMap.structured/unstructured.Start ==> Informer.Run
	cm.waitForCache()
	for _, c := range cm.leaderElectionRunnables {
		ctrl := c
		go func() {
            // 启动Controller
			if err := ctrl.Start(cm.internalStop); err != nil {...}
		}()
	}
}

type Controller interface {
	// Reconciler is called to reconcile an object by Namespace/Name
	reconcile.Reconciler
	// Watch takes events provided by a Source and uses the EventHandler to enqueue reconcile.Requests in response to the events.
	Watch(src source.Source, eventhandler handler.EventHandler, predicates ...predicate.Predicate) error
	// Start starts the controller.  Start blocks until the context is closed or a controller has an error starting.
	Start(ctx context.Context) error
	...
}
// sigs.k8s.io/controller-runtime/pkg/controller/controller.go
type Controller struct {
	Name string // Name is used to uniquely identify a Controller in tracing, logging and monitoring. Name is required.
	Do reconcile.Reconciler
	Client client.Client // Client is a lazily initialized Client.  
	Scheme *runtime.Scheme
	Cache cache.Cache
	Config *rest.Config // Config is the rest.Config used to talk to the apiserver.  
	// Queue is an listeningQueue that listens for events from Informers and adds object keys to the Queue for processing
	Queue workqueue.RateLimitingInterface
    ...
}
func New(name string, mgr manager.Manager, options Options) (Controller, error) {
	c, err := NewUnmanaged(name, mgr, options)
	// Add the controller as a Manager components
	return c, mgr.Add(c)
}

type Controller struct {
	// startWatches maintains a list of sources, handlers, and predicates to start when the controller is started.
   	startWatches []watchDescription
}

// 这里明确了 Controller 监听哪些Type， 或者说哪些 event 会触发Controller
func (r *TFJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
	c, err := controller.New(r.ControllerName(), mgr, controller.Options{Reconciler: r,})
	// using onOwnerCreateFunc is easier to set defaults
	c.Watch(&source.Kind{Type: &tfv1.TFJob{}}, &handler.EnqueueRequestForObject{},predicate.Funcs{...}) 
	// inject watching for job related pod
	c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{...}, predicate.Funcs{...)
	// inject watching for job related service
	...
	return nil
}
// watch 就是启动 source.start
func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prct ...predicate.Predicate) error {
	// Inject Cache into arguments
	c.SetFields(src)
	c.SetFields(evthdler)
	for _, pr := range prct {
		c.SetFields(pr)
	}
	c.Log.Info("Starting EventSource", "source", src)
	return src.Start(c.ctx, evthdler, c.Queue, prct...)
}

// sigs.k8s.io/controller-runtime/pkg/internal/controller/controller.go
func (c *Controller) Start(stop <-chan struct{}) error {
	err := func() error {
        ...
		// Launch workers to process resources
		for i := 0; i < c.MaxConcurrentReconciles; i++ {
			// Process work items
			go wait.Until(c.worker, c.JitterPeriod, stop)
		}
		return nil
	}()
	return nil
}
func (c *Controller) worker() {
	for c.processNextWorkItem() {
	}
}
// 从队列中取出 变更的对象（也就是需要处理的对象），包括队列操作相关的线速、重试等，并触发Reconcile 逻辑
func (c *Controller) processNextWorkItem() bool {
	obj, shutdown := c.Queue.Get()
	if shutdown {
		return false// Stop working
	}
	defer c.Queue.Done(obj)
	return c.reconcileHandler(obj)
}
func (c *Controller) reconcileHandler(obj interface{}) bool {
	if req, ok = obj.(reconcile.Request); !ok {...}
	// RunInformersAndControllers the syncHandler, passing it the namespace/Name string of the resource to be synced.
	if result, err := c.Do.Reconcile(req); err != nil {
		c.Queue.AddRateLimited(req)
		return false
	} else if result.RequeueAfter > 0 {
		c.Queue.Forget(obj)
		c.Queue.AddAfter(req, result.RequeueAfter)
		return true
	} else if result.Requeue {
		c.Queue.AddRateLimited(req)
		return true
	}
	c.Queue.Forget(obj)
	return true
}

// controller-runtime/pkg/runtime/inject/inject.go
type Cache interface {	// 很像spring 中的XXAware
	InjectCache(cache cache.Cache) error
}
func CacheInto(c cache.Cache, i interface{}) (bool, error) {
	if s, ok := i.(Cache); ok {
		return true, s.InjectCache(c)
	}
	return false, nil
}
type Config interface {
	InjectConfig(*rest.Config) error
}
type Client interface {
	InjectClient(client.Client) error
}
type Scheme interface {
	InjectScheme(scheme *runtime.Scheme) error
}
...

func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prct ...predicate.Predicate) error {
	// Inject Cache into arguments   
	if err := c.SetFields(src); err != nil {...}
	if err := c.SetFields(evthdler); err != nil {...}
	for _, pr := range prct {
		if err := c.SetFields(pr); err != nil {...}
    }
    ...
}
func (cm *controllerManager) SetFields(i interface{}) error {
	if _, err := inject.ConfigInto(cm.config, i); err != nil {return err}
	if _, err := inject.ClientInto(cm.client, i); err != nil {return err}
	if _, err := inject.APIReaderInto(cm.apiReader, i); err != nil {return err}
	if _, err := inject.SchemeInto(cm.scheme, i); err != nil {return err}
	if _, err := inject.CacheInto(cm.cache, i); err != nil {return err}
	if _, err := inject.InjectorInto(cm.SetFields, i); err != nil {return err}
	if _, err := inject.StopChannelInto(cm.internalStop, i); err != nil {return err}
	if _, err := inject.MapperInto(cm.mapper, i); err != nil {return err}
	return nil
}
// ReplicaSetReconciler 实现了 Client 接口
type ReplicaSetReconciler struct {
	client.Client
}
func (a *ReplicaSetReconciler) InjectClient(c client.Client) error {
	a.Client = c
	return nil
}

// 从B 中获取A得名字
aFn := handler.MapFunc(
	func(object client.Object) []reconcile.Request {
		b := object.(*v1alpha1.B)
		requests := append(requests, reconcile.Request{
				NamespacedName: types.NamespacedName{
					Namespace: object.GetNamespace(),
					Name:      b.Annotations["aName"],
				},
			})
		return requests
	})
_, err := ctrl.NewControllerManagedBy(mgr).
	WithOptions(controller.Options{
		CacheSyncTimeout:        controllerCtx.Config.ControllerConfig.SyncTimeout.Duration,
		MaxConcurrentReconciles: controllerCtx.Config.ControllerConfig.Concurrency,
	}).
	For(&v1alpha1.A{}).
	// 当B 变更时，也触发A
	Watches(&source.Kind{Type: &v1alpha1.B{}}, handler.EnqueueRequestsFromMapFunc(aFn)).
	Build(r)