informer cache自定义索引

informer cache默认通过namespace/name作为key把对象保存到map中。
条件查询时一般通过labels.Selector来过滤，但这需要遍历所有元素，informer cache可以类似于MySQL那样建立索引，来提高查询速度。

// map根据指定的key来给对象分类
// IndexFunc knows how to compute the set of indexed values for an object.
type IndexFunc func(obj interface{}) ([]string, error)

package main

import (
	"fmt"
	"os"
	"os/signal"
	"sync"
	"syscall"

	core_v1 "k8s.io/api/core/v1"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/clientcmd"
	klog "k8s.io/klog/v2"
)

func SetupSignalHandler() chan struct{} {
	stop := make(chan struct{})
	c := make(chan os.Signal, 2)
	signal.Notify(c, []os.Signal{os.Interrupt, syscall.SIGTERM}...)
	go func() {
		<-c
		close(c)
		close(stop)
	}()

	return stop
}

func nodeLabelIndexFunc(obj interface{}) ([]string, error) {
	if node, ok := obj.(*core_v1.Node); !ok {
		return nil, fmt.Errorf("type is not node")
	} else {
		indexKeys := make([]string, len(node.Labels))
		for k, v := range node.Labels {
			indexKeys = append(indexKeys, fmt.Sprintf("%s=%s", k, v))
		}
		return indexKeys, nil
	}
}

func addListWatchCfgAndClient(stopCh chan struct{}, wg *sync.WaitGroup) {
	wg.Add(1)
	defer wg.Done()

	cfg, err := clientcmd.BuildConfigFromFlags("", "/root/.kube/config")
	if err != nil {
		klog.Fatalf("Error building kubeconfig: %s", err.Error())
	}

	kubeClient, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		klog.Fatalf("Error building kubernetes clientset: %s", err.Error())
	}

	informerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
	nodeInformer := informerFactory.Core().V1().Nodes().Informer()
	nodeLabelIndexer := cache.Indexers{"label": nodeLabelIndexFunc}
	if err = nodeInformer.AddIndexers(nodeLabelIndexer); err != nil {
		klog.Fatalf("add index for node informer failed, err is %v", err)
	}
	hasSynced := nodeInformer.HasSynced
	informerFactory.Core().V1().Nodes().Lister()

	informerFactory.Start(stopCh)
	defer informerFactory.Shutdown()

	if ok := cache.WaitForCacheSync(stopCh, hasSynced); !ok {
		klog.Fatalf("failed to wait for caches to sync")
	}

	if nodes, err := nodeInformer.GetIndexer().ByIndex("label", "test=a"); err != nil {
		klog.Errorf("get node by label test=a failed, err is %v", err)
	} else {
		nodeNames := make([]string, len(nodes))
		for i, node := range nodes {
			nodeNames[i] = node.(*core_v1.Node).Name
		}
		klog.Infof("get node by label, node name is %v", nodeNames)
	}
}

func main() {
	stopCh := SetupSignalHandler()
	wg := sync.WaitGroup{}
	addListWatchCfgAndClient(stopCh, &wg)
	<-stopCh
	wg.Wait()
}