关于devops:clientgo-gin的简单整合六listwatch二关于Rs与Pod以及Deployment的完善

41次阅读

共计 13266 个字符，预计需要花费 34 分钟才能阅读完成。

背景：

后面实现了 client-go gin 的简略整合五 -list-watch deployment 利用, 进一步把 Rs Pod 也实现 list-watch!
后面少写的：

/src/core/deployment_init.go

// 更新
func (depmap *DeploymentMap) Update(dep *v1.Deployment) error {if list, ok := depmap.data.Load(dep.Namespace); ok {for i, range_dep := range list.([]*v1.Deployment) {
            if range_dep.Name == dep.Name {list.([]*v1.Deployment)[i] = dep
                depmap.data.Store(dep.Namespace, list)
            }
        }
        return nil
    }
    return fmt.Errorf("deployment-%s not found", dep.Name)
}

遗记了 Store 数据 ….depmap.data.Store(dep.Namespace, list)！

client-go gin 的简略整合六 -list-watch 二（对于 Rs 与 Pod）

对于 Rs list-watch

/src/core/rs_init.go

package core

import (
    "errors"
    "fmt"
    v1 "k8s.io/api/apps/v1"
    "log"
    "sync"
)

type RSMapStruct struct {Data sync.Map}

func (rsmap *RSMapStruct) Add(rs *v1.ReplicaSet) {
    key := rs.Namespace
    if value, ok := rsmap.Data.Load(key); ok {value = append(value.([]*v1.ReplicaSet), rs)
        rsmap.Data.Store(key, value)
    } else {rsmap.Data.Store(key, []*v1.ReplicaSet{rs})
    }
}
func (rsmap *RSMapStruct) Update(rs *v1.ReplicaSet) error {
    key := rs.Namespace
    if value, ok := rsmap.Data.Load(key); ok {for index, r := range value.([]*v1.ReplicaSet) {
            if r.Name == rs.Name {value.([]*v1.ReplicaSet)[index] = rs
                rsmap.Data.Store(key, value)
                return nil
            }
        }
    }

    return fmt.Errorf("rs-%s not found", rs.Name)
}

func (rsmap *RSMapStruct) Delete(rs *v1.ReplicaSet) {
    key := rs.Namespace
    if value, ok := rsmap.Data.Load(key); ok {for index, r := range value.([]*v1.ReplicaSet) {
            if r.Name == rs.Name {value = append(value.([]*v1.ReplicaSet)[0:index], value.([]*v1.ReplicaSet)[index+1:]...)
                rsmap.Data.Store(key, value)
                return
            }
        }
    }
}

func (rsmap *RSMapStruct) ListByNS(ns string) ([]*v1.ReplicaSet, error) {if list, ok := rsmap.Data.Load(ns); ok {return list.([]*v1.ReplicaSet), nil
    }
    return nil, errors.New("rs record not found")
}

func (rsmap *RSMapStruct) GetRsLabelsByDeployment(deploy *v1.Deployment) ([]map[string]string, error) {rs, err := rsmap.ListByNS(deploy.Namespace)
    if err != nil {return nil, err}
    ret := make([]map[string]string, 0)
    for _, item := range rs {//if item.Annotations["deployment.kubernetes.io/revision"] != deploy.Annotations["deployment.kubernetes.io/revision"] {
        //    continue
        //}
        for _, v := range item.OwnerReferences {
            if v.Name == deploy.Name {ret = append(ret, item.Labels)
                break
            }
        }
    }
    return ret, nil
}

type RSHandler struct {
}

func (rsmap *RSHandler) OnAdd(obj interface{}) {RSMap.Add(obj.(*v1.ReplicaSet))
}
func (rsmap *RSHandler) OnUpdate(oldObj interface{}, newObj interface{}) {err := RSMap.Update(newObj.(*v1.ReplicaSet))
    if err != nil {log.Println(err)
    }
}
func (rsmap *RSHandler) OnDelete(obj interface{}) {RSMap.Delete(obj.(*v1.ReplicaSet))
}

var RSMap *RSMapStruct

func init() {RSMap = &RSMapStruct{}
}

对于 Pod list-watch

先写一个 util 办法(就为了判断两个 label 是否雷同！)：
/src/core/Util.go

func IsValidLabel(m1, m2 map[string]string) bool {
    for key := range m2 {if m2[key] != m1[key] {return false}
    }

    return true
}

根本 copy 自 deployment_init.go:
/src/core/pod_int.go

package core

import (
    "fmt"
    corev1 "k8s.io/api/core/v1"
    "log"
    "sync"
)

type PodMapStruct struct {data sync.Map}

func (podmap *PodMapStruct) Add(pod *corev1.Pod) {
    key := pod.Namespace
    if value, ok := podmap.data.Load(key); ok {value = append(value.([]*corev1.Pod), pod)
        podmap.data.Store(key, value)
    } else {podmap.data.Store(key, []*corev1.Pod{pod})
    }
}
func (podmap *PodMapStruct) Update(pod *corev1.Pod) error {
    key := pod.Namespace
    if value, ok := podmap.data.Load(key); ok {for index, p := range value.([]*corev1.Pod) {
            if p.Name == pod.Name {value.([]*corev1.Pod)[index] = pod
                podmap.data.Store(key, value)
                return nil
            }
        }
    }

    return fmt.Errorf("pod-%s not found", pod.Name)
}
func (podmap *PodMapStruct) Delete(pod *corev1.Pod) {
    key := pod.Namespace
    if value, ok := podmap.data.Load(key); ok {for index, p := range value.([]*corev1.Pod) {
            if p.Name == pod.Name {value = append(value.([]*corev1.Pod)[0:index], value.([]*corev1.Pod)[index+1:]...)
                podmap.data.Store(key, value)
                return
            }
        }
    }
}
func (podmap *PodMapStruct) ListByNS(ns string) ([]*corev1.Pod, error) {

    if ns != "" {if list, ok := podmap.data.Load(ns); ok {return list.([]*corev1.Pod), nil
        }
    }
    return nil, fmt.Errorf("pods not found")
}
func (podmap *PodMapStruct) ListByLabels(ns string, labels []map[string]string) ([]*corev1.Pod, error) {pods, err := podmap.ListByNS(ns)
    if err != nil {return nil, err}
    ret := make([]*corev1.Pod, 0)
    for _, pod := range pods {
        for _, label := range labels {if IsValidLabel(pod.Labels, label) {ret = append(ret, pod)
            }
        }
    }
    return ret, nil
}

type PodHandler struct {
}
var PodMap *PodMapStruct

func init() {PodMap = &PodMapStruct{}
}
func (podmap *PodHandler) OnAdd(obj interface{}) {PodMap.Add(obj.(*corev1.Pod))
}
func (podmap *PodHandler) OnUpdate(oldObj interface{}, newObj interface{}) {err := PodMap.Update(newObj.(*corev1.Pod))
    if err != nil {log.Println(err)
    }
}
func (podmap *PodHandler) OnDelete(obj interface{}) {PodMap.Delete(obj.(*corev1.Pod))
}

deployment informer AddEventHandler

/src/core/deployment_init.go

package core

import (
    "fmt"
    "k8s-demo1/src/lib"
    v1 "k8s.io/api/apps/v1"
    "k8s.io/apimachinery/pkg/util/wait"
    "k8s.io/client-go/informers"
    "log"
    "sync"
)

type DeploymentMap struct {data sync.Map}

func (depmap *DeploymentMap) Add(dep *v1.Deployment) {if list, ok := depmap.data.Load(dep.Namespace); ok {list = append(list.([]*v1.Deployment), dep)
        depmap.data.Store(dep.Namespace, list)
    } else {depmap.data.Store(dep.Namespace, []*v1.Deployment{dep})
    }
}
func (depmap *DeploymentMap) Update(dep *v1.Deployment) error {if list, ok := depmap.data.Load(dep.Namespace); ok {for i, range_dep := range list.([]*v1.Deployment) {
            if range_dep.Name == dep.Name {list.([]*v1.Deployment)[i] = dep
            }
        }
        return nil
    }
    return fmt.Errorf("deployment-%s not found", dep.Name)
}

// 删除
func (depmap *DeploymentMap) Delete(dep *v1.Deployment) {if list, ok := depmap.data.Load(dep.Namespace); ok {for i, range_dep := range list.([]*v1.Deployment) {
            if range_dep.Name == dep.Name {newList := append(list.([]*v1.Deployment)[:i], list.([]*v1.Deployment)[i+1:]...)
                depmap.data.Store(dep.Namespace, newList)
                break
            }
        }
    }
}
func (depmap *DeploymentMap) ListByNS(ns string) ([]*v1.Deployment, error) {if list, ok := depmap.data.Load(ns); ok {return list.([]*v1.Deployment), nil
    }
    return nil, fmt.Errorf("record not found")
}

var DepMap *DeploymentMap

func init() {DepMap = &DeploymentMap{}
}

type DepHandler struct {
}

func (d *DepHandler) OnAdd(obj interface{}) {//fmt.Println(obj.(*v1.Deployment).Name)
    DepMap.Add(obj.(*v1.Deployment))
}
func (d *DepHandler) OnUpdate(oldObj, newObj interface{}) {err := DepMap.Update(newObj.(*v1.Deployment))
    if err != nil {log.Println(err)
    }
}
func (d *DepHandler) OnDelete(obj interface{}) {if d, ok := obj.(*v1.Deployment); ok {DepMap.Delete(d)
    }
}

func InitDeployment() {factory := informers.NewSharedInformerFactory(lib.K8sClient, 0)
    depinformer := factory.Apps().V1().Deployments()
    depinformer.Informer().AddEventHandler(&DepHandler{})
    podInformer := factory.Core().V1().Pods().Informer()
    podInformer.AddEventHandler(&PodHandler{})

    rsInformer := factory.Apps().V1().ReplicaSets().Informer()
    rsInformer.AddEventHandler(&RSHandler{})
    factory.Start(wait.NeverStop)
}

Pod.go

/src/service/Pod.go

package service

import (
    "context"
    "fmt"
    "github.com/gin-gonic/gin"
    "k8s-demo1/src/core"
    . "k8s-demo1/src/lib"
    v1 "k8s.io/api/apps/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

type Pod struct {
    Namespace  string
    Name       string
    Status     string
    Images     string
    NodeName   string
    CreateTime string
    Labels     map[string]string
}

func ListallPod(g *gin.Context) {ns := g.Query("ns")

    //pods, err := K8sClient.CoreV1().Pods(ns).List(context.Background(), metav1.ListOptions{})
    pods, err := core.PodMap.ListByNS(ns)
    if err != nil {g.Error(err)
    }
    ret := make([]*Pod, 0)
    for _, item := range pods {

        ret = append(ret, &Pod{
            Namespace:  item.Namespace,
            Name:       item.Name,
            Status:     string(item.Status.Phase),
            Labels:     item.Labels,
            NodeName:   item.Spec.NodeName,
            Images:     item.Spec.Containers[0].Image,
            CreateTime: item.CreationTimestamp.Format("2006-01-02 15:04:05"),
        })

    }
    g.JSON(200, ret)
    return
}

deployment.go 的批改

deployment.go 也要批改一下（上次没有将 GetPodsByDep 等办法批改为 list-wath!）
/src/service/deployment.go：

package service

import (
    "fmt"
    "github.com/gin-gonic/gin"
    "k8s-demo1/src/core"
    v1 "k8s.io/api/apps/v1"
    "log"
)

type Deployment struct {
    Namespace           string
    Name                string
    Replicas            int32
    AvailableReplicas   int32
    UnavailableReplicas int32
    Images              string
    CreateTime          string
    Labels              map[string]string
    Pods                []*Pod}

func ListDeployment(g *gin.Context) {ns := g.Query("ns")
    deplist, err := core.DepMap.ListByNS(ns)
    //dps, err := K8sClient.AppsV1().Deployments(ns).List(context.Background(), metav1.ListOptions{})
    if err != nil {g.Error(err)
    }
    ret := make([]*Deployment, 0)
    for _, item := range deplist {
        ret = append(ret, &Deployment{
            Namespace:           item.Namespace,
            Name:                item.Name,
            Replicas:            item.Status.Replicas,
            AvailableReplicas:   item.Status.AvailableReplicas,
            UnavailableReplicas: item.Status.UnavailableReplicas,
            Images:              item.Spec.Template.Spec.Containers[0].Image,
            Labels:              item.GetLabels(),
            Pods:                GetPodsByDep(ns, *item),
            CreateTime:          item.CreationTimestamp.Format("2006-01-02 15:03:04"),
        })

    }
    g.JSON(200, ret)
    return
}

func GetLabels(m map[string]string) string {
    labels := ""
    // aa=xxx,xxx=xx
    for k, v := range m {
        if labels != "" {labels += ","}
        labels += fmt.Sprintf("%s=%s", k, v)
    }
    return labels
}
func GetPodsByDep(ns string, dep v1.Deployment) []*Pod {rsLabelsMap, err := core.RSMap.GetRsLabelsByDeployment(&dep)
    if err != nil {log.Fatal(err)
    }
    pods, err := core.PodMap.ListByLabels(dep.Namespace, rsLabelsMap)
    if err != nil {log.Fatal(err)
    }
    ret := make([]*Pod, 0)
    for _, pod := range pods {
        ret = append(ret, &Pod{
            Name:       pod.Name,
            Namespace:  pod.Namespace,
            Images:     pod.Spec.Containers[0].Image,
            NodeName:   pod.Spec.NodeName,
            Labels:     pod.Labels,
            Status:     string(pod.Status.Phase),
            CreateTime: pod.CreationTimestamp.Format("2006-01-02 15:04:05"),
        })
    }
    return ret
}

运行 main.go

运行 main.go 看是否能拜访 pod 列表？
http://127.0.0.1:8080/pods?ns…

deployments 下的坑 ……

运行 main.go 也顺便看了一眼 deployment 是否失常：
http://127.0.0.1:8080/deploym…

what……deployment 的 pod 串了！想一下也是的：GetRsLabelsByDeployment 的时候并没有判断 rc 与 pod 的关系，pod 的 label 判断还是不准确的，没有想到好的方法，偷了一个懒：
/src/core/rs_int.go

package core

import (
    "errors"
    "fmt"
    v1 "k8s.io/api/apps/v1"
    "log"
    "sync"
)

type RSMapStruct struct {Data sync.Map}

func (rsmap *RSMapStruct) Add(rs *v1.ReplicaSet) {
    key := rs.Namespace
    if value, ok := rsmap.Data.Load(key); ok {value = append(value.([]*v1.ReplicaSet), rs)
        rsmap.Data.Store(key, value)
    } else {rsmap.Data.Store(key, []*v1.ReplicaSet{rs})
    }
}
func (rsmap *RSMapStruct) Update(rs *v1.ReplicaSet) error {
    key := rs.Namespace
    if value, ok := rsmap.Data.Load(key); ok {for index, r := range value.([]*v1.ReplicaSet) {
            if r.Name == rs.Name {value.([]*v1.ReplicaSet)[index] = rs
                rsmap.Data.Store(key, value)
                return nil
            }
        }
    }

    return fmt.Errorf("rs-%s not found", rs.Name)
}

func (rsmap *RSMapStruct) Delete(rs *v1.ReplicaSet) {
    key := rs.Namespace
    if value, ok := rsmap.Data.Load(key); ok {for index, r := range value.([]*v1.ReplicaSet) {
            if r.Name == rs.Name {value = append(value.([]*v1.ReplicaSet)[0:index], value.([]*v1.ReplicaSet)[index+1:]...)
                rsmap.Data.Store(key, value)
                return
            }
        }
    }
}

func (rsmap *RSMapStruct) ListByNS(ns string) ([]*v1.ReplicaSet, error) {if list, ok := rsmap.Data.Load(ns); ok {return list.([]*v1.ReplicaSet), nil
    }
    return nil, errors.New("rs record not found")
}

func (rsmap *RSMapStruct) GetRsLabelsByDeployment(deploy *v1.Deployment) ([]map[string]string, error) {rs, err := rsmap.ListByNS(deploy.Namespace)
    if err != nil {return nil, err}
    ret := make([]map[string]string, 0)
    for _, item := range rs {//if item.Annotations["deployment.kubernetes.io/revision"] != deploy.Annotations["deployment.kubernetes.io/revision"] {
        //    continue
        //}
        for _, v := range item.OwnerReferences {
            if v.Name == deploy.Name {ret = append(ret, item.Labels)
                break
            }
        }
    }
    return ret, nil
}
func (rsmap *RSMapStruct) GetRsLabelsByDeploymentname(deploy *v1.Deployment) string {rs, err := rsmap.ListByNS(deploy.Namespace)
    if err != nil {fmt.Println(err)
    }
    for _, item := range rs {//if item.Annotations["deployment.kubernetes.io/revision"] != deploy.Annotations["deployment.kubernetes.io/revision"] {
        //    continue
        //}
        for _, v := range item.OwnerReferences {
            if v.Name == deploy.Name {return item.Name}
        }
    }
    return ""
}

type RSHandler struct {
}

func (rsmap *RSHandler) OnAdd(obj interface{}) {RSMap.Add(obj.(*v1.ReplicaSet))
}
func (rsmap *RSHandler) OnUpdate(oldObj interface{}, newObj interface{}) {err := RSMap.Update(newObj.(*v1.ReplicaSet))
    if err != nil {log.Println(err)
    }
}
func (rsmap *RSHandler) OnDelete(obj interface{}) {RSMap.Delete(obj.(*v1.ReplicaSet))
}

var RSMap *RSMapStruct

func init() {RSMap = &RSMapStruct{}
}

模拟 GetRsLabelsByDeployment 写了一个 GetRsLabelsByDeploymentname 办法， 恩返回 rc 的 name 等于 pod 的 OwnerReferences 字段中的 name 字段!

/src/service/deployment.go

package service

import (
    "fmt"
    "github.com/gin-gonic/gin"
    "k8s-demo1/src/core"
    v1 "k8s.io/api/apps/v1"
    "log"
)

type Deployment struct {
    Namespace           string
    Name                string
    Replicas            int32
    AvailableReplicas   int32
    UnavailableReplicas int32
    Images              string
    CreateTime          string
    Labels              map[string]string
    Pods                []*Pod}

func ListDeployment(g *gin.Context) {ns := g.Query("ns")
    deplist, _ := core.DepMap.ListByNS(ns)
    //dps, err := K8sClient.AppsV1().Deployments(ns).List(context.Background(), metav1.ListOptions{})
    //if err != nil {//    g.Error(err)
    //}
    ret := make([]*Deployment, 0)
    for _, item := range deplist {
        ret = append(ret, &Deployment{
            Namespace:           item.Namespace,
            Name:                item.Name,
            Replicas:            item.Status.Replicas,
            AvailableReplicas:   item.Status.AvailableReplicas,
            UnavailableReplicas: item.Status.UnavailableReplicas,
            Images:              item.Spec.Template.Spec.Containers[0].Image,
            Labels:              item.GetLabels(),
            Pods:                GetPodsByDep(*item),
            CreateTime:          item.CreationTimestamp.Format("2006-01-02 15:03:04"),
        })

    }
    g.JSON(200, ret)
    return
}

func GetLabels(m map[string]string) string {
    labels := ""
    // aa=xxx,xxx=xx
    for k, v := range m {
        if labels != "" {labels += ","}
        labels += fmt.Sprintf("%s=%s", k, v)
    }
    return labels
}
func GetPodsByDep(dep v1.Deployment) []*Pod {rsLabelsMap, err := core.RSMap.GetRsLabelsByDeployment(&dep)
    if err != nil {log.Fatal(err)
    }

    pods, err := core.PodMap.ListByRsLabels(dep.Namespace, rsLabelsMap)
    if err != nil {log.Fatal(err)
    }
    ret := make([]*Pod, 0)

    for _, pod := range pods {if core.RSMap.GetRsLabelsByDeploymentname(&dep) == pod.OwnerReferences[0].Name {
            ret = append(ret, &Pod{
                Name:       pod.Name,
                Namespace:  pod.Namespace,
                Images:     pod.Spec.Containers[0].Image,
                NodeName:   pod.Spec.NodeName,
                Labels:     pod.Labels,
                Status:     string(pod.Status.Phase),
                CreateTime: pod.CreationTimestamp.Format("2006-01-02 15:04:05"),
            })
        }
    }
    return ret
}

运行 mai.go，拜访 http://127.0.0.1:8080/deploym…

算是精准匹配了，还不晓得会不会有什么坑 ……

总结：

总结一下这过程学到的：

  1. deployment rs pod 之间的关系
  2. OwnerReferences 的匹配
  3. 上面还是持续 crud 吧 …… 这个中央让我了解的有点长 list-watch

正文完
devops
发表至： devops
2022-06-10
 0
关于devops:我在京东做研发-揭秘支撑京东万人规模技术人员协作的行云DevOps平台
关于devops:企业如何通过DevOps在数字化浪潮中C位出道
关于devops:项目管理全新升级博云DevOps产品-V34正式发布
关于devops:研发效能谈敏捷开发团队转型中的协作化与自动化