Kubernetes client-go 使用 Informers 监视部署

我正在尝试使用 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 捕获。日志中没有打印任何内容,它进入了一个没有错误消息的崩溃循环。


萧十郎
浏览 203回答 1
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 goroutinefactory.Start(wait.NeverStop)factory.WaitForCacheSync(wait.NeverStop)
打开App,查看更多内容
随时随地看视频慕课网APP

相关分类

Go