go - Kubernetes client-go 使用 Informers 监视部署

Question

我正在尝试使用 client-go informers 来获取部署的副本数。每当自动缩放更改副本数时，我都需要检索它以处理其他一些逻辑。我之前使用的是 Watch() 函数，但在超时和连接断开方面存在一些不一致。

下面的代码显示了一个实现示例:

labelOptions := informers.WithTweakListOptions(func(opts *v1.ListOptions) {
    opts.FieldSelector = "metadata.name=" + name
})
factory := informers.NewSharedInformerFactoryWithOptions(clientSet, 2*time.Second, informers.WithNamespace(namespace), labelOptions)
informer := factory.Apps().V1().Deployments().Informer()

// Using the channels and goroutines below didn't show changes:
stopper := make(chan struct{})
defer close(stopper)
//go func() {
informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc: func(obj interface{}) {
        mObj, ok := obj.(*appsv1.Deployment)
        if !ok {
            panic(spew.Sdump("informer returned invalid type", mObj))
        }

        replicas := int(*mObj.Spec.Replicas)
        logger.Infof("updating replicas to %d", replicas)

        sendUpdates() // use updates elsewhere
    },

    
    UpdateFunc: func(oldObj, newObj interface{}) {
        old, ok := oldObj.(*appsv1.Deployment)
        if !ok {
            panic(spew.Sdump("informer returned invalid type", old))
        }
        newDeployment, ok := newObj.(*appsv1.Deployment)
        if !ok {
            panic(spew.Sdump("informer returned invalid type", newDeployment))
        }
        oldReplicas := int(*old.Spec.Replicas)
        newReplicas := int(*newDeployment.Spec.Replicas)
        if oldReplicas != newReplicas {
            sendUpdates()
        }
    },
})

//factory.Start(wait.NeverStop)
//factory.WaitForCacheSync(wait.NeverStop)
informer.Run(stopper)

当 Kubernetes 自动缩放或我手动更改 Deployments 副本时，我得到 deployment.apps/app scaled 但它没有被 Informer 捕获。日志中没有打印任何内容，它进入了一个没有错误消息的崩溃循环。

我使用了以下资源:

• https://dev.to/davidsbond/go-creating-dynamic-kubernetes-informers-1npi
• What's the best way to get notified when kubernetes Deployments change using the k8s.io/client-go library?
• kubernetes filter objects in Informer

Answer 1

一些注意事项:

• 在调用 informerFactory.Start() 之前，确保直接调用 Informer(informer := factory.Apps().V1().Deployments().Informer()) 或 Start() 不会启动任何东西。
• 使用 goroutine 启动 SharedInformerFactory 是没有意义的，因为 informerFactory.Start() 在内部使用了一个。
• 它还会停止 informerFactory.WaitForCacheSync() 方法的工作，导致它为启动的线人获取错误数据。

labelOptions := informers.WithTweakListOptions(func(opts *v1.ListOptions) {
    opts.FieldSelector = "metadata.name=" + name
})
factory := informers.NewSharedInformerFactoryWithOptions(clientSet, 2*time.Second, informers.WithNamespace(namespace), labelOptions)
informer := factory.Apps().V1().Deployments().Informer()

informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc: func(obj interface{}) {
        mObj, ok := obj.(*appsv1.Deployment)
        if !ok {
            doSomething()
        }
        replicas := int(*mObj.Spec.Replicas)
        doSomething() 
    },

    
    UpdateFunc: func(oldObj, newObj interface{}) {
        old, ok := oldObj.(*appsv1.Deployment)
        if !ok {
            doSomething()
        }
        newDeployment, ok := newObj.(*appsv1.Deployment)
        if !ok {
            doSomething()
        }
        oldReplicas := int(*old.Spec.Replicas)
        newReplicas := int(*newDeployment.Spec.Replicas)
        if oldReplicas != newReplicas {
            doSomething()
        }
    },
})

// Initializes all active informers and starts the internal goroutine
factory.Start(wait.NeverStop)
factory.WaitForCacheSync(wait.NeverStop)