pkg
├── builder
├── cache
├── client  # client用于操作k8s的对象
├── cluster
├── config
├── controller  # controller逻辑
├── envtest
├── event
├── finalizer
├── handler
├── internal  # 核心代码 controller的具体实现
├── leaderelection
├── manager   # 核心代码
├── predicate
├── ratelimiter
├── reconcile
├── recorder
├── scheme
├── source
└── webhook

  // 构建manager对象，主要的初始化参数包括
	// - kubeconfig
	// - controller的option参数
	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme:                 scheme,
        Metrics:                metricsserver.Options{BindAddress: opt.MetricsAddr},
        HealthProbeBindAddress: opt.ProbeAddr,
        LeaderElection:         opt.EnableLeaderElection,
        LeaderElectionID:       "52609143.huweihuang.com",
        Controller: config.Controller{
            MaxConcurrentReconciles: opt.MaxConcurrentReconciles,
        },
    })

	// 将controller注册到manager中，并初始化controller对象。
    if err = (&venuscontroller.RedisReconciler{
        Client: mgr.GetClient(),
        Scheme: mgr.GetScheme(),
    }).SetupWithManager(mgr); err != nil {
        setupLog.Error(err, "unable to create controller", "controller", "Redis")
        return err
    }

	// 运行manager对象，从而运行controller中的reconcile逻辑。
    if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
        setupLog.Error(err, "problem running manager")
        return err
    }

type Manager interface {
	// Cluster holds a variety of methods to interact with a cluster.
	cluster.Cluster

	// Add will set requested dependencies on the component, and cause the component to be
	// started when Start is called.
	// Depending on if a Runnable implements LeaderElectionRunnable interface, a Runnable can be run in either
	// non-leaderelection mode (always running) or leader election mode (managed by leader election if enabled).
  
  // 通过Runnable接口将具体的controller注册到manager中。
	Add(Runnable) error

	// Start starts all registered Controllers and blocks until the context is cancelled.
	// Returns an error if there is an error starting any controller.
	//
	// If LeaderElection is used, the binary must be exited immediately after this returns,
	// otherwise components that need leader election might continue to run after the leader
	// lock was lost.
  
  // 运行具体的逻辑
	Start(ctx context.Context) error
}

func New(config *rest.Config, options Options) (Manager, error) {
    // Set default values for options fields
    options = setOptionsDefaults(options)
    ... 

    errChan := make(chan error)
    runnables := newRunnables(options.BaseContext, errChan)
    return &controllerManager{
        stopProcedureEngaged:          pointer.Int64(0),
        cluster:                       cluster,
        runnables:                     runnables,
        errChan:                       errChan,
        recorderProvider:              recorderProvider,
        resourceLock:                  resourceLock,
        metricsServer:                 metricsServer,
        controllerConfig:              options.Controller,
        logger:                        options.Logger,
        elected:                       make(chan struct{}),
        webhookServer:                 options.WebhookServer,
        leaderElectionID:              options.LeaderElectionID,
        leaseDuration:                 *options.LeaseDuration,
        renewDeadline:                 *options.RenewDeadline,
        retryPeriod:                   *options.RetryPeriod,
        healthProbeListener:           healthProbeListener,
        readinessEndpointName:         options.ReadinessEndpointName,
        livenessEndpointName:          options.LivenessEndpointName,
        pprofListener:                 pprofListener,
        gracefulShutdownTimeout:       *options.GracefulShutdownTimeout,
        internalProceduresStop:        make(chan struct{}),
        leaderElectionStopped:         make(chan struct{}),
        leaderElectionReleaseOnCancel: options.LeaderElectionReleaseOnCancel,
    }, nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *RedisReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&venusv1.Redis{}).
        Complete(r)
}

// Builder builds a Controller.
type Builder struct {
    forInput         ForInput
    ownsInput        []OwnsInput
    watchesInput     []WatchesInput
    mgr              manager.Manager
    globalPredicates []predicate.Predicate
    ctrl             controller.Controller
    ctrlOptions      controller.Options
    name             string
}

// ControllerManagedBy returns a new controller builder that will be started by the provided Manager.
func ControllerManagedBy(m manager.Manager) *Builder {
    return &Builder{mgr: m}
}

// Build builds the Application Controller and returns the Controller it created.
func (blder *Builder) Build(r reconcile.Reconciler) (controller.Controller, error) {

    // Set the ControllerManagedBy
    if err := blder.doController(r); err != nil {  // 初始化controller
        return nil, err
    }

    // Set the Watch
    if err := blder.doWatch(); err != nil {   // 添加event handler
        return nil, err
    }

    return blder.ctrl, nil
}

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

// NewUnmanaged returns a new controller without adding it to the manager. The
// caller is responsible for starting the returned controller.
func NewUnmanaged(name string, mgr manager.Manager, options Options) (Controller, error) {
    if options.Reconciler == nil {
        return nil, fmt.Errorf("must specify Reconciler")
    }
    ...
    // Create controller with dependencies set
    return &controller.Controller{
        Do: options.Reconciler,   // 将具体的reconciler函数传递到controller的reconciler。
        MakeQueue: func() workqueue.RateLimitingInterface {
            return workqueue.NewRateLimitingQueueWithConfig(options.RateLimiter, workqueue.RateLimitingQueueConfig{
                Name: name,
            })
        }, // 初始化任务队列
        MaxConcurrentReconciles: options.MaxConcurrentReconciles, // 设置controller的并发数
        CacheSyncTimeout:        options.CacheSyncTimeout,
        Name:                    name,
        LogConstructor:          options.LogConstructor,
        RecoverPanic:            options.RecoverPanic,
        LeaderElected:           options.NeedLeaderElection,
    }, nil

// Watch implements controller.Controller.
func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prct ...predicate.Predicate) error {
    c.mu.Lock()
    defer c.mu.Unlock()

    // Controller hasn't started yet, store the watches locally and return.
    //
    // These watches are going to be held on the controller struct until the manager or user calls Start(...).
    if !c.Started {
        c.startWatches = append(c.startWatches, watchDescription{src: src, handler: evthdler, predicates: prct})
        return nil
    }

    c.LogConstructor(nil).Info("Starting EventSource", "source", src)
    return src.Start(c.ctx, evthdler, c.Queue, prct...)
}

func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface,
    prct ...predicate.Predicate) error {
    // Informer should have been specified by the user.
    if is.Informer == nil {
        return fmt.Errorf("must specify Informer.Informer")
    }

    _, err := is.Informer.AddEventHandler(internal.NewEventHandler(ctx, queue, handler, prct).HandlerFuncs())
    if err != nil {
        return err
    }
    return nil
}

func (cm *controllerManager) Start(ctx context.Context) (err error) {
	// Start and wait for caches.
	if err := cm.runnables.Caches.Start(cm.internalCtx); err != nil {
	}

	// Start the non-leaderelection Runnables after the cache has synced.
	if err := cm.runnables.Others.Start(cm.internalCtx); err != nil {
	}

	// Start the leader election and all required runnables.
	{
		ctx, cancel := context.WithCancel(context.Background())
		cm.leaderElectionCancel = cancel
		go func() {
			if cm.resourceLock != nil {
				if err := cm.startLeaderElection(ctx); err != nil {
					cm.errChan <- err
				}
			} else {
				// Treat not having leader election enabled the same as being elected.
				if err := cm.startLeaderElectionRunnables(); err != nil {
					cm.errChan <- err
				}
				close(cm.elected)
			}
		}()
	}
	...
}

func (c *Controller) Start(ctx context.Context) error {
  ...
	err := func() error {
		...
		for _, watch := range c.startWatches {
			syncingSource, ok := watch.src.(source.SyncingSource)
      // 同步list-watch中的cache数据。
			if err := func() error {
				if err := syncingSource.WaitForSync(sourceStartCtx); err != nil {
          ...
				}
			}
		}

    // 运行指定并发数的processNextWorkItem任务。
		// Launch workers to process resources
		c.LogConstructor(nil).Info("Starting workers", "worker count", c.MaxConcurrentReconciles)
		wg.Add(c.MaxConcurrentReconciles)
		for i := 0; i < c.MaxConcurrentReconciles; i++ {
			go func() {
				defer wg.Done()
				// Run a worker thread that just dequeues items, processes them, and marks them done.
				// It enforces that the reconcileHandler is never invoked concurrently with the same object.
				for c.processNextWorkItem(ctx) {
				}
			}()
		}
	...
}

func (c *Controller) processNextWorkItem(ctx context.Context) bool {
	obj, shutdown := c.Queue.Get()
	if shutdown {
		// Stop working
		return false
	}

	defer c.Queue.Done(obj)

	ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(1)
	defer ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(-1)

	c.reconcileHandler(ctx, obj)
	return true
}

func (c *Controller) reconcileHandler(ctx context.Context, obj interface{}) {
	// 获取k8s crd的具体对象
	req, ok := obj.(reconcile.Request)
	if !ok {
		// As the item in the workqueue is actually invalid, we call
		// Forget here else we'd go into a loop of attempting to
		// process a work item that is invalid.
		c.Queue.Forget(obj)
		c.LogConstructor(nil).Error(nil, "Queue item was not a Request", "type", fmt.Sprintf("%T", obj), "value", obj)
		// Return true, don't take a break
		return
	}

	// 调用用户定义的Reconcile函数，并对返回结果进行处理。
	result, err := c.Reconcile(ctx, req)
	switch {
	case err != nil:
    // 如果错误不为空，则重新入队，等待处理
		if errors.Is(err, reconcile.TerminalError(nil)) {
			ctrlmetrics.TerminalReconcileErrors.WithLabelValues(c.Name).Inc()
		} else {
			c.Queue.AddRateLimited(req)
		}
	case result.RequeueAfter > 0:
		// 如果指定了延迟入队，则做延迟入队处理
		c.Queue.Forget(obj)
		c.Queue.AddAfter(req, result.RequeueAfter)
	case result.Requeue:
		// 如果指定了马上入队，则做相应处理
		c.Queue.AddRateLimited(req)
	default:
    // 如果错误为空，表示reconcile成功，则移除队列的任务
		log.V(5).Info("Reconcile successful")
		// Finally, if no error occurs we Forget this item so it does not
		// get queued again until another change happens.
		c.Queue.Forget(obj)
	}
}