8. 最精简的 Controller

为什么先裸写

下一章我们会用 Kubebuilder，几行代码搞定一个 Controller。但直接上框架，你会变成"调用黑盒"——出问题不知道往哪查。

这一章我们不用任何框架，纯 client-go 写一个 Controller。代码会长一点，但每一行都能看清在干什么。理解了底层，再看 Kubebuilder 帮你省了什么，才心里有数。

打个比方：驾校先让你开手动挡，理解离合、油门、挡位的关系。之后再开自动挡，知道每个操作背后发生了什么。直接上自动挡，坡道溜车你都不知道为什么。

目标

写一个 Controller，管上一章定义的 MySQL CRD：

• 监听 MySQL 资源变化
• 看到新建的 MySQL，打印一条日志（真实场景这里会创建 StatefulSet、Service 等）
• 看到删除的 MySQL，清理关联资源

完整代码不到 200 行，分四块看。

第一块：定义资源类型

client-go 没有我们的 MySQL 类型，得自己定义：

// api/v1/types.go
package v1

import (
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// MySQLSpec 是用户写的期望状态
type MySQLSpec struct {
    Replicas int32  `json:"replicas"`
    Version  string `json:"version"`
    Storage  string `json:"storage"`
}

// MySQLStatus 是 Controller 写的实际状态
type MySQLStatus struct {
    Ready     bool   `json:"ready"`
    Phase     string `json:"phase"`
    Replicas  int32  `json:"replicas"`
}

// +kubebuilder:object:root=true
// +kubebuilder:object:subresource:status
type MySQL struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   MySQLSpec   `json:"spec,omitempty"`
    Status MySQLStatus `json:"status,omitempty"`
}

// +kubebuilder:object:root=true
type MySQLList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata,omitempty"`
    Items           []MySQL `json:"items"`
}

这两个结构体跟上一章 CRD 的 schema 一一对应——spec 里 replicas/version/storage，status 里 ready/phase/replicas。

TypeMeta 和 ObjectMeta 是 K8S 资源的公共字段：apiVersion、kind、name、namespace、labels 这些。嵌入它们，MySQL 就有了 K8S 资源的标准头。

那两个 +kubebuilder:object 注释先忽略——下一章 Kubebuilder 会用到，本章手动注册。

第二块：注册到 Scheme

// api/v1/register.go
package v1

import (
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/runtime/schema"
)

var GroupVersion = schema.GroupVersion{Group: "database.example.com", Version: "v1"}

var SchemeBuilder = &scheme.Builder{GroupVersion: GroupVersion}

func init() {
    SchemeBuilder.Register(&MySQL{}, &MySQLList{})
}

func AddToScheme(scheme *runtime.Scheme) error {
    return SchemeBuilder.AddToScheme(scheme)
}

前面 GVK/GVR 那章讲过 Scheme 是 GVK → Go 类型的映射表。这段代码就是建表——把 database.example.com/v1/MySQL 映射到 &MySQL{}，database.example.com/v1/MySQLList 映射到 &MySQLList{}。

AddToScheme 是个标准接口，后面 Controller 初始化时调它。

第三块：Controller 主体

这是核心，分几段看。

初始化

// controller.go
package main

import (
    "context"
    "fmt"
    "time"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/util/runtime"
    "k8s.io/apimachinery/pkg/util/wait"
    "k8s.io/client-go/kubernetes/scheme"
    "k8s.io/client-go/tools/cache"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/util/workqueue"
    "k8s.io/client-go/informers"

    mygroupv1 "yourmodule/api/v1"
)

type Controller struct {
    informer cache.SharedIndexInformer
    queue    workqueue.RateLimitingInterface
    indexer  cache.Indexer
}

Controller 持有三个东西：

• informer：监听 MySQL 资源变化
• queue：事件队列，去重 + 退避
• indexer：本地缓存，Reconcile 时从这里读

跟第 2 章画的图对应：

Informer ──→ Workqueue ──→ Reconcile
                ↑
            Indexer（缓存）

启动 Informer

func NewController(kubeconfig string) (*Controller, error) {
    // 1. 建 client
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        return nil, err
    }

    // 2. 注册 Scheme
    myScheme := scheme.Scheme
    if err := mygroupv1.AddToScheme(myScheme); err != nil {
        return nil, err
    }

    // 3. 建 Informer 工厂
    //    informers.NewSharedInformerFactory 会复用底层 Watch 连接，
    //    多个 Informer 共享一个 ListWatch，省 API Server 连接
    factory := informers.NewSharedInformerFactoryWithOptions(
        nil, 0, // 这里简化了，真实代码传 clientset
    )

    // 4. 拿 MySQL 的 Informer
    informer := factory.InformerFor(
        &mygroupv1.MySQL{},
        func(client cache.Getter, duration time.Duration) cache.SharedIndexInformer {
            return cache.NewSharedIndexInformer(
                &cache.ListWatch{
                    ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
                        return dynamicClient.Resource(gvr).List(ctx, opts)
                    },
                    WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
                        return dynamicClient.Resource(gvr).Watch(ctx, opts)
                    },
                },
                &mygroupv1.MySQL{},
                10*time.Hour,  // resync 间隔
                cache.Indexers{},
            )
        },
    )

    queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())

    // 5. 注册事件回调
    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            key, err := cache.MetaNamespaceKeyFunc(obj)
            if err == nil {
                queue.Add(key)
            }
        },
        UpdateFunc: func(_, new interface{}) {
            key, err := cache.MetaNamespaceKeyFunc(new)
            if err == nil {
                queue.Add(key)
            }
        },
        DeleteFunc: func(obj interface{}) {
            key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
            if err == nil {
                queue.Add(key)
            }
        },
    })

    return &Controller{
        informer: informer,
        queue:    queue,
        indexer:  informer.GetIndexer(),
    }, nil
}

简化了部分代码（真实代码要处理 dynamic client、GVR 解析等），但骨架就这样。

关键是第 5 步——三个回调 AddFunc、UpdateFunc、DeleteFunc。Informer 收到事件后调它们，它们干的事只有一件：算出 key（namespace/name），塞进 queue。

不打 API Server，不做业务逻辑，不调 reconcile——只 enqueue。这叫"事件处理要轻"，重活留给 Reconcile。

Reconcile

func (c *Controller) Reconcile(key string) error {
    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        return err
    }

    // 1. 从缓存读资源
    obj, exists, err := c.indexer.GetByKey(key)
    if err != nil {
        return err
    }

    if !exists {
        // 资源被删了
        fmt.Printf("MySQL %s/%s 被删除，清理关联资源\n", namespace, name)
        return c.cleanup(namespace, name)
    }

    mysql := obj.(*mygroupv1.MySQL)

    // 2. 跳过正在删除的资源（DeletionTimestamp 非空表示用户删了，还在等 finalizer）
    if !mysql.DeletionTimestamp.IsZero() {
        return c.cleanup(namespace, name)
    }

    // 3. 调谐：对比期望 vs 实际
    fmt.Printf("Reconcile MySQL %s/%s: replicas=%d, version=%s\n",
        namespace, name, mysql.Spec.Replicas, mysql.Spec.Version)

    // 真实场景这里会：
    // - 查关联的 StatefulSet 是否存在
    // - 不存在就创建，存在就比对 spec 是否一致
    // - 不一致就更新
    if err := c.ensureStatefulSet(mysql); err != nil {
        return err  // 返回 error 会触发 requeue
    }

    // 4. 更新 status
    if err := c.updateStatus(mysql); err != nil {
        return err
    }

    return nil
}

Reconcile 的逻辑就是第 2 章讲的那五步：读期望 → 读实际 → 算差异 → 调谐 → 写 status。

注意所有错误都 return err——返回 error 后 worker 会把这个 key requeue，带退避重试。这是 Controller 自愈的关键：临时失败不用慌，下轮再来。

Worker 循环

func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
    defer runtime.HandleCrash()
    defer c.queue.ShutDown()

    // 启动 Informer
    go c.informer.Run(stopCh)

    // 等缓存同步完成
    if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
        runtime.HandleError(fmt.Errorf("缓存同步超时"))
        return
    }

    fmt.Println("Controller 启动，开始调谐")

    // 启动 N 个 worker
    for i := 0; i < workers; i++ {
        go wait.Until(c.worker, time.Second, stopCh)
    }

    <-stopCh
    fmt.Println("Controller 停止")
}

func (c *Controller) worker() {
    for c.processNextItem() {
    }
}

func (c *Controller) processNextItem() bool {
    // 1. 从队列取一个 key
    key, quit := c.queue.Get()
    if quit {
        return false
    }
    defer c.queue.Done(key)

    // 2. 调 Reconcile
    err := c.Reconcile(key.(string))

    // 3. 成功就 Forget，失败就 requeue
    if err == nil {
        c.queue.Forget(key)
    } else if c.queue.NumRequeues(key) < 5 {
        // 失败但重试次数 < 5，带退避重试
        c.queue.AddRateLimited(key)
    } else {
        // 超过 5 次，放弃
        runtime.HandleError(err)
        c.queue.Forget(key)
    }

    return true
}

Run 干三件事：

1. 启动 Informer：开始 List + Watch，建本地缓存
2. 等缓存同步：WaitForCacheSync 阻塞等 List 完成。缓存没好就开 reconcile 会读到空数据
3. 启 worker：默认开多个 worker 并发消费 queue

processNextItem 是 worker 的主循环——取 key、调 Reconcile、处理结果。Get/Done/Forget/AddRateLimited 是 Workqueue 的标准用法：

方法	作用
Get()	取出 key，标记为"处理中"
Done(key)	标记处理完成（不管成功失败）
Forget(key)	清掉重试计数（成功时调）
AddRateLimited(key)	带退避重新入队（失败时调）
NumRequeues(key)	这个 key 重试了几次

main 函数

func main() {
    kubeconfig := os.Getenv("KUBECONFIG")

    ctrl, err := NewController(kubeconfig)
    if err != nil {
        log.Fatal(err)
    }

    stopCh := make(chan struct{})
    defer close(stopCh)

    // 开 2 个 worker
    ctrl.Run(2, stopCh)
}

跑起来，然后：

kubectl apply -f mysql-crd.yaml
kubectl apply -f mysql-instance.yaml
# Controller 日志：Reconcile MySQL default/my-db: replicas=3, version=8.0

几个关键设计

为什么 worker 可以开多个

多个 worker 并发从 queue 取 key 处理。但同一个 key 在 queue 里只存一份（去重），所以同一个资源不会被并发处理。不同资源可以并行——这就是 worker 并发的意义。

queue: [A, B, C, D]
        │  │  │  │
worker1: A  C
worker2: B  D

A 和 B 可以并行处理，但 A 不会被两个 worker 同时处理。

为什么缓存没同步好不能开 reconcile

WaitForCacheSync 这一行不能省。如果缓存还在 List 阶段，Indexer 是空的，Reconcile 读到"资源不存在"，误以为该创建——但其实资源在集群里有，只是缓存还没拉到。结果重复创建。

打比方：新来的仓管员还没盘点完，客户来问"还有多少货"，他看账本是空的就说"没货"——实际货在仓库里，他只是还没数到。

Finalizer：删除前的清理

if !mysql.DeletionTimestamp.IsZero() {
    return c.cleanup(namespace, name)
}

DeletionTimestamp 非空表示用户执行了 kubectl delete。但如果资源有 finalizer，API Server 不会真删——会等 finalizer 清理完才删。

metadata:
  finalizers:
    - mysql.database.example.com/finalizer

Controller 看到删除请求，先做清理（删关联的 StatefulSet、Service、PVC），然后移除 finalizer，资源才真删。

这跟装修退租一个道理：房东收到你的退租通知（DeletionTimestamp），但不会立刻放人——等你把家具搬走、卫生打扫完（finalizer 清理），才退押金（真删资源）。

这个 Controller 缺什么

这个 Controller 能跑，但离生产级还差几样：

缺什么	影响	下章谁解决
领导选举（Leader Election）	多副本时多个 Controller 同时调谐，会冲突	Kubebuilder 自带
优雅退出	收到 SIGTERM 时没等队列排空	Kubebuilder 的 ctx 管理
结构化日志	fmt.Printf 没法过滤、聚合	Kubebuilder 用 logr
指标暴露	看不到 reconcile 次数、队列长度	Kubebuilder 自带 Prometheus 指标
Webhook	没法在提交时校验 spec	Kubebuilder 生成 webhook 骨架

这些下一章 Kubebuilder 全帮你搞定。你写 200 行裸 Controller 的价值在于——知道这些"黑盒"里到底是什么。

前置阅读

• 控制器模式与CRD — Starter 第 12 章，Informer/Workqueue 的概念入门
• 理解控制循环 — 本章前置，讲 Informer/Workqueue/Reconcile 的原理
• 编写CRD — 本章前置，Controller 操作的资源是上一章定义的 MySQL CRD
• K8S架构 — Starter 第 2 章，Controller 通过 API Server Watch，理解架构更清楚

思考题

问题 1：processNextItem 里 Done(key) 和 Forget(key) 都调了，但失败时只调 Done 不调 Forget。为什么？

问题 2：worker 开 1 个和开 10 个有什么区别？什么情况下开多了反而不好？

问题 3：WaitForCacheSync 超时了直接 return，不启 worker。如果跳过这步直接启 worker 会怎样？

---

解答

问题 1：Done 是告诉 queue"这个 key 处理完了，可以从处理中状态移除"。Forget 是清掉重试计数。失败时不能 Forget——计数还在，下次 NumRequeues 才知道重试了几次，超过上限就放弃。如果失败时也 Forget，计数永远是 0，Controller 会无限重试，把系统打爆。

问题 2：开多个 worker 是为了并行处理不同资源，提升吞吐。但开多了不一定好——每个 worker 都要调 API Server，开 100 个 worker 会把 API Server 打爆。而且如果资源之间有依赖（A 的 reconcile 依赖 B 的状态），并发处理可能读到 B 的中间状态。一般 1-5 个 worker 够用。

问题 3：缓存没同步好，Indexer 是空的。worker 跑起来后，第一个 key 出队，indexer.GetByKey 返回 exists=false，Controller 以为资源被删了，触发清理逻辑——但实际资源在集群里，只是缓存还没拉到。结果误删关联资源。这就是为什么 WaitForCacheSync 必须等。

下一章

理解了底层，下一章我们使用 Kubebuilder——用框架重写这个 Controller，看它帮你省了什么，又引入了什么。