Watch and react to Kubernetes objects changes

#go #kubernetes #clientgo #cloudnative

The Kubernetes API server exposes a REST interface consumable by any client.

client-go is the official client library for the Go programming language. It is used both internally by Kubernetes itself (for example, inside kubectl) as well as by numerous external consumers: operators, higher level frameworks and many more.

Using this library you can write Go applications to access kubernetes' API Server and you can programmatically add, delete, modify, and check kubernetes resources.

client-go introduces different types of clients such as: RESTClient, Clientset and dynamic.Interface. All these clients make available the Watch verb, which offers an event interface that reacts to objects changes: add, update, delete, etc.

In all cases, the returned object is an implementation of watch.Interface that looks like this:

Let's implement a watcher for namespaces changes. The application will do these things:

attempts to begin watching the namespaces resource getting a watch.Interface on success
iterates all the events produced by the watcher
when an event is of type "namespace added" adds a custom label patching the namespace
when en event is of type "namespace deleted" greet the namespace gone

In this article I'll show you how to use RESTClient to watch and then react to namespaces changes.

Here the source code - you can grab all the code @ https://github.com/lucasepe/using-client-go.

package main


import (
  "context"
  "fmt"
  "time"


  corev1 "k8s.io/api/core/v1"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  "k8s.io/apimachinery/pkg/types"
  "k8s.io/apimachinery/pkg/watch"
  "k8s.io/client-go/kubernetes/scheme"
  "k8s.io/client-go/rest"
  "k8s.io/client-go/tools/clientcmd"
)

func main() {
  // Using the default configuration rules get the info 
  // to connect to the Kubernetes cluster
  configLoader := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
    clientcmd.NewDefaultClientConfigLoadingRules(),
    &clientcmd.ConfigOverrides{},
  )

  // create the Config object
  cfg, err := configLoader.ClientConfig()
  if err != nil {
    panic(err)
  }

  // we want to use the core API (namespaces lives here)
  cfg.APIPath = "/api"
  cfg.GroupVersion = &corev1.SchemeGroupVersion
  cfg.NegotiatedSerializer = scheme.Codecs.WithoutConversion()

  // create a RESTClient
  rc, err := rest.RESTClientFor(cfg)
  if err != nil {
    panic(err.Error())
  }

  // utility function to create a int64 pointer
  i64Ptr := func(i int64) *int64 { return &i }

  opts := metav1.ListOptions{
    TimeoutSeconds: i64Ptr(120),
    Watch:          true,
  }

  // attempts to begin watching the namespaces
  // returns a `watch.Interface`, or an error
  watcher, err := rc.Get().Resource("namespaces").
    VersionedParams(&opts, scheme.ParameterCodec).
    Timeout(time.Duration(*opts.TimeoutSeconds)).
    Watch(context.TODO())
  if err != nil {
    panic(err)
  }

  // the patch data, just add a custom label
  pd := []byte(`{"metadata":{"labels":{"modified-by":"lucasepe"}}}`)

  // the patch type
  pt := types.MergePatchType

  // who did this patch?
  po := metav1.PatchOptions{
    FieldManager: "my-cool-app",
  }

  // here we iterate all the events streamed by the watch.Interface
  for event := range watcher.ResultChan() {
    // retrieve the Namespace
    item := event.Object.(*corev1.Namespace)

    switch event.Type {
    // when a namespace is deleted...
    case watch.Deleted:
      // let's say hello!
      fmt.Printf("- '%s' %v ...bye bye\n", item.GetName(), event.Type)

    // when a namespace is added...
    case watch.Added:
      fmt.Printf("+ '%s' %v  ", item.GetName(), event.Type)

      // try to patch it!
      err = rc.Patch(pt).Resource("namespaces").
        Name(item.Name).
        VersionedParams(&po, scheme.ParameterCodec).
        Body(pd).
        Do(context.TODO()).
        Error()
      if err != nil {
        panic(err)
      }

      fmt.Println(" ...patched!")
    }
  }
}

To see what the program does, open two terminal windows.

In the first terminal run the code as usual:

$ go run main.go

In the second terminal create a namespace using kubectl:

$ kubectl create namespace demo-system

In the first terminal you should see a new line like this:

+ 'demo-system' ADDED   ...patched!

Going back to the second terminal and typing:

$ kubectl describe namespace demo-system

Name:         demo-syste
Labels:       kubernetes.io/metadata.name=demo-system
              modified-by=lucasepe
Annotations:  <none>
Status:       Active


No resource quota.


No LimitRange resource

You can see that the program added a new label to to the newly created namespace.

Now deleting the namespace with:

$ kubectl delete namespace demo-system

DEV Community

DEV Community is a community of 822,529 amazing developers

Watch and react to Kubernetes objects changes