Kubernetes watch golang实现

第一种方法

使用k8s.io/client-go/tools/cache

func StartWatchingServices(c *Client) {
    watchlist := cache.NewListWatchFromClient(
        c.K8sClientset.CoreV1().RESTClient(),
        string(v1.ResourceServices),
        v1.NamespaceAll,
        fields.Everything(),
    )

    _, controller := cache.NewInformer(
        watchlist,
        &v1.Service{},
        0, //Duration is int64
        cache.ResourceEventHandlerFuncs{
            AddFunc: func(obj interface{}) {
                fmt.Printf("added: %s \n", obj)
            },
            DeleteFunc: func(obj interface{}) {
                fmt.Printf(deleted: %s \n", obj)
            },
            UpdateFunc: func(oldObj, newObj interface{}) {
                fmt.Printf("changed \n")
            },
        },
    )

    stop := make(chan struct{})
    defer close(stop)
    go controller.Run(stop)
    for {
        time.Sleep(time.Second)
    }
}

第二种方法

使用k8s.io/client-go/informers，可以watch CRD资源但需要使用code-generator生成clientset和informer(指CRD资源)。

func StartWatchingServices2(c *Client) {
    kubeInformerFactory := informers.NewSharedInformerFactory(c.K8sClientset, time.Second*30)
    svcInformer := kubeInformerFactory.Core().V1().Services().Informer()

    svcInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            fmt.Printf("added: %s \n", obj)
        },
        DeleteFunc: func(obj interface{}) {
            fmt.Printf("deleted: %s \n", obj)
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            fmt.Printf("changed: %s \n", newObj)
        },
    })

    stop := make(chan struct{})
    defer close(stop)
    kubeInformerFactory.Start(stop)
    for {
        time.Sleep(time.Second)
    }
}

第三种方法

使用runtimeCache "sigs.k8s.io/controller-runtime/pkg/cache", 可以watch CRD资源，无需特殊处理。下例v1alpha1.Component是CRD资源。

func StartWatchingComponents(c *Client) {
    informerCache, err := runtimeCache.New(c.RestConfig, runtimeCache.Options{})
    if err != nil {
        panic(err)
    }

    informer, err := informerCache.GetInformer(&v1alpha1.Component{})
    if err != nil {
        panic(err)
    }

    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            fmt.Printf("added: %s \n", obj)
        },
        DeleteFunc: func(obj interface{}) {
            fmt.Printf("deleted: %s \n", obj)
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            fmt.Printf("changed: %s \n", newObj)
        },
        })

    stop := make(chan struct{})
    defer close(stop)
    informerCache.Start(stop)
    // for {
    //  time.Sleep(time.Second)
    // }
}

有疑问加站长微信联系（非本文作者）