使用Go Shared Informers监听K8s镜像变更未获正确新旧版本求助

核心问题原因 #

你监听Pod的onUpdate事件无法拿到新旧镜像版本，主要有两个关键原因：

1. Deployment更新镜像时，并非直接修改现有Pod的镜像，而是创建新ReplicaSet并启动使用新镜像的Pod，同时逐步删除旧ReplicaSet下的旧Pod。单个Pod的镜像从创建到销毁不会变化，因此Pod的onUpdate事件仅会因Pod状态（如Pending→Running）或元数据变更触发，新旧Pod实例的镜像始终一致。
2. 你看到的三次onUpdate触发，是Pod生命周期中状态变更的正常现象（比如调度分配节点、容器启动完成等），并非镜像版本变化导致。

最优解决方案：监听Deployment的更新事件 #

直接监听Deployment的变更，因为Deployment的spec.template.spec.containers[0].Image字段会直接反映镜像版本的修改，在onUpdate回调中可以直接对比新旧Deployment的镜像值。

示例代码

package main

import (
	"context"
	"fmt"
	"time"

	appsv1 "k8s.io/api/apps/v1"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/clientcmd"
)

func main() {
	// 加载K8s配置（本地开发用~/.kube/config，集群内用ServiceAccount）
	config, err := clientcmd.BuildConfigFromFlags("", clientcmd.RecommendedHomeFile)
	if err != nil {
		panic(err.Error())
	}

	// 创建K8s客户端
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(err.Error())
	}

	// 创建SharedInformer工厂，设置30分钟的重新同步周期
	factory := informers.NewSharedInformerFactory(clientset, time.Minute*30)

	// 获取Deployment的Informer
	deploymentInformer := factory.Apps().V1().Deployments().Informer()

	// 注册事件处理函数
	deploymentInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		UpdateFunc: func(oldObj, newObj interface{}) {
			// 类型断言转为Deployment对象
			oldDeploy, ok := oldObj.(*appsv1.Deployment)
			if !ok {
				fmt.Println("Failed to cast old object to Deployment")
				return
			}
			newDeploy, ok := newObj.(*appsv1.Deployment)
			if !ok {
				fmt.Println("Failed to cast new object to Deployment")
				return
			}

			// 获取新旧镜像版本
			oldImage := oldDeploy.Spec.Template.Spec.Containers[0].Image
			newImage := newDeploy.Spec.Template.Spec.Containers[0].Image

			// 仅当镜像变化时输出
			if oldImage != newImage {
				fmt.Printf("Deployment [%s/%s] image updated: Old=%s, New=%s\n",
					oldDeploy.Namespace, oldDeploy.Name, oldImage, newImage)
			}
		},
	})

	// 启动Informer
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	factory.Start(ctx.Done())

	// 等待缓存同步完成
	for _, ok := range factory.WaitForCacheSync(ctx.Done()) {
		if !ok {
			panic("Failed to sync cache")
		}
	}

	// 阻塞程序运行
	<-ctx.Done()
}

备选方案：监听Pod的创建/删除事件（复杂场景） #

如果业务需求必须通过Pod层面跟踪镜像变化，可以监听Pod的AddFunc和DeleteFunc，通过Pod的ownerReferences关联到所属的ReplicaSet和Deployment，从而匹配同一Deployment下的新旧Pod镜像。

示例代码片段

import (
	"context"
	"fmt"

	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
)

// 初始化Pod缓存，记录Pod名称与镜像的映射
podImageCache := make(map[string]string)

// 注册Pod事件处理函数
podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
	AddFunc: func(obj interface{}) {
		pod := obj.(*corev1.Pod)
		deployName := getDeploymentFromPod(clientset, pod)
		if deployName == "" {
			return
		}
		image := pod.Spec.Containers[0].Image
		podImageCache[pod.Name] = image
		fmt.Printf("New pod [%s] for deployment [%s] created, image: %s\n", pod.Name, deployName, image)
	},
	DeleteFunc: func(obj interface{}) {
		pod := obj.(*corev1.Pod)
		deployName := getDeploymentFromPod(clientset, pod)
		if deployName == "" {
			return
		}
		oldImage, exists := podImageCache[pod.Name]
		if exists {
			fmt.Printf("Old pod [%s] for deployment [%s] deleted, image: %s\n", pod.Name, deployName, oldImage)
			delete(podImageCache, pod.Name)
		}
	},
})

// 辅助函数：通过Pod的OwnerReferences找到所属Deployment
func getDeploymentFromPod(clientset *kubernetes.Clientset, pod *corev1.Pod) string {
	for _, ref := range pod.OwnerReferences {
		if ref.Kind == "ReplicaSet" {
			rs, err := clientset.AppsV1().ReplicaSets(pod.Namespace).Get(context.TODO(), ref.Name, metav1.GetOptions{})
			if err != nil {
				return ""
			}
			for _, rsRef := range rs.OwnerReferences {
				if rsRef.Kind == "Deployment" {
					return rsRef.Name
				}
			}
		}
	}
	return ""
}

注意事项

• 该方案需要额外处理ReplicaSet的查询，建议通过ReplicaSet的SharedInformer缓存数据，避免直接调用API带来的性能损耗。
• 需要处理Pod被意外删除（非Deployment更新导致）的场景，避免误判镜像变更。

内容的提问来源于stack exchange，提问作者Jananath Banuka