Trong hệ sinh thái Kubernetes, kube-controller-manager đóng vai trò là một vòng lặp điều khiển (control loop) liên tục. Nhiệm vụ chính của nó là theo dõi trạng thái tài nguyên thông qua kube-apiserver, so sánh trạng thái hiện tại (current state) với trạng thái mong muốn (desired state) và thực hiện các thay đổi cần thiết để đưa hai trạng thái này về mức đồng nhất.

Cấu trúc khởi tạo của kube-controller-manager

Tương tự như các thành phần khác trong Kubernetes như kube-scheduler, kube-controller-manager cũng sử dụng thư viện Cobra để xây dựng giao diện dòng lệnh. Quy trình khởi chạy bắt đầu từ việc thiết lập các tùy chọn và cấu hình cơ bản. Dưới đây là cấu trúc mã nguồn mô phỏng quá trình khởi tạo:

// cmd/kube-controller-manager/app/controllermanager.go
func NewControllerManagerCommand() *cobra.Command {
    opts, _ := options.NewKubeControllerManagerOptions()
    
    cmd := &cobra.Command{
        RunE: func(cmd *cobra.Command, args []string) error {
            // Khởi tạo cấu hình dựa trên các tham số đầu vào
            config, err := opts.Config(KnownControllers(), ControllersDisabledByDefault(), ControllerAliases())
            if err != nil {
                return err
            }
            
            // Thực thi hàm Run để bắt đầu vòng lặp điều khiển
            return Run(context.Background(), config.Complete())
        },
    }
    return cmd
}

Hàm Run là nơi logic chính bắt đầu thực hiện. Nó chịu trách nhiệm khởi tạo ControllerContext, kích hoạt các bộ điều khiển và khởi chạy các Informer để theo dõi tài nguyên.

func Run(ctx context.Context, cfg *config.CompletedConfig) error {
    runEngine := func(ctx context.Context, descriptors map[string]*ControllerDescriptor) {
        // Tạo context cho bộ điều khiển
        ctrlCtx, err := CreateControllerContext(logger, cfg, rootBuilder, clientBuilder, ctx.Done())
        if err != nil {
            klog.Fatalf("Failed to create controller context: %v", err)
        }

        // Kích hoạt toàn bộ các controller đã đăng ký
        if err := StartControllers(ctx, ctrlCtx, descriptors, mux, healthz); err != nil {
            klog.Fatalf("Error starting controllers: %v", err)
        }

        // Bắt đầu chạy các Informer để cache dữ liệu từ API Server
        ctrlCtx.InformerFactory.Start(ctx.Done())
        ctrlCtx.ObjectOrMetadataInformerFactory.Start(ctx.Done())
        
        <-ctx.Done()
    }

    // Nếu cấu hình Leader Election, engine sẽ chỉ chạy khi node là leader
    if !cfg.ComponentConfig.Generic.LeaderElection.LeaderElect {
        descriptors := NewControllerDescriptors()
        runEngine(ctx, descriptors)
        return nil
    }
    // Logic cho Leader Election được thực hiện tại đây...
    return nil
}

Đăng ký và kích hoạt Controller

Hàm NewControllerDescriptors đóng vai trò là một registry, nơi tất cả các bộ điều khiển tài nguyên (như Endpoints, Replication, PodGC) được đăng ký với một hàm khởi tạo (init function) tương ứng.

func NewControllerDescriptors() map[string]*ControllerDescriptor {
    registry := make(map[string]*ControllerDescriptor)
    
    add := func(desc *ControllerDescriptor) {
        registry[desc.name] = desc
    }

    // Đăng ký các bộ điều khiển mặc định
    add(newReplicaSetControllerDescriptor())
    add(newEndpointsControllerDescriptor())
    add(newPodGarbageCollectorControllerDescriptor())
    
    return registry
}

Khi StartControllers được gọi, nó sẽ duyệt qua danh sách các ControllerDescriptor và thực thi hàm initFunc của từng bộ điều khiển để bắt đầu tiến trình xử lý riêng biệt của chúng.

Phân tích bộ điều khiển ReplicaSet

Lấy ReplicaSet làm ví dụ, khi được khởi tạo, nó sẽ tạo ra một ReplicaSetController. Đối tượng này chứa các Informer để lắng nghe thay đổi từ cả ReplicaSet và Pod.

// pkg/controller/replicaset/replica_set.go
func NewBaseController(logger klog.Logger, rsInformer appsinformers.ReplicaSetInformer, podInformer coreinformers.PodInformer, ...) *ReplicaSetController {
    rsc := &ReplicaSetController{
        // Khởi tạo hàng đợi công việc (Work Queue)
        queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "replicaset"),
    }

    // Lắng nghe sự kiện từ ReplicaSet
    rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    rsc.addRS,
        UpdateFunc: rsc.updateRS,
        DeleteFunc: rsc.deleteRS,
    })

    // Lắng nghe sự kiện từ Pod để kiểm tra tính toàn vẹn của ReplicaSet
    podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    rsc.addPod,
        UpdateFunc: rsc.updatePod,
        DeleteFunc: rsc.deletePod,
    })

    rsc.syncHandler = rsc.syncReplicaSet
    return rsc
}

Sau khi khởi tạo, hàm Run của bộ điều khiển sẽ bắt đầu các worker để xử lý hàng đợi:

func (rsc *ReplicaSetController) Run(ctx context.Context, workers int) {
    // Chờ cache đồng bộ hoàn toàn
    if !cache.WaitForNamedCacheSync("ReplicaSet", ctx.Done(), rsc.podListerSynced, rsc.rsListerSynced) {
        return
    }

    for i := 0; i < workers; i++ {
        go wait.UntilWithContext(ctx, rsc.worker, time.Second)
    }
    <-ctx.Done()
}

Quy trình xử lý khi một Pod bị xóa

Để hiểu cách kube-controller-manager duy trì trạng thái, hãy xem xét kịch bản một Pod thuộc ReplicaSet bị xóa thủ công: 1. **Sự kiện xóa Pod**: podInformer nhận thấy sự thay đổi và kích hoạt deletePod. 2. **Đưa vào hàng đợi**: Hàm deletePod xác định ReplicaSet sở hữu Pod đó và đẩy key của ReplicaSet vào hàng đợi workqueue. 3. **Xử lý (Sync)**: Worker lấy key ra và gọi syncReplicaSet. Tại đây, bộ điều khiển sẽ liệt kê tất cả các Pod hiện có liên quan đến ReplicaSet này. 4. **So sánh trạng thái**: Nếu số lượng Pod thực tế nhỏ hơn spec.replicas, hàm manageReplicas sẽ được thực thi. 5. **Khôi phục**: manageReplicas gọi API CreatePods để tạo Pod mới nhằm bù đắp lượng thiếu hụt.

func (rsc *ReplicaSetController) manageReplicas(ctx context.Context, currentPods []*v1.Pod, rs *apps.ReplicaSet) error {
    gap := len(currentPods) - int(*(rs.Spec.Replicas))
    
    if gap < 0 {
        // Thiếu Pod: Thực hiện tạo mới theo lô (batch)
        _, err := slowStartBatch(abs(gap), ..., func() error {
            return rsc.podControl.CreatePods(ctx, rs.Namespace, &rs.Spec.Template, rs, ...)
        })
        return err
    }
    return nil
}

Cuối cùng, updateReplicaSetStatus sẽ được gọi để cập nhật số lượng replica hiện tại vào status của ReplicaSet trên API Server. Quá trình này lặp đi lặp lại liên tục, đảm bảo tính tự phục hồi (self-healing) của hệ thống Kubernetes.