Kubernetes Informer 机制详解

r.store.Add(event.Object)

type DeltaFIFO struct {
    // 锁，保护队列安全
    lock sync.RWMutex
    
    // items 是 Map，key 是资源对象的 key (例如 "default/nginx")
    // value 是一个 list，记录了这个对象发生的所有事件变化 (Deltas)
    items map[string]Deltas
    
    // queue 记录了 key 的顺序，这是一个先进先出队列
    queue []string
}

type Delta struct {
    Type   DeltaType   // Added, Updated, Deleted, Sync
    Object interface{} // 具体的 Pod 对象
}

type Deltas []Delta

func (c *controller) processLoop() {
    for {
        // 重点！！！
        // 这里死循环调用 Queue.Pop
        // c.config.Queue 就是 DeltaFIFO
        obj, err := c.config.Queue.Pop(c.config.Process)
        
        if err != nil {
            // ...
        }
    }
}

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
    f.lock.Lock()
    defer f.lock.Unlock()
    
    for {
        // 1. 如果队列是空的，就阻塞等待（Cond.Wait）
        for len(f.queue) == 0 {
            // ... wait logic ...
        }
        
        // 2. 取出队头的 key (FIFO)
        id := f.queue[0]
        f.queue = f.queue[1:]
        
        // 3. 拿到这个 key 对应的所有事件列表 (Deltas)
        item, ok := f.items[id]
        delete(f.items, id) // 从 map 里删掉，表示正在处理
        
        // 4. 调用 process 函数（这里最终会触发你写的 OnAdd/OnUpdate）
        err := process(item)
        // process 函数会做两件事情：
        // 1. 更新 indexer 缓存
        // 2. 分发事件
    }
}

type Controller struct {
    // 1. 客户端：用于读写 k8s 资源
    kubeclientset kubernetes.Interface
    
    // 2. Lister：用于从本地缓存（Indexer）里快速读取数据
    servicesLister corelisters.ServiceLister
    servicesSynced cache.InformerSynced
    
    // 3. WorkQueue: 核心，限速队列
    // 作用：去重、重试、并发控制
    workqueue workqueue.RateLimitingInterface
}

func NewController(kubeclientset kubernetes.Interface, informerFactory informers.SharedInformerFactory) *Controller {
    // 获取 Service 的 Informer
    serviceInformer := informerFactory.Core().V1().Services()
    
    c := &Controller{
        kubeclientset:  kubeclientset,
        servicesLister: serviceInformer.Lister(),
        servicesSynced: serviceInformer.Informer().HasSynced,
        // 使用默认的限速队列
        workqueue: workqueue.NewNamedRateLimitingQueue(
            workqueue.DefaultControllerRateLimiter(), 
            "Services",
        ),
    }
    
    fmt.Println("Setting up event handlers...")
    // 注册 Event Handler
    // 注意：我们在这里不写业务逻辑，只负责把 Key 丢进队列
    serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: c.enqueueService,
        UpdateFunc: func(old, new interface{}) {
            // 只有当 Annotation 发生变化时才入队，避免无意义的 loop
            oldSvc := old.(*corev1.Service)
            newSvc := new.(*corev1.Service)
            if oldSvc.ResourceVersion == newSvc.ResourceVersion {
                return
            }
            c.enqueueService(new)
        },
        // Delete 暂时不做处理
    })
    return c
}

func (c *Controller) enqueueService(obj interface{}) {
    key, err := cache.MetaNamespaceKeyFunc(obj)
    if err != nil {
        runtime.HandleError(err)
        return
    }
    // Add(): 这是一个非阻塞操作
    c.workqueue.Add(key)
}

func (c *Controller) Run(stopCh <-chan struct{}) error {
    // 1. 捕获 Panic，防止程序直接崩溃退出（这是 Controller 的保命符）
    defer runtime.HandleCrash()
    
    // 2. 优雅关闭队列
    defer c.workqueue.ShutDown()
    
    fmt.Println("Starting Service controller")
    
    // 1. 等待缓存同步完成
    // 这是最重要的一步，必须等本地缓存有了数据再开始干活
    if !cache.WaitForCacheSync(stopCh, c.servicesSynced) {
        return fmt.Errorf("failed to wait for caches to sync")
    }
    
    fmt.Println("Cache synced. Starting workers...")
    
    // 2. 启动 Worker
    // 这里的 wait.Until 是一个死循环，每秒钟启动一次 runWorker
    // 只要 runWorker 退出（处理完一个），它就会再次启动它
    go wait.Until(c.runWorker, time.Second, stopCh)
    
    fmt.Println("Worker started")
    <-stopCh
    return nil
}

func (c *Controller) processNextItem() bool {
    // 1. Get(): 阻塞直到队列里有数据
    key, quit := c.workqueue.Get()
    if quit {
        return false
    }
    defer c.workqueue.Done(key)
    
    // 2. 调用业务逻辑
    err := c.syncHandler(key.(string))
    
    // 3. 处理错误
    c.handleErr(err, key)
    
    return true
}

// syncHandler：真正的业务逻辑（调谐 Reconcile）
// 这里的逻辑必须是"幂等"的！
func (c *Controller) syncHandler(key string) error {
    // 解析 Key -> namespace, name
    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        return err
    }
    
    // 1. 从本地缓存 (Lister) 获取 Service 对象
    // 注意：这里读的是内存，极快
    service, err := c.servicesLister.Services(namespace).Get(name)
    if err != nil {
        if errors.IsNotFound(err) {
            // 如果对象被删除了，我们不需要做任何事
            return nil
        }
        return err
    }
    
    // 2. 检查 Annotation
    strategy, ok := service.Annotations[StrategyAnnotation]
    if !ok {
        // 如果没有我们关心的 Annotation，忽略
        return nil
    }
    
    // 🌟 核心演示：模拟崩溃 🌟
    // 如果 annotation 设置为 "panic"，我们直接让程序崩溃
    if strategy == "panic" {
        fmt.Printf("💣 BOOM! Triggered panic for %s\n", key)
        panic("Simulated Controller Crash!")
    }
    
    fmt.Printf("⚡ Syncing Service: %s, Strategy: %s\n", name, strategy)
    
    // 3. 检查 Selector 是否符合预期
    // 预期：app = {strategy} (例如 app=blue)
    currentSelector := service.Spec.Selector["app"]
    if currentSelector == strategy {
        // 状态一致，无需操作
        return nil
    }
    
    // 4. 执行更新 (Side Effect)
    // 因为 Lister 拿出来的对象是只读指针，必须 DeepCopy 才能修改
    serviceCopy := service.DeepCopy()
    if serviceCopy.Spec.Selector == nil {
        serviceCopy.Spec.Selector = make(map[string]string)
    }
    serviceCopy.Spec.Selector["app"] = strategy
    
    // 5. 调用 APIServer 更新
    _, err = c.kubeclientset.CoreV1().Services(namespace).Update(
        context.TODO(), 
        serviceCopy, 
        metav1.UpdateOptions{},
    )
    if err != nil {
        return err
    }
    
    fmt.Printf("✅ Successfully updated Service %s selector to %s\n", name, strategy)
    return nil
}

func (c *Controller) handleErr(err error, key interface{}) {
    if err == nil {
        // 成功，清除重试计数器
        c.workqueue.Forget(key)
        return
    }
    
    // 如果出现错误（比如 APIServer 冲突），通过 AddRateLimited 重新入队
    // 这会实现指数退避（Wait 1s, 2s, 4s, 8s...）
    if c.workqueue.NumRequeues(key) < 5 {
        fmt.Printf("Error syncing '%v': %v. Retrying...\n", key, err)
        c.workqueue.AddRateLimited(key)
        return
    }
    
    // 重试次数过多，放弃
    c.workqueue.Forget(key)
    runtime.HandleError(err)
    fmt.Printf("Dropping service '%q' out of the queue: %v\n", key, err)
}