Kubernetes 编程 / Operator 专题【左扬精讲】—— Client-go 源代码分析：生产级 Controller 实践：并发安全、资源清理与高可用设计

当我们把 Controller 部署到生产环境时，会遇到很多开发环境不会出现的问题：多线程并发安全、内存泄漏、资源清理、高可用部署……这些问题如果不在设计阶段就考虑清楚，等上线后再修复代价会非常大。

这一篇文章，我们来系统地学习生产级 Controller 的最佳实践。

Kubernetes 生产实践 高可用 资源泄漏 v1.36.1

一、并发安全：Controller 编程的核心挑战

Controller 天然是多线程并发执行的：Informer 的事件处理在 goroutine 中运行，Worker 处理任务也在多个 goroutine 中运行。如果不处理好并发安全，就会出现数据竞争、状态不一致等问题。

1. WorkQueue 的并发安全保证

WorkQueue 本身是并发安全的，Get() 返回同一个 key 时，只有一个 goroutine 能获取到：

// WorkQueue 的并发安全由其内部锁保证
// 同一时间只有一个 worker 能 Get 到同一个 key
key1, _ := queue.Get()  // goroutine A 获取了 "deploy-1"
key2, _ := queue.Get()  // goroutine B 获取了 "deploy-1"（如果 A 没有 Done）
// 实际上，A 调用 Get() 后，key1 就从队列中移除了
// B 只能获取到其他 key，或者等待

2. Reconcile 的幂等性设计

WorkQueue 保证同一时间只有一个 worker 处理同一个 key，但不同时间可能有多个 worker 处理同一个 key（因为失败重试）。所以 Reconcile 必须设计成幂等的：

// 幂等的 Reconcile 示例：确保 Pod 带有特定标签
func (c *Controller) reconcile(ctx context.Context, deployment *appsv1.Deployment) error {
    // 获取当前 Pod 列表（从 Lister，可能不是最新）
    pods, err := c.podLister.Pods(deployment.Namespace).List(labels.Everything())
    if err != nil {
        return err
    }

    // 筛选属于这个 Deployment 的 Pod
    ownedPods := filterOwnedPods(deployment, pods)

    // 比较期望状态和实际状态
    for _, pod := range ownedPods {
        if pod.Labels["app"] != deployment.Labels["app"] {
            // 更新 Pod（使用 Patch 而不是 Replace，避免冲突）
            patch, _ := json.Marshal([]map[string]interface{}{
                {
                    "op":    "add",
                    "path":  "/metadata/labels/app",
                    "value": deployment.Labels["app"],
                },
            })
            _, err := c.kubeclientset.CoreV1().Pods(pod.Namespace).Patch(
                ctx, pod.Name, types.JSONPatchType, patch, metav1.PatchOptions{},
            )
            if err != nil {
                return err
            }
        }
    }

    return nil  // 幂等：重复执行结果一致
}

3. 避免读取过期数据

Lister 返回的是本地缓存的数据，可能不是最新状态。当需要最新状态时，应该直接查询 APIServer：

// 当需要最新状态时，直接从 APIServer 获取
func (c *Controller) reconcile(ctx context.Context, deployment *appsv1.Deployment) error {
    // Lister 可能有过期数据
    pods, err := c.podLister.Pods(deployment.Namespace).List(...)
    
    // 某些情况下需要最新状态
    latestDeployment, err := c.kubeclientset.AppsV1().Deployments(
        deployment.Namespace).Get(ctx, deployment.Name, metav1.GetOptions{})
    if err != nil {
        return err
    }
    
    // 检查是否有其他 Controller 修改过这个资源
    if latestDeployment.ResourceVersion != deployment.ResourceVersion {
        klog.V(4).Infof("Deployment %s/%s has been modified, will retry",
            deployment.Namespace, deployment.Name)
        return fmt.Errorf("deployment was modified")
    }
    
    // ... 继续处理 ...
}

二、资源清理：Finalizer 的正确使用

当 Controller 管理的资源被删除时，我们需要确保清理逻辑能够执行。Finalizer（最终一致性机制）就是来解决这个问题的。

什么是 Finalizer？

Finalizer 是 Kubernetes 的一种机制，它告诉 APIServer：在删除对象之前，必须先移除某个 finalizer。这个机制常用于：

• 清理外部资源（如云存储卷、外部数据库）
• 等待相关资源删除完成（如等待 Pod 删除）
• 完成数据同步或备份

Finalizer 的工作流程

┌──────────────────────────────────────────────────────────────────────────┐
│                      Finalizer 工作流程                                      │
├──────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  1. 创建资源时，添加 finalizer                                            │
│     metadata.finalizers = ["example.com/cleanup"]                         │
│                                                                          │
│  2. 用户发起删除请求                                                      │
│     kubectl delete foo my-foo                                            │
│                                                                          │
│  3. APIServer 设置 deletionTimestamp，但不立即删除                         │
│     metadata.deletionTimestamp = "2024-01-01T12:00:00Z"                  │
│     → Controller 收到 Update/Delete 事件                                  │
│                                                                          │
│  4. Controller 执行清理逻辑                                               │
│     - 清理关联资源                                                       │
│     - 清理外部资源                                                       │
│     - 完成必要的清理工作                                                   │
│                                                                          │
│  5. Controller 移除 finalizer                                             │
│     PATCH metadata.finalizers = []                                       │
│                                                                          │
│  6. APIServer 完成删除                                                   │
│     ✓ 资源被真正删除                                                     │
│                                                                          │
└──────────────────────────────────────────────────────────────────────────┘

Finalizer 实现示例

const finalizerName = "example.com/cleanup"

func (c *Controller) reconcileFoo(ctx context.Context, foo *examplev1.Foo) error {
    // 检查是否需要添加 finalizer
    if !hasFinalizer(foo, finalizerName) {
        foo = foo.DeepCopy()
        foo.Finalizers = append(foo.Finalizers, finalizerName)
        _, err := c.exampleClient.ExampleV1().Foos(foo.Namespace).Update(ctx, foo, metav1.UpdateOptions{})
        return err
    }

    // 检查是否正在被删除
    if foo.DeletionTimestamp != nil {
        // 执行清理逻辑
        if err := c.cleanupResources(ctx, foo); err != nil {
            return err
        }

        // 移除 finalizer，允许资源被删除
        foo = foo.DeepCopy()
        foo.Finalizers = removeFinalizer(foo.Finalizers, finalizerName)
        _, err := c.exampleClient.ExampleV1().Foos(foo.Namespace).Update(ctx, foo, metav1.UpdateOptions{})
        return err
    }

    // 正常处理逻辑
    // ...
    return nil
}

func hasFinalizer(obj metav1.Object, finalizer string) bool {
    for _, f := range obj.GetFinalizers() {
        if f == finalizer {
            return true
        }
    }
    return false
}

func removeFinalizer(finalizers []string, finalizer string) []string {
    var result []string
    for _, f := range finalizers {
        if f != finalizer {
            result = append(result, f)
        }
    }
    return result
}

⚠️ 警告
如果 Controller 因为 bug 无法移除 finalizer，资源会永远无法删除。这时需要手动介入：kubectl edit foo my-foo 并移除 finalizers 数组。

三、优雅关闭：Controller 的生命周期管理

当 Controller 收到 SIGTERM（终止信号）时，应该优雅地关闭，而不是立即退出。

优雅关闭的步骤

1. 1停止接收新任务：不再接受新的 Reconcile 请求
2. 2等待正在处理的任务完成：给 worker 一段时间处理完当前任务
3. 3关闭 Informer：停止 Watch，释放连接
4. 4释放锁（如使用了 Leader 选举）：让其他副本可以接管

优雅关闭实现

func (c *Controller) Run(ctx context.Context, workers int) error {
    // 等待缓存同步
    if !cache.WaitForCacheSync(ctx.Done(), c.deploymentSynced, c.podSynced) {
        return fmt.Errorf("failed to wait for caches to sync")
    }

    // 启动 workers
    for i := 0; i < workers; i++ {
        c.workerGroup.Go(func() error {
            return c.runWorker(ctx, i)
        })
    }

    // 等待 context 取消
    <-ctx.Done()

    // Context 取消后，开始优雅关闭
    klog.Info("Shutdown signal received, gracefully stopping...")

    // 通知所有 worker 停止
    c.workerGroup.Wait()

    // 关闭 Informer Factory
    c.factory.Shutdown()

    klog.Info("Shutdown complete")
    return nil
}

func (c *Controller) runWorker(ctx context.Context, id int) error {
    for {
        select {
        case <-ctx.Done():
            return nil  // Context 取消，退出
        default:
            if !c.processNextWorkItem(ctx) {
                return nil
            }
        }
    }
}

四、内存泄漏：Informer 的陷阱

Informer 是最容易导致内存泄漏的地方。常见的原因有：

1. 没有调用 Shutdown

// 错误示例：没有关闭 Informer
func main() {
    factory := informers.NewSharedInformerFactory(clientset, 30*time.Second)
    factory.Start(ctx.Done())
    
    // ... 使用 factory ...
    
    // 程序退出时没有调用 Shutdown
    // goroutine 继续运行，持有的内存无法释放
}

// 正确示例
func main() {
    factory := informers.NewSharedInformerFactory(clientset, 30*time.Second)
    
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    factory.Start(ctx.Done())
    
    // ... 使用 factory ...
    
    // 程序退出前调用 Shutdown
    factory.Shutdown()
}

2. 注册了 Handler 但没有清理

// 错误示例：在 Controller 运行时注册了新的 Handler
func (c *Controller) updateHandler(newHandler cache.ResourceEventHandler) {
    // 每次调用都会添加一个新的 Handler！
    c.informer.AddEventHandler(newHandler)
    // 旧 Handler 永远不会被移除
}

// 正确做法：Handler 只注册一次，使用 channel 或其他机制传递事件
func (c *Controller) Run() {
    // 只注册一次 Handler
    c.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            c.eventCh

3. Indexer 持有过期对象

Indexer 会缓存所有见过的对象。如果对象被大量创建和删除，Indexer 会持续积累内存。使用带过期时间的索引可以缓解这个问题：

// 使用带 TTL 的 ThreadSafeStore
indexers := cache.Indexers{
    cache.NamespaceIndex:    cache.MetaNamespaceIndexFunc,
    "byUID": func(obj interface{}) ([]string, error) {
        meta, ok := obj.(metav1.Object)
        if !ok {
            return nil, fmt.Errorf("object is not metav1.Object")
        }
        return []string{string(meta.GetUID())}, nil
    },
}

// 或者定期清理不需要的索引
func (c *Controller) cleanupIndexer() {
    items := c.indexer.List()
    for _, item := range items {
        meta, _ := item.(metav1.Object)
        // 删除过期的对象
        if time.Since(meta.GetDeletionTimestamp().Time) > 5*time.Minute {
            c.indexer.Delete(item)
        }
    }
}

五、高可用部署：多副本 + Leader 选举

生产环境的 Controller 通常需要部署多个副本以保证高可用。

Deployment + Leader 选举模式

┌──────────────────────────────────────────────────────────────────────────┐
│                      高可用部署架构                                          │
├──────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│   ┌─────────────────────────────────────────────────────────────────┐   │
│   │  Deployment (replicas: 3)                                        │   │
│   │                                                                  │   │
│   │   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │   │
│   │   │   Pod-1    │  │   Pod-2    │  │   Pod-3    │          │   │
│   │   │  (Leader)  │  │  (Standby) │  │  (Standby) │          │   │
│   │   └──────┬──────┘  └─────────────┘  └─────────────┘          │   │
│   │          │                                                      │   │
│   │          ▼                                                      │   │
│   │   ┌─────────────┐                                              │   │
│   │   │ Lease Lock │  只允许一个 Pod 处理事件                        │   │
│   │   │(in cluster)│                                              │   │
│   │   └─────────────┘                                              │   │
│   │                                                                  │   │
│   └─────────────────────────────────────────────────────────────────┘   │
│                                                                          │
└──────────────────────────────────────────────────────────────────────────┘

Deployment 配置

apiVersion: apps/v1
kind: Deployment
metadata:
  name: example-operator
  namespace: operators
spec:
  replicas: 3
  selector:
    matchLabels:
      app: example-operator
  template:
    metadata:
      labels:
        app: example-operator
    spec:
      serviceAccountName: example-operator
      containers:
      - name: operator
        image: example/operator:v1.0.0
        args:
        - --leader-elect=true
        - --leader-election-namespace=operators
        # Pod 必须配置优雅终止宽限期
      terminationGracePeriodSeconds: 30

六、监控指标：生产级 Controller 的必备功能

生产环境的 Controller 应该暴露 metrics，方便监控和告警。

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    reconcileTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "controller_reconcile_total",
            Help: "Total number of reconcile attempts",
        },
        []string{"result"},  // result: success, error
    )

    reconcileDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "controller_reconcile_duration_seconds",
            Help:    "Duration of reconcile operations",
            Buckets: prometheus.DefBuckets,
        },
        []string{"result"},
    )

    workqueueDepth = promauto.NewGauge(
        prometheus.GaugeOpts{
            Name: "controller_workqueue_depth",
            Help: "Current depth of the workqueue",
        },
    )
)

func (c *Controller) reconcile(ctx context.Context, key string) error {
    start := time.Now()
    defer func() {
        duration := time.Since(start).Seconds()
        reconcileDuration.WithLabelValues("success").Observe(duration)
        reconcileTotal.WithLabelValues("success").Inc()
    }()

    err := c.doReconcile(ctx, key)
    if err != nil {
        reconcileDuration.WithLabelValues("error").Observe(time.Since(start).Seconds())
        reconcileTotal.WithLabelValues("error").Inc()
        return err
    }

    return nil
}

🌟 实用技巧
建议暴露以下 metrics：reconcile 总次数和成功率、reconcile 耗时分布、workqueue 深度、Leader 选举切换次数。

七、总结

这一节是 client-go 系列的最后一篇，我们总结了生产级 Controller 的最佳实践：

• 并发安全：Reconcile 必须是幂等的，避免读取过期数据
• Finalizer：确保清理逻辑能够执行，避免资源泄漏
• 优雅关闭：正确处理 SIGTERM，给 worker 时间完成当前任务
• 内存泄漏：正确关闭 Informer，避免注册重复 Handler
• 高可用：多副本部署 + Leader 选举
• 监控指标：暴露 metrics，方便监控和告警

至此，client-go 系列的全部文章已经完成。从 Informer 源码、DeltaFIFO、Indexer、WorkQueue，到 SharedInformerFactory、Controller 开发模式、Leader 选举、限速器、Watch 机制、DynamicClient，再到调试工具和生产实践，希望这一系列文章能帮助您真正掌握 Kubernetes 客户端编程。

Kubernetes 编程 / Operator 专题【左扬精讲】—— 生产级 Controller 实践 · 来源：Kubernetes v1.36.1 client-go 源码分析