【深入浅出】Kubernetes控制器：云原生架构的无形守护者

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/informers"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/rest"
    "k8s.io/client-go/tools/cache"
    "k8s.io/client-go/util/workqueue"
    "k8s.io/apimachinery/pkg/fields"
    "k8s.io/apimachinery/pkg/watch"
    corev1 "k8s.io/api/core/v1"
)

// Controller 代表管理Pod生命周期的控制器
type Controller struct {
    clientset     kubernetes.Interface
    queue         workqueue.RateLimitingInterface
    informer      cache.SharedIndexInformer
}

// NewController 创建一个新的Controller
func NewController() *Controller {
    // 使用Kubernetes client-go的in-cluster配置创建一个新配置
    // In-cluster配置会使用kubernetes为pods提供的服务账户来连接到集群
    config, err := rest.InClusterConfig()
    if err != nil {
        panic(err.Error())
    }

    // 创建一个新的clientset，包括所有不同API组的rest客户端
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err.Error())
    }

    // 创建一个pod informer来观察pod变化
    podInformer := informers.NewSharedInformerFactoryWithOptions(
        clientset, 
        time.Minute*10,
        informers.WithTweakListOptions(func(options *metav1.ListOptions) {
            options.FieldSelector = fields.OneTermEqualSelector("status.phase", string(corev1.PodRunning)).String()
        }),
    ).Core().V1().Pods().Informer()

    // 为控制器创建一个限速的工作队列
    queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())

    // 定义增加/更新/删除事件的处理方法
    podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            key, err := cache.MetaNamespaceKeyFunc(obj)
            if err == nil {
                queue.Add(key)
            }
        },
        UpdateFunc: func(old, new interface{}) {
            key, err := cache.MetaNamespaceKeyFunc(new)
            if err == nil {
                queue.Add(key)
            }
        },
    })

    // 返回一个新的Controller
    return &Controller{
        clientset: clientset,
        informer:  podInformer,
        queue:     queue,
    }
}

// Run 启动控制器
func (c *Controller) Run(stopCh <-chan struct{}) {
    // 运行informer以开始监听更新事件
    go c.informer.Run(stopCh)

    // 等待缓存同步完成后再开始工作
    if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
        runtime.HandleError(fmt.Errorf("缓存同步错误"))
        return
    }

    // 开始一个工作循环来处理工作队列
    go wait.Until(c.runWorker, time.Second, stopCh)

    // 阻塞直到stop通道被关闭
    <-stopCh
}

// runWorker 是一个长时间运行的函数，不断调用processNextItem函数
// 以便读取并处理工作队列上的消息
func (c *Controller) runWorker() {
    for c.processNextItem() {
        // 继续处理
    }
}
// processNextItem 从工作队列中读取单个工作项并尝试处理它
func (c *Controller) processNextItem() bool {
    // 从工作队列中获取下一个项目
    key, quit := c.queue.Get()
    if quit {
        return false
    }

    // 告诉队列我们已经完成了这个键的处理
    defer c.queue.Done(key)

    // 处理键
    err := c.syncHandler(key.(string))
    if err == nil {
        // 没有错误，重置速率限制计数器
        c.queue.Forget(key)
    } else if c.queue.NumRequeues(key) < 5 {
        // 重试处理
        c.queue.AddRateLimited(key)
    } else {
        // 重试太多次，放弃项目
        c.queue.Forget(key)
        runtime.HandleError(err)
    }

    return true
}

// syncHandler 处理键并执行业务逻辑
func (c *Controller) syncHandler(key string) error {
    // 将键分割为命名空间和对象名
    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        return err
    }

    // 从集群中获取这个命名空间/名字的Pod资源
    pod, err := c.clientset.CoreV1().Pods(namespace).Get(context.Background(), name, metav1.GetOptions{})
    if err != nil {
        return err
    }

    // 在这里写你的控制器逻辑，例如，打印Pod名称和它的状态
    fmt.Printf("正在处理Pod: %s, 状态: %s\n", pod.GetName(), pod.Status.Phase)

    // 在这个例子中，我们只是返回nil
    return nil
}

func main() {
    // 构建我们的控制器并运行它
    controller := NewController()

    // 创建一个信号通道来指示我们何时想要停止控制器
    stop := make(chan struct{})

    // 运行控制器
    controller.Run(stop)
}