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 start() {
  	scheme := runtime.NewScheme()
  	_ = corev1.AddToScheme(scheme)
  	// 1. init Manager	初始化 Manager，同时生成一个默认配置的 Cache
  	mgr, _ := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
    	Scheme: scheme,
    	Port:   9443,
  	})
  	// 2. init Reconciler（Controller）
  	_ = ctrl.NewControllerManagedBy(mgr).	// 使用了builder 模式
    	For(&corev1.Pod{}).				// 指定 watch 的资源类型
		// .Owns()						// 表示某资源是我关心资源的从属，其 event 也会进去 Controller 的队列中；
    	Complete(&ApplicationReconciler{})	// 将用户的 Reconciler 注册进 Controller，并生成 watch 资源的默认 eventHandler，同时执行 Controller 的 watch 函数；
  	// 3. start Manager
  	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
  	}
}
type ApplicationReconciler struct {
}
func (a ApplicationReconciler) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
  	return reconcile.Result{}, nil
}

func start() {
  scheme := runtime.NewScheme()
  _ = corev1.AddToScheme(scheme)
  // 1. init Manager
  mgr, _ := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
    Scheme: scheme,
    Port:   9443,
  })
  // 2. init Reconciler（Controller）
  c, _ := controller.New("app", mgr, controller.Options{
	Reconciler: &ApplicationReconciler{},
  })
  // 监控Pod变动并将key写入workqueue
  _ = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForObject{}, predicate.Funcs{
    CreateFunc: func(event event.CreateEvent) bool {...},
    UpdateFunc: func(updateEvent event.UpdateEvent) bool {...},
    DeleteFunc: func(deleteEvent event.DeleteEvent) bool {...},
  })
  // 3. start Manager
  if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
  }
}
type ApplicationReconciler struct {
	...
}
func (a ApplicationReconciler) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
  	// 初始化log
	log := log.FromContext(ctx)
	// 从缓存中获取Pod
	pod := &corev1.Pod{}
	err := r.client.Get(ctx, request.NamespacedName, rs)	// 使用controller-runtime的Client时，我们只需要提供资源名（Namespace/Name）、资源类型（结构体指针），即可开始CRUD操作。
	if errors.IsNotFound(err) {
		log.Error(nil, "Could not find Pod")
		return reconcile.Result{}, nil
	}
	if err != nil {
		return reconcile.Result{}, fmt.Errorf("could not fetch Pod: %+v", err)
	}
	// 打印Pod
	log.Info("Reconciling Pod", "pod name", pod.Name)
	return reconcile.Result{}, 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)
}

// 这里明确了 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...)
}

// Kind 用于提供来自集群内部的事件源，这些事件来自于 Watches（例如 Pod Create 事件）
type Kind struct {
 	// Type 是 watch 对象的类型，比如 &v1.Pod{}
 	Type runtime.Object
 	// cache 用于 watch 的 APIs 接口
 	cache cache.Cache
}
func (ks *Kind) Start(handler handler.EventHandler, queue workqueue.RateLimitingInterface,prct ...predicate.Predicate) error {
 	// 从 Cache 中获取 Informer 并添加一个事件处理程序来添加队列
 	i, err := ks.cache.GetInformer(context.TODO(), ks.Type)
	i.AddEventHandler(internal.EventHandler{Queue: queue, EventHandler: handler, Predicates: prct})
 	return nil
}

// 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
}

type ApplicationReconciler struct {
	client.Client
	Scheme *runtime.Scheme
    Log    logr.Logger
}
type Reconciler interface {
	Reconcile(Request) (Result, error)
}
type Request struct {
	types.NamespacedName
}
type NamespacedName struct {
	Namespace string
	Name      string
}
type Result struct {
	Requeue bool
	RequeueAfter time.Duration
}

// 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
}