K8s list-watch 机制和 Informer 模块

type DepHandler struct{}
func (this *DepHandler) OnAdd(obj interface{}) {}
func (this *DepHandler) OnUpdate(oldObj, newObj interface{}) {
	if dep, ok := newObj.(*v1.Deployment); ok {
		fmt.Println(dep.Name)
	}
}
func (this *DepHandler) OnDelete(obj interface{}) {}

func main() {
	_, c := cache.NewInformer(
		// 监听 default 命名空间中 deployment 的变化
		cache.NewListWatchFromClient(K8sClient.AppsV1().RESTClient(),
			"deployments", "default", fields.Everything()),
		&v1.Deployment{},
		0,				// 重新同步时间
		&DepHandler{},	// 实现类
	)
	c.Run(wait.NeverStop)
	select {}
}

informerFactory := informers.NewSharedInformerFactory(K8sClient, 0)

// 构建一个 deployment informer
depInformer := informerFactory.Apps().V1().Deployments()
depInformer.Informer().AddEventHandler(&DepHandler{})

informerFactory.Start(wait.NeverStop)
select {}

// 全局对象，存储所有deployments
var DepMapImpl *DeploymentMap

func init() {
	DepMapImpl = &DeploymentMap{Data: new(sync.Map)}
}

type DeploymentMap struct {
  Data *sync.Map	// key:namespace value:[]*v1.Deployments
}

// 添加
func (this *DeploymentMap) Add(deployment *v1.Deployment) {
	if depList, ok := this.Data.Load(deployment.Namespace); ok {
		depList = append(depList.([]*v1.Deployment), deployment)
		this.Data.Store(deployment.Namespace, depList)
	} else {
		this.Data.Store(deployment.Namespace, []*v1.Deployment{deployment})
	}
}

// 获取列表
func (this *DeploymentMap) ListByNs(namespace string) ([]*v1.Deployment, error) {
	if depList, ok := this.Data.Load(namespace); ok {
		return depList.([]*v1.Deployment), nil
	}

	return nil, fmt.Errorf("record not found")
}

// 更新
func (this *DeploymentMap) Update(deployment *v1.Deployment) error {
	if depList, ok := this.Data.Load(deployment.Namespace); ok {
		depList := depList.([]*v1.Deployment)
		for i, dep := range depList {
			if dep.Name == deployment.Name {
				depList[i] = deployment
				break
			}
		}
		return nil
	}

	return fmt.Errorf("deployment [%s] not found", deployment.Name)
}

// 删除
func (this *DeploymentMap) Delete(deployment *v1.Deployment) {
	if depList, ok := this.Data.Load(deployment.Namespace); ok {
		depList := depList.([]*v1.Deployment)
		for i, dep := range depList {
			if dep.Name == deployment.Name {
				newDepList := append(depList[:i], depList[i+1:]...)
				this.Data.Store(deployment.Namespace, newDepList)
				break
			}
		}
	}
}

// informer实现
type DepHandler struct{}
func (this *DepHandler) OnAdd(obj interface{}) {
	DepMapImpl.Add(obj.(*v1.Deployment))
}
func (this *DepHandler) OnUpdate(oldObj, newObj interface{}) {
	err := DepMapImpl.Update(newObj.(*v1.Deployment))
	if err != nil {
		log.Println(err)
	}
}
func (this *DepHandler) OnDelete(obj interface{}) {
	DepMapImpl.Delete(obj.(*v1.Deployment))
}

// 执行监听
func InitDeployments() {
	informerFactory := informers.NewSharedInformerFactory(K8sClient, 0)

	depInformer := informerFactory.Apps().V1().Deployments()
	depInformer.Informer().AddEventHandler(&DepHandler{})

	informerFactory.Start(wait.NeverStop)
}

// 从本地缓存中取出所有的rs
rsList, err := RSMapImpl.ListByNs(namespace)

// 获取 labels
labels, err := GetListWatchRsLabelByDeployment(deployment, rsList)

// list-watch方式 根据deployment获取当前ReplicaSet的标签
func GetListWatchRsLabelByDeployment(deployment *v1.Deployment, rsList []*v1.ReplicaSet) (map[string]string, error) {
	for _, rs := range rsList {
		if IsCurrentRsByDeployment(rs, deployment) {
			selector, err := metaV1.LabelSelectorAsMap(rs.Spec.Selector)
			if err != nil {
				return nil, err
			}
			return selector, nil
		}
	}

	return nil, nil
}

// 判断rs是否对应当前deployment
func IsCurrentRsByDeployment(set *v1.ReplicaSet, deployment *v1.Deployment) bool {
	if set.ObjectMeta.Annotations["deployment.kubernetes.io/revision"] != deployment.ObjectMeta.Annotations["deployment.kubernetes.io/revision"] {
		return false
	}

	for _, rf := range set.OwnerReferences {
		if rf.Kind == "Deployment" && rf.Name == deployment.Name {
			return true
		}
	}

	return false
}

// 根据标签获取Pod列表
func (this *PodMap) ListByLabels(ns string, labels map[string]string) ([]*v1.Pod, error) {
	ret := make([]*v1.Pod, 0)
	if podList, ok := this.Data.Load(ns); ok {
		podList := podList.([]*v1.Pod)
		for _, p := range podList {
			// 判断标签完全匹配
			if reflect.DeepEqual(p.Labels, labels) {
				ret = append(ret, p)
			}
		}

		return ret, nil
	}

	return nil, fmt.Errorf("pods not found")
}

// EventMapImpl 全局对象，存储所有Event
var EventMapImpl *EventMap

func init() {
	EventMapImpl = &EventMap{Data: new(sync.Map)}
}

type EventMap struct {
	Data *sync.Map // key:namespace_kind_name value: *v1.Event
}

func (this *EventMap) GetKey(event *v1.Event) string {
	key := fmt.Sprintf("%s_%s_%s", event.Namespace, event.InvolvedObject.Kind, event.InvolvedObject.Name)
	return key
}

// Add 添加
func (this *EventMap) Add(event *v1.Event) {
	EventMapImpl.Data.Store(this.GetKey(event), event)
}

// Delete 删除
func (this *EventMap) Delete(event *v1.Event) {
	EventMapImpl.Data.Delete(this.GetKey(event))
}

// 获取最新一条event message
func (this *EventMap) GetMessage(ns string, kind string, name string) string {
	key := fmt.Sprintf("%s_%s_%s", ns, kind, name)
	if v, ok := this.Data.Load(key); ok {
		return v.(*v1.Event).Message
	}

	return ""
}

// EventHandler informer实现
type EventHandler struct{}

func (this *EventHandler) OnAdd(obj interface{}) {
	EventMapImpl.Add(obj.(*v1.Event))
}
func (this *EventHandler) OnUpdate(oldObj, newObj interface{}) {
	EventMapImpl.Add(newObj.(*v1.Event))
}
func (this *EventHandler) OnDelete(obj interface{}) {
	EventMapImpl.Delete(obj.(*v1.Event))
}

// 评估Pod是否就绪
func GetPodIsReady(pod *coreV1.Pod) bool {
	for _, condition := range pod.Status.Conditions {
		if condition.Type == "ContainersReady" && condition.Status != "True" {
			return false
		}
	}
	
	for _, rg := range pod.Spec.ReadinessGates {
		for _, condition := range pod.Status.Conditions {
			if condition.Type == rg.ConditionType && condition.Status != "True" {
				return false
			}
		}
	}

	return true
}

// 获取pods DTO 把原生的 pod 对象转换为自己的实体对象
func GetPodsByLabels(ns string, labels []map[string]string) (pods []*model.PodModel) {
	podList, err := PodMapImpl.ListByLabels(ns, labels)
	lib.CheckError(err)

	pods = make([]*model.PodModel, len(podList))

	for i, pod := range podList {
		pods[i] = &model.PodModel{
			Name:      pod.Name,
			NodeName:  pod.Spec.NodeName,
			Images:    GetPodImages(pod.Spec.Containers),
			Phase:     string(pod.Status.Phase),
			IsReady:   GetPodIsReady(pod),
			Message:   EventMapImpl.GetMessage(pod.Namespace, "Pod", pod.Name),
			CreatedAt: pod.CreationTimestamp.Format("2006-01-02 15:04:05"),
		}
	}

	return
}