Kubernetes 事件驱动：Informer 机制不只是给控制器用

Kubernetes 事件驱动：Informer 机制不只是给控制器用

一、Informer 为什么值得所有 K8s 开发者深入理解

Kubernetes 中几乎所有控制器都基于 Informer 实现：Deployment Controller、ReplicaSet Controller、Scheduler 等。Informr 是 client-go 的核心组件，负责从 API Server Watch 资源变更、维护本地缓存、触发回调处理。大多数人只知道"控制器用 Informer"，但实际上任何需要响应集群状态变化的组件都应该用 Informer——运维工具、监控采集器、事件告警系统、自定义调度器。

基础设施不需要漂亮话，轮询 API Server 获取资源列表就是低效操作，Watch 机制才是 K8s 事件驱动的基础。轮询方式每秒一次 List 请求，100 个工具 Pod 同时轮询，API Server 负载不可忽视。Informer 通过 Watch + List 的组合，只在启动时全量 List 一次，后续增量 Watch，内存缓存提供 Read 操作，API Server 只接收 Watch 连接。

Informer 不是"高级用法"，而是 K8s 编程模型的基本范式。理解 Informer 机制后，自定义运维工具、事件处理器、调度扩展的开发效率显著提升。

二、Informer 的内部机制与事件流

Informer 由四个核心组件构成：Reflector、DeltaQueue、LocalStore、EventHandler。

sequenceDiagram participant API as API Server participant Ref as Reflector participant DQ as Delta Queue participant LS as Local Store participant EH as Event Handler participant App as 业务处理函数 Ref->>API: List (全量获取资源) API-->>Ref: 返回所有对象 Ref->>DQ:推送 Added 事件 Ref->>LS: 全量写入本地缓存 loop Watch 增量流 Ref->>API: Watch (持续监听变更) API-->>Ref: 返回增量事件流 Ref->>DQ: 推送 Added/Updated/Deleted DQ->>LS: 更新本地缓存 DQ->>EH: 分发事件到回调 EH->>App: OnAdd/OnUpdate/OnDelete end Note over LS: 业务读取直接查 Local Store，不再请求 API Server

关键机制说明：

Resync 机制。resyncPeriod控制定期全量同步间隔。Watch 连接可能因网络中断丢失事件，resync 确保缓存与 API Server 最终一致。默认 resync 为 0（不定期同步），生产环境建议设为 30 分钟。resync 不是重新 List，而是将缓存中所有对象重新推送 Added 事件到 DeltaQueue。

SharedInformer。同一资源的多个 Informer 实例共享一个 Reflector 和 LocalStore，减少 Watch 连接数和内存占用。K8s 内部所有控制器通过 SharedInformerFactory 共享 Informer，API Server 的 Watch 连接数等于资源类型数，而非控制器数。

三、Informer 事件处理器实现

基于 Informer 的 Pod 事件告警工具：

// pod_event_monitor.go — 基于 Informer 的 Pod 事件监控 package main import ( "context" "fmt" "log" "os" "os/signal" "path/filepath" "syscall" "time" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/fields" "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/tools/clientcmd" ) // PodEventHandler Pod 事件处理器 type PodEventHandler struct { namespace string // 监控的命名空间 alertRules []AlertRule // 告警规则列表 stopCh chan struct{} // 停止信号通道 eventBuffer []PodEvent // 事件缓冲区 flushInterval time.Duration // 缓冲区刷新间隔 } // AlertRule 告警规则 type AlertRule struct { Name string // 规则名称 Condition func(*corev1.Pod) bool // 触发条件 Severity string // 告警级别: critical, warning, info Message string // 告警消息模板 } // PodEvent Pod 事件记录 type PodEvent struct { Timestamp time.Time PodName string Namespace string EventType string // Added, Updated, Deleted Reason string // 告警触发原因 Severity string Details string } func NewPodEventHandler(namespace string, flushInterval time.Duration) *PodEventHandler { handler := &PodEventHandler{ namespace: namespace, stopCh: make(chan struct{}), flushInterval: flushInterval, } // 注册告警规则 handler.alertRules = []AlertRule{ { Name: "Pod 重启次数过多", Condition: func(pod *corev1.Pod) bool { for _, cs := range pod.Status.ContainerStatuses { if cs.RestartCount > 5 { return true } } return false }, Severity: "warning", Message: "Pod %s 重启次数超过 5 次", }, { Name: "Pod 处于 Failed 状态", Condition: func(pod *corev1.Pod) bool { return pod.Status.Phase == corev1.PodFailed }, Severity: "critical", Message: "Pod %s 进入 Failed 状态", }, { Name: "Pod 长时间未就绪", Condition: func(pod *corev1.Pod) bool { for _, cond := range pod.Status.Conditions { if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionFalse { // 未就绪超过 5 分钟 if time.Since(cond.LastTransitionTime.Time) > 5*time.Minute { return true } } } return false }, Severity: "warning", Message: "Pod %s 超过 5 分钟未就绪", }, { Name: "GPU Pod 被 OOMKill", Condition: func(pod *corev1.Pod) bool { for _, cs := range pod.Status.ContainerStatuses { if cs.LastTerminationState.OOMKilled != nil { // 检查是否请求了 GPU for r := range pod.Spec.Containers[0].Resources.Requests { if r.String() == "nvidia.com/gpu" { return true } } } } return false }, Severity: "critical", Message: "GPU Pod %s 被 OOMKill", }, } return handler } // OnAdd Pod 新增事件回调 func (h *PodEventHandler) OnAdd(obj interface{}) { pod, ok := obj.(*corev1.Pod) if !ok { log.Printf("OnAdd 类型转换失败: %T", obj) return } log.Printf("[OnAdd] Pod 新增: %s/%s", pod.Namespace, pod.Name) // 检查是否匹配告警规则 for _, rule := range h.alertRules { if rule.Condition(pod) { h.eventBuffer = append(h.eventBuffer, PodEvent{ Timestamp: time.Now(), PodName: pod.Name, Namespace: pod.Namespace, EventType: "Added", Reason: rule.Name, Severity: rule.Severity, Details: fmt.Sprintf(rule.Message, pod.Name), }) } } } // OnUpdate Pod 更新事件回调 func (h *PodEventHandler) OnUpdate(oldObj, newObj interface{}) { oldPod, ok := oldObj.(*corev1.Pod) if !ok { return } newPod, ok := newObj.(*corev1.Pod) if !ok { return } // 只关注有意义的状态变化 if oldPod.Status.Phase == newPod.Status.Phase && oldPod.Status.ContainerStatuses == nil && newPod.Status.ContainerStatuses == nil { return // 状态未变化，跳过 } log.Printf("[OnUpdate] Pod 状态变化: %s/%s phase=%s->%s", newPod.Namespace, newPod.Name, oldPod.Status.Phase, newPod.Status.Phase) // 检查告警规则 for _, rule := range h.alertRules { if rule.Condition(newPod) { h.eventBuffer = append(h.eventBuffer, PodEvent{ Timestamp: time.Now(), PodName: newPod.Name, Namespace: newPod.Namespace, EventType: "Updated", Reason: rule.Name, Severity: rule.Severity, Details: fmt.Sprintf(rule.Message, newPod.Name), }) } } } // OnDelete Pod 删除事件回调 func (h *PodEventHandler) OnDelete(obj interface{}) { pod, ok := obj.(*corev1.Pod) if !ok { // 处理 DeletedFinalStateUnknown（缓存中已删除的对象） tombstone, ok := obj.(cache.DeletedFinalStateUnknown) if ok { pod, ok = tombstone.Obj.(*corev1.Pod) } if !ok { log.Printf("OnDelete 类型转换失败: %T", obj) return } } log.Printf("[OnDelete] Pod 删除: %s/%s", pod.Namespace, pod.Name) } // Start 启动 Informer 和事件处理循环 func (h *PodEventHandler) Start(clientset *kubernetes.Clientset) { // 创建 SharedInformerFactory，30 分钟 resync factory := informers.NewSharedInformerFactoryWithOptions( clientset, 30*time.Minute, // resync 间隔 informers.WithNamespace(h.namespace), // 限定命名空间 ) // 获取 Pod Informer podInformer := factory.Core().V1().Pods().Informer() // 注册事件回调 podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: h.OnAdd, UpdateFunc: h.OnUpdate, DeleteFunc: h.OnDelete, }) // 启动 Factory（所有 Informer 共享底层 Watch 连接） factory.Start(h.stopCh) // 等待初始缓存同步完成 if !cache.WaitForCacheSync(h.stopCh, podInformer.HasSynced) { log.Println("缓存同步失败，退出") return } log.Println("Informer 缓存同步完成，开始监控") // 定时刷新事件缓冲区 go h.flushLoop() // 等待停止信号 sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT) <-sigCh log.Println("收到停止信号，关闭 Informer") close(h.stopCh) } // flushLoop 定时刷新事件缓冲区 func (h *PodEventHandler) flushLoop() { ticker := time.NewTicker(h.flushInterval) defer ticker.Stop() for { select { case <-ticker.C: if len(h.eventBuffer) > 0 { h.flushEvents() } case <-h.stopCh: // 最终刷新 if len(h.eventBuffer) > 0 { h.flushEvents() } return } } } // flushEvents 批量发送告警事件 func (h *PodEventHandler) flushEvents() { events := h.eventBuffer h.eventBuffer = nil // 清空缓冲区 for _, event := range events { log.Printf("[ALERT] severity=%s pod=%s/%s reason=%s details=%s", event.Severity, event.Namespace, event.PodName, event.Reason, event.Details) // 实际生产环境：发送到告警系统 // alertClient.Send(event) } log.Printf("刷新告警事件 %d 条", len(events)) } // 从缓存查询 Pod（不请求 API Server） func (h *PodEventHandler) ListPodsFromCache(informer cache.SharedIndexInformer) []*corev1.Pod { var pods []*corev1.Pod objs := informer.GetIndexer().List() for _, obj := range objs { pod, ok := obj.(*corev1.Pod) if ok { pods = append(pods, pod) } } return pods } func main() { // 创建 K8s 客户端 config, err := rest.InClusterConfig() if err != nil { // 集群外运行时加载 kubeconfig kubeconfig := filepath.Join(os.Getenv("HOME"), ".kube", "config") config, err = clientcmd.BuildConfigFromKubeconfigGetter("", func() (*clientcmd.Config, error) { return clientcmd.LoadFromFile(kubeconfig) }) if err != nil { log.Fatalf("创建 K8s 配置失败: %v", err) } } clientset, err := kubernetes.NewForConfig(config) if err != nil { log.Fatalf("创建 K8s 客户端失败: %v", err) } // 启动 Pod 事件监控 namespace := "ai-inference" // 监控推理命名空间 handler := NewPodEventHandler(namespace, 10*time.Second) handler.Start(clientset) }

四、Informer 使用边界与性能考量

场景一：Watch 连接中断。网络抖动、API Server 重启都会导致 Watch 连接中断。Reflector 自动重连，但中断期间的事件可能丢失。resync 机制保证最终一致性，但不是强一致性。对一致性要求极高的场景（如资源配额管理），需要定期 List 校验缓存与实际状态的一致性。

场景二：大规模集群的内存占用。SharedInformer 在内存中缓存所有 Watch 的资源对象。10 万个 Pod 的集群，每个 Pod 对象约 5KB，缓存占用约 500MB。选择性地只 Watch 特定命名空间或标签的资源，用WithNamespace和WithTweakListOptions过滤，减少缓存大小。

场景三：事件处理速度。EventHandler 的回调函数在 DeltaQueue 的消费线程中执行。如果回调处理耗时过长，DeltaQueue 堆积事件，缓存更新延迟。耗时操作（如调用外部 API）应该异步处理：回调中只记录事件到缓冲区，独立工作线程批量处理。上面的实现中 flushLoop 就是这个模式。

场景四：自定义索引。LocalStore 支持按 Indexer 查询，比如按标签索引 Pod。podInformer.GetIndexer().ByIndex("namespace", "ai-inference")直接查询，比遍历所有对象高效。自定义索引需要在 Informer 创建时注册 Indexers。

五、总结

Informer 是 Kubernetes 事件驱动编程的核心组件，通过 List + Watch + LocalStore 实现高效资源监听。Reflector 负责从 API Server 获取和监听变更，DeltaQueue 保证事件有序分发，LocalStore 提供内存缓存避免重复请求，EventHandler 触发业务回调。SharedInformerFactory 让多个消费者共享同一 Watch 连接和缓存，减少 API Server 负载。resync 机制通过定期全量同步保证缓存最终一致。回调处理应异步执行避免阻塞事件分发。大规模集群需要通过命名空间和标签过滤减少缓存大小。Informer 不只是控制器的内部组件，任何需要响应集群状态变化的工具都应该用它替代轮询。