求助：Kubernetes中Go应用Informer因信号量阻塞问题排查

问题现象 #

我们在Kubernetes Pod中运行的Go应用（包含多个Informer）出现异常，通过goroutine dump分析发现，某个goroutine长期阻塞在信号量获取环节，推测是资源竞争或I/O延迟导致。同时，自定义资源Informer出现监听失败错误，日志显示与Kubernetes API Server的变更同步存在问题。

相关代码 #

Pod Informer实现 #

func CreateInformer(label string, podname PodName) {
    kube, err := kube.NewKubeClient()

    if err != nil {
        log.Error().Err(err).Msg("Error while getting kube client")
        return
    }
    informerOption1 := informers.WithTweakListOptions(func(opts *metav1.ListOptions) {
        opts.FieldSelector = "status.phase=Running"
        opts.LabelSelector = fmt.Sprintf("app=%s", label)
    })
    informerFactory1 := informers.NewSharedInformerFactoryWithOptions(kube, 0, informerOption1)
    podInformer1 := informerFactory1.Core().V1().Pods()
    podInformer1.Informer().AddEventHandler(kubecache.ResourceEventHandlerFuncs{

        AddFunc: func(new interface{}) {
            if pod, ok := new.(*corev1.Pod); ok {
                log.Debug().Msg("Pod " + pod.Name + " Created")
                UpdateCacheDetails(podInformer1, podname)
            }
        },
        UpdateFunc: func(old, new interface{}) {
            if pod, ok := new.(*corev1.Pod); ok {
                log.Debug().Msg("Pod " + pod.Name + " updated")
                UpdateCacheDetails(podInformer1, podname)
            }
        },
        DeleteFunc: func(old interface{}) {
            if pod, ok := old.(*corev1.Pod); ok {
                log.Debug().Msg("Pod " + pod.Name + " deleted")
                UpdateCacheDetails(podInformer1, podname)
            }
        },
    })
    go func() {
        informerFactory1.Start(wait.NeverStop)
    }()
    go func() {
        informerFactory1.WaitForCacheSync(wait.NeverStop)
    }()

}

CRD Informer实现 #

func CreateCRDInformer() {
    _, kubeconfig, err := kube.NewKubeClientWithConfig()
    if err != nil {
        log.Error().Err(err).Msg("Error while getting kube client")
        return
    }
    dynamicClient, err := dynamic.NewForConfig(kubeconfig)
    if err != nil {
        log.Error().Err(err).Msg("Error while getting dynamic client")
        return
    }
    resource := schema.GroupVersionResource{Group: "stable.com", Version: "v1", Resource: "workloads"}
    factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient,
        time.Second, "", nil)
    informer := factory.ForResource(resource).Informer()
    crdEventHandler := func(obj interface{}) {
        lister := dynamiclister.New(informer.GetIndexer(), resource)
        workloads, err := lister.List(labels.Everything())
        if err != nil {
            log.Error().Err(err).Msg("Error while getting workload data")
            return
        }
        log.Info().Msg("UpdateWorkloadCache : Create/Delete")
        UpdateWorkloadCache(workloads)
    }
    informer.AddEventHandler(kubecache.ResourceEventHandlerFuncs{
        AddFunc: crdEventHandler,
        UpdateFunc: func(oldObj, newObj interface{}) {
            oldObjJson, err := json.Marshal(oldObj)
            if err != nil {
                log.Err(err).Msg("oldObjJson marshall failed")
            }
            newObjJson, err := json.Marshal(newObj)
            if err != nil {
                log.Err(err).Msg("oldObjJson marshall failed")
            }

            opts := jsondiff.DefaultConsoleOptions()
            opts.PrintTypes = false
            result, _ := jsondiff.Compare(newObjJson, oldObjJson, &opts)
            if result != jsondiff.FullMatch {
                log.Info().Msg("UpdateWorkloadCache : Update")
                lister := dynamiclister.New(informer.GetIndexer(), resource)
                workloads, err := lister.List(labels.Everything())
                if err != nil {
                    log.Error().Err(err).Msg("Error while getting workload data")
                    return
                }
                UpdateWorkloadCache(workloads)
            }
        },
        DeleteFunc: crdEventHandler,
    })
    go informer.Run(wait.NeverStop)

    // Wait for the informer cache to be synced
    if !kubecache.WaitForCacheSync(wait.NeverStop, informer.HasSynced) {
        log.Error().Err(err).Msg("Error syncing informer cache")
        return
    }
}

错误日志 #

E0627 09:19:08.998638 1 reflector.go:140] pkg/mod/k8s.io/client-go@v0.26.1/tools/cache/reflector.go:169: Failed to watch *unstructured.Unstructured: unknown.

排查与解决方案 #

针对CRD监听失败（unknown错误） #

• 验证CRD资源定义：确认stable.com/v1/workloads CRD已正确安装到集群，检查CRD的spec.names.plural是否为workloads（必须与代码中Resource字段完全一致）。
• 检查API Server权限：应用所在ServiceAccount需具备workloads.stable.com资源的list和watch权限，可通过kubectl auth can-i list workloads.stable.com -n <namespace> --as=system:serviceaccount:<namespace>:<sa-name>验证。
• 修复DynamicClient配置：代码中NewFilteredDynamicSharedInformerFactory的namespace参数传了空字符串，若CRD仅存在于特定namespace，需指定对应namespace；若为集群级资源，保持空字符串但需确认权限覆盖集群范围。
• 版本兼容性：检查client-go版本（v0.26.1）与集群Kubernetes版本是否兼容，避免因API版本不匹配导致的未知资源错误。

针对goroutine信号量阻塞问题 #

• 检查缓存更新函数实现：UpdateCacheDetails和UpdateWorkloadCache是Informer事件回调的核心逻辑，需确认是否存在未加锁的共享资源修改或同步I/O操作（如数据库调用、HTTP请求），这些会导致回调阻塞，进而占用Informer的worker池，引发信号量获取等待。建议将耗时操作异步化，比如在回调中启动新goroutine处理缓存更新，避免阻塞Informer的事件处理循环。
• 优化Informer Worker池配置：默认Informer的worker数量有限，若事件量较大，可通过WithWorkers选项调整：

  // 针对Pod Informer工厂
  informerFactory1 := informers.NewSharedInformerFactoryWithOptions(kube, 0, informerOption1, informers.WithWorkers(10))
  // 针对CRD Informer
  factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, time.Second, "", nil, dynamicinformer.WithWorkers(10))
  

• 避免重复全量List操作：CRD的UpdateFunc中每次都调用lister.List(labels.Everything())全量拉取数据，若资源数量大，会增加延迟和内存占用，可考虑仅传递变更的单个对象到UpdateWorkloadCache，减少不必要的全量查询。
• 日志输出优化：避免在回调中频繁打印Debug级日志，大量日志IO可能阻塞goroutine，建议调整日志级别或异步输出日志。

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