背景:

完成了client-go gin的简单整合七-继续完善,准备整合一下前面的代码(当然了关于上一节的坑还是先留着……没有想好怎么搞:https://www.yuque.com/duiniwukenaihe/hg6ymd/ipd4bf#R0XxW)。这里以service为例也实现一下list watch……

list watch实现service list:

模仿/src/core/deployment_int.go写一个/src/core/service_int.go:

/src/core/service_int.go

  1. package core
  2. import (
  3. "fmt"
  4. corev1 "k8s.io/api/core/v1"
  5. "log"
  6. "sync"
  7. )
  8. type ServiceMap struct {
  9. data sync.Map
  10. }
  11. func (svcmap *ServiceMap) Add(dep *corev1.Service) {
  12. if list, ok := svcmap.data.Load(dep.Namespace); ok {
  13. list = append(list.([]*corev1.Service), dep)
  14. svcmap.data.Store(dep.Namespace, list)
  15. } else {
  16. svcmap.data.Store(dep.Namespace, []*corev1.Service{dep})
  17. }
  18. }
  19. func (svcmap *ServiceMap) Update(dep *corev1.Service) error {
  20. if list, ok := svcmap.data.Load(dep.Namespace); ok {
  21. for i, range_dep := range list.([]*corev1.Service) {
  22. if range_dep.Name == dep.Name {
  23. list.([]*corev1.Service)[i] = dep
  24. svcmap.data.Store(dep.Namespace, list)
  25. }
  26. }
  27. return nil
  28. }
  29. return fmt.Errorf("Service-%s not found", dep.Name)
  30. }
  31. // 删除
  32. func (svcmap *ServiceMap) Delete(dep *corev1.Service) {
  33. if list, ok := svcmap.data.Load(dep.Namespace); ok {
  34. for i, range_dep := range list.([]*corev1.Service) {
  35. if range_dep.Name == dep.Name {
  36. newList := append(list.([]*corev1.Service)[:i], list.([]*corev1.Service)[i+1:]...)
  37. svcmap.data.Store(dep.Namespace, newList)
  38. break
  39. }
  40. }
  41. }
  42. }
  43. func (svcmap *ServiceMap) ListByNS(ns string) ([]*corev1.Service, error) {
  44. if list, ok := svcmap.data.Load(ns); ok {
  45. return list.([]*corev1.Service), nil
  46. }
  47. return nil, fmt.Errorf("record not found")
  48. }
  49. func (svcmap *ServiceMap) GetService(ns string, name string) (*corev1.Service, error) {
  50. depList, err := svcmap.ListByNS(ns)
  51. if err != nil {
  52. return nil, fmt.Errorf("Service not found")
  53. }
  54. for _, item := range depList {
  55. if item.Name == name {
  56. return item, nil
  57. }
  58. }
  59. return nil, fmt.Errorf("Service not found")
  60. }
  61. var svcmap *ServiceMap
  62. func init() {
  63. svcmap = &ServiceMap{}
  64. }
  65. type SvcHandler struct {
  66. }
  67. func (d *SvcHandler) OnAdd(obj interface{}) {
  68. //fmt.Println(obj.(*corev1.Service).Name)
  69. svcmap.Add(obj.(*corev1.Service))
  70. }
  71. func (d *SvcHandler) OnUpdate(oldObj, newObj interface{}) {
  72. err := svcmap.Update(newObj.(*corev1.Service))
  73. if err != nil {
  74. log.Println(err)
  75. }
  76. }
  77. func (d *SvcHandler) OnDelete(obj interface{}) {
  78. if d, ok := obj.(*corev1.Service); ok {
  79. svcmap.Delete(d)
  80. }
  81. }

Int

/src/core/deployment_init.go Init中增加ServiceInformer相关配置:

func InitDeployment() {
    factory := informers.NewSharedInformerFactory(lib.K8sClient, 0)
    depinformer := factory.Apps().V1().Deployments()
    depinformer.Informer().AddEventHandler(&DepHandler{})
    Podinformer := factory.Core().V1().Pods()
    Podinformer.Informer().AddEventHandler(&PodHandler{})
    Rsinformer := factory.Apps().V1().ReplicaSets()
    Rsinformer.Informer().AddEventHandler(&RSHandler{})
    eventInformer := factory.Core().V1().Events()
    eventInformer.Informer().AddEventHandler(&EventHandler{})
    ServiceInformer := factory.Core().V1().Services()
    ServiceInformer.Informer().AddEventHandler(&SvcHandler{})
    factory.Start(wait.NeverStop)
}

注意:只截取了相关配置,没有复制整个文件!

修改Service.go为list watch模式

/src/service/Service.go not list-watch原代码:

package service

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

type Service struct {
    Name       string
    Type       string
    ClusterIp  string
    ExternalIp []string
    Ports      []string
    Select     map[string]string
}

func ListService(g *gin.Context) {
    ns := g.Query("ns")
    svc, err := K8sClient.CoreV1().Services(ns).List(context.Background(), metav1.ListOptions{})
    if err != nil {
        g.Error(err)
        return
    }
    ret := make([]*Service, 0)
    for _, item := range svc.Items {
        ret = append(ret, &Service{
            Name:       item.Name,
            Type:       string(item.Spec.Type),
            ClusterIp:  item.Spec.ClusterIP,
            ExternalIp: item.Spec.ExternalIPs,
            Select:     item.Spec.Selector,
        })

    }
    g.JSON(200, ret)
    return
}

修改成list watch方式:

package service

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

type Service struct {
    Name       string
    Type       string
    ClusterIp  string
    ExternalIp []string
    Ports      []corev1.ServicePort
    Select     map[string]string
}

func ListService(g *gin.Context) {
    ns := g.Query("ns")
    //svc, err := K8sClient.CoreV1().Services(ns).List(context.Background(), metav1.ListOptions{})
    svclist, _ := core.SVCMap.ListByNS(ns)
    ret := make([]*Service, 0)
    for _, item := range svclist {
        ret = append(ret, &Service{
            Name:       item.Name,
            Type:       string(item.Spec.Type),
            ClusterIp:  item.Spec.ClusterIP,
            ExternalIp: item.Spec.ExternalIPs,
            Ports:      item.Spec.Ports,
            Select:     item.Spec.Selector,
        })

    }
    g.JSON(200, ret)
    return
}

运行main.go

package main

import (
    "github.com/gin-gonic/gin"
    "k8s-demo1/src/core"
    "k8s-demo1/src/service"
)

func main() {
    r := gin.Default()
    r.GET("/", func(context *gin.Context) {
        context.JSON(200, "hello")
    })
    r.GET("/namespaces", service.ListNamespace)
    r.GET("/deployments", service.ListDeployment)
    r.GET("/service", service.ListService)
    r.GET("pods", service.ListallPod)
    core.InitDeployment()
    r.Run()
}

image.png
做一个svc

[zhangpeng@zhangpeng k8s]$ cat nginx1-deployment.yaml 
apiVersion: v1
kind: Service
metadata:
  name: nginx1
  labels:
    app: nginx
spec:
  ports:
  - port: 80
    protocol: TCP
    targetPort: 80
  selector:
    app: nginx
[zhangpeng@zhangpeng k8s]$ kubectl apply -f  nginx1-deployment.yaml 
service/nginx1 created
[zhangpeng@zhangpeng k8s]$ kubectl get svc
NAME         TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)   AGE
kubernetes   ClusterIP   10.96.0.1      <none>        443/TCP   47d
nginx1       ClusterIP   10.111.134.5   <none>        80/TCP    48s

image.png
注:访问 svc就免了,我这里肯定会有问题的。我的app label都是nginx。所以我不能精确指定到deployment1的(我的初衷),所以尽量是app label去区分deployment!。随便修改一下nginx中index.html就发现他都会调度的几个deployments。所以我前面整deployment的时候也是一种耍流氓了…不过我还是想区分一下搞出来!
image.png

总结:

  1. list watch 就实现一个service了 statefulset configmap 其他的最后实现吧,curd先实现一下简单的
  2. deployment 的坑还是先留着……
  3. 下面继续准备实现一下create delete update的方法……