Kubernetes的 client-go

[wangjc0216]

ref：
https://mp.weixin.qq.com/s/BY5z9JlqSibGwPlM3T2DTQ
https://blog.crazytaxii.com/posts/k8s_client_go_informer1/

client主要分为以下几类：

• *kubernetes.Clientset : 用于固有的api请求
• *dynamic.DynamicClient ：用于自定义api请求
• *discovery.DiscoveryClient： 用于服务发现，比如kubectl api-resources命令

[wangjc0216]

Informer主要有两个方法： List 和 Watch ；
List是一次拉取全量数据，Watch是通过http的chunk来拉取增量数据，并保持一个http长连接。

[wangjc0216]

DeltaFifo中

type DeltaFIFO struct {
	// lock/cond protects access to 'items' and 'queue'.
	lock sync.RWMutex
	cond sync.Cond

	// `items` maps a key to a Deltas.
	// Each such Deltas has at least one Delta.
	items map[string]Deltas

	// `queue` maintains FIFO order of keys for consumption in Pop().
	// There are no duplicates in `queue`.
	// A key is in `queue` if and only if it is in `items`.
	queue []string
...

需要注意的两个字段，一个queue slice，一个items map。queue用于保证顺序，items中的value是Deltas，本质也是一个slice。

DeltaFifo会按照对象进行分桶，并把同一个对象的变更操作放在一个Deltas中。如果该对象第一次添加，则会append queue，否则直接更改Deltas并进行duplicate。比如把服务副本scale成 +1 后马上又-1，这就可以起到去重的效果。

[wangjc0216]

Reflector 反应器，反射器。 反光装置，反光灯
reflect 反射，物理中光的反射
反射类似于投影，本质是对实体的一种投射或映射。

Reflector将实体进行映射。

	df := cache.NewDeltaFIFOWithOptions(cache.DeltaFIFOOptions{KeyFunction: podkeyFunc})
	rf := cache.NewReflector(lwc, &v1.Pod{}, df, time.Second*0)
	rsCH := make(chan struct{})
	go func() {
		rf.Run(rsCH)
	}()

Reflector会把ListWatch、Store注册进来，不断Watch和Backoff重试 List来监听资源，把变更记录到Store中。

[wangjc0216]

Informer有4中类型,Informer,IndexInformer,SharedInformer,SharedIndexInformer,下面有一段代码示例：

func Example(configPath, namespace, resource string) error {
	cfg, err := clientcmd.BuildConfigFromFlags("", configPath)
	if err != nil {
		return err
	}
	cs, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		return err
	}

	lwc := cache.NewListWatchFromClient(cs.CoreV1().RESTClient(), resource, namespace, fields.Everything())
	podkeyFunc := func(obj interface{}) (string, error) {
		return obj.(*v1.Pod).Name, nil
	}
	//Reflector
	df := cache.NewDeltaFIFOWithOptions(cache.DeltaFIFOOptions{KeyFunction: podkeyFunc})
	rf := cache.NewReflector(lwc, &v1.Pod{}, df, time.Second*0)
	rsCH := make(chan struct{})
	go func() {
		rf.Run(rsCH)
	}()

	//Informer
	store, controller := cache.NewInformer(lwc, &v1.ConfigMap{}, 0, &CmdHandler{})
	go controller.Run(wait.NeverStop)
	_ = store

	//IndexInformer
	myIndexer := cache.Indexers{"app": LabelsIndexFunc}
	i, c := cache.NewIndexerInformer(lwc, &v1.Pod{}, 0, &CmdHandler{}, myIndexer)
	go c.Run(wait.NeverStop)
	_ = i

	//sharedInformer
	shareInformer := cache.NewSharedInformer(lwc, &v1.ConfigMap{}, 0)
	shareInformer.AddEventHandler(&CmdHandler{})
	shareInformer.AddEventHandler(&CmdHandler{})
	shareInformer.GetStore()
	shareInformer.GetController()
	shareInformer.GetController().Run(wait.NeverStop)

	//shareIndexInformer
	shareIndexInformer := cache.NewSharedIndexInformer(lwc, &v1.ConfigMap{}, 0, myIndexer)
	shareIndexInformer.AddEventHandler(&CmdHandler{})
	shareIndexInformer.AddIndexers(myIndexer)

	factory := informers.NewSharedInformerFactoryWithOptions(cs, 0, informers.WithNamespace("default"))

	cmGVR := schema.GroupVersionResource{
		Group:    "",
		Version:  "v1",
		Resource: "configmaps",
	}

	genericInformer, err := factory.ForResource(cmGVR)
	informer := genericInformer.Informer()

	go informer.Run(wait.NeverStop)
	return nil

}

[wangjc0216]

[wangjc0216]

当创建一个SharedInformer()的时候，可以传入indexers和EventHandler。
informer Run的时候，底层会生成controller，并传入indexer做为store；controller会生成Reflector，对store封装成DeltaFifo；Reflector会不断ListWatch。