Client-Go, How to watch for newly created Pods in Kubernetes-CodePudding

I need to write a golang application with the help of client-go which will listen/watch a particular namespace for any of these events:

A new pod has been created
A pod has been deleted
A new container has been added to existing pods
Image for container for any pod has changed

And I want to communicate this information to another application application running in other namespace.

I am really new to the client-go library and I searched their documentation but couldn't find something similar to Events in Kopf

I am new to this library and I couldn't find a method/function of doing this. I don't need to have the full code of doing this, but I appreciate where I can look into, so I can find my way out

Can someone help me on this?

CodePudding user response：

You could use something like kubernetes-event-exporter or kube-eventer and send messages with different sinks.

CodePudding user response：

You can create a clientset from parsing the kubeconfig file and then use this clientset to create a sharedInformerfactory for your particular namespace. Get a informer for your pods and add Event Handler functions. Implement those functions according to your requirement. You can check for container updates between oldPod and newPod in the OnUpdate function. Use the clientset for however you want to communicate with other applications. I would say explore the methods that clientset implements to get a detailed idea how it works.

package main

import (
    "flag"
    "fmt"
    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/util/runtime"
    "k8s.io/client-go/informers"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/cache"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/util/homedir"
    "k8s.io/klog/v2"
    "path/filepath"
    "time"
)

func main() {
    // parse the .kubeconfig file
    var kubeconfig *string
    if home := homedir.HomeDir(); home != "" {
        kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
    } else {
        kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
    }
    flag.Parse()

    // create config from the kubeconfig
    config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
    if err != nil {
        panic(err)
    }

    // create the clientset
    clientSet, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err)
    }

    // stop signal for the informer
    stopper := make(chan struct{})
    defer close(stopper)
    
    // create shared informers for resources in all known API group versions with a reSync period and namespace
    factory := informers.NewSharedInformerFactoryWithOptions(clientSet, 10*time.Second, informers.WithNamespace("demo"))
    podInformer := factory.Core().V1().Pods().Informer()

    defer runtime.HandleCrash()

    // start informer ->
    go factory.Start(stopper)

    // start to sync and call list
    if !cache.WaitForCacheSync(stopper, podInformer.HasSynced) {
        runtime.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))
        return
    }

    podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    onAdd, // register add eventhandler
        UpdateFunc: onUpdate,
        DeleteFunc: onDelete,
    })

    // block the main go routine from exiting
    <-stopper
}

func onAdd(obj interface{}) {
    pod := obj.(*corev1.Pod)
    klog.Infof("POD CREATED: %s/%s", pod.Namespace, pod.Name)
}

func onUpdate(oldObj interface{}, newObj interface{}) {
    oldPod := oldObj.(*corev1.Pod)
    newPod := newObj.(*corev1.Pod)
    klog.Infof(
        "POD UPDATED. %s/%s %s",
        oldPod.Namespace, oldPod.Name, newPod.Status.Phase,
    )
}

func onDelete(obj interface{}) {
    pod := obj.(*corev1.Pod)
    klog.Infof("POD DELETED: %s/%s", pod.Namespace, pod.Name)
}