Kubernetes 中 Informer 的使用

原文链接：https://mp.weixin.qq.com/s?__biz=MzU4MjQ0MTU4Ng==&mid=2247485580&idx=1&sn=7392dbadff9ab450d93c5dd0449dace5&chksm=fdb90791cace8e871b1dcdd00be21f16a23504f48634f4fb7d3c87a552b3f8ce2765862fb4e9&mpshare=1&scene=1&srcid=0803nJYcFq4iqUIwBQvoRpEW&sharer_sharetime=1596412436655&sharer_shareid=0451566035f23b154ae335ab62d3ee6e&key=7f54b3443b6830333a899f57c837cf0b2f4a3d3c0e20afd7e068ee8b3fafa2cd56d0d7f8b0d130e1f1f19d0181323df9ac59911c3a419cdab13d54706c6324c22cb21fa8bcccb120a903304cd114749d&ascene=1&uin=MTA3NzMyNzIxMA%3D%3D&devicetype=Windows+10+x64&version=62090529&lang=zh_CN&exportkey=A5%2FAm5993DMViljNUoRHTOI%3D&pass_ticket=%2BCfC3DWFa74MKq4qShjYN4qHAtR49lQsIe5FKJmx3QqGKiKCwfe%2B37k33drGSRjx

demo

package main

import (
  "flag"
  "fmt"
  "path/filepath"
  "time"

  v1 "k8s.io/api/apps/v1"
  "k8s.io/apimachinery/pkg/labels"
  "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"
  "k8s.io/client-go/util/homedir"
)

func main() {
  var err error
  var config *rest.Config

  var kubeconfig *string

  if home := homedir.HomeDir(); home != "" {
    kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "[可选] kubeconfig 绝对路径")
  } else {
    kubeconfig = flag.String("kubeconfig", "", "kubeconfig 绝对路径")
  }
  // 初始化 rest.Config 对象
  if config, err = rest.InClusterConfig(); err != nil {
    if config, err = clientcmd.BuildConfigFromFlags("", *kubeconfig); err != nil {
      panic(err.Error())
    }
  }
  // 创建 Clientset 对象
  clientset, err := kubernetes.NewForConfig(config)
  if err != nil {
    panic(err.Error())
  }

  // 初始化 informer factory（为了测试方便这里设置每30s重新 List 一次）
  informerFactory := informers.NewSharedInformerFactory(clientset, time.Second*30)
  // 对 Deployment 监听
  deployInformer := informerFactory.Apps().V1().Deployments()
  // 创建 Informer（相当于注册到工厂中去，这样下面启动的时候就会去 List & Watch 对应的资源）
  informer := deployInformer.Informer()
  // 创建 Lister
  deployLister := deployInformer.Lister()
  // 注册事件处理程序
  informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc:    onAdd,
    UpdateFunc: onUpdate,
    DeleteFunc: onDelete,
  })

  stopper := make(chan struct{})
  defer close(stopper)

  // 启动 informer，List & Watch
  informerFactory.Start(stopper)
  // 等待所有启动的 Informer 的缓存被同步
  informerFactory.WaitForCacheSync(stopper)

  // 从本地缓存中获取 default 中的所有 deployment 列表
  deployments, err := deployLister.Deployments("default").List(labels.Everything())
  if err != nil {
    panic(err)
  }
  for idx, deploy := range deployments {
    fmt.Printf("%d -> %s
", idx+1, deploy.Name)
  }
  <-stopper
}

func onAdd(obj interface{}) {
  deploy := obj.(*v1.Deployment)
  fmt.Println("add a deployment:", deploy.Name)
}

func onUpdate(old, new interface{}) {
  oldDeploy := old.(*v1.Deployment)
  newDeploy := new.(*v1.Deployment)
  fmt.Println("update deployment:", oldDeploy.Name, newDeploy.Name)
}

func onDelete(obj interface{}) {
  deploy := obj.(*v1.Deployment)
  fmt.Println("delete a deployment:", deploy.Name)
}

运行效果：

$ go run main.go
add a deployment: dingtalk-hook
add a deployment: spin-orca
add a deployment: argocd-server
add a deployment: istio-egressgateway
add a deployment: vault-agent-injector
add a deployment: rook-ceph-osd-0
add a deployment: rook-ceph-osd-2
add a deployment: code-server
......
1 -> nginx
2 -> helloworld-v1
3 -> productpage-v1
4 -> details-v1
......