Monthly Archives: July 2019

In the last few posts, we have looked in detail at the various parts of the machinery behind Kubernetes controllers – informers, queues, stores and so forth. In this final post, we will wrap up a bit and see how all this comes together in the Kubernetes sample controller.

The flow of control

The two main source code files that make up the sample controller are main.go and controller.go. The main function is actually rather simple. It creates a clientset, an informer factory for Foo objects and an informer factory for deployments and then uses those items to create a controller.

Once the controller exists, main starts the informers using the corresponding functions of the factory, which will bring up the goroutines of the two informers. Finally, the controllers main loop is started by calling its Run method.

The controller is more interesting. When it is created using NewController, it attaches event handlers to the informes for Deployments and Foo resources. As we have seen, both event handlers will eventually put Foo objects that require attention into a work queue.

The items in this work queue are processed by worker threads that are created in the controllers Run method. As the queue servers act as a synchronization point, we can run as many worker threads that we want and still make sure that each event is processed by only one worker thread. The main function of the worker thread is processNextWorkItem which retrieves elements from the queue, i.e. the keys of the Foo objects that need to be reconciled, and calls syncHandler for each of them. If that functions fails, the item is put back onto the work queue.

The syncHandler function contains the actual reconciliation logic. It first splits the key into namespace and name of the Foo resource. Then it tries to receive an existing deployment for this Foo resource and creates one if it does not yet exist. If a deployment is found, but deviates from the target state, it is replaced by a new deployment as returned by the function newDeployment. Finally, the status subresource of the Foo object is updated.

So this simple controller logic realizes some of the recommendations for building controllers.

• It uses queues to decouple workers from event handlers and to allow for concurrent processing
• Its logic is level based, not edge based, as a controller might be down for some time and miss updates
• It uses shared informers and caches. The cached objects are never updated directly, but if updates are needed, deep copies are used and modified and updates are done via the API
• It waits for all caches to sync before starting worker threads
• It collapses all work to be done into a single queue

Package structure and generated code

If you browse the sourcecode of the sample controller, you might at the first glance be overwhelmed by the number of files in the repository. However, there is good news – most of this code is actually generated! In fact, the diagram below shows the most important files and packages. Generated files are in italic font, and files that serve as input for the code generator or are referenced by the generated code are in bold face.

We see that the number of files that we need to provide is comparatively small. Apart from, of course, the files discussed above (main.go and controller.go), these are

• register.go which adds our new types to the scheme and contains some lookup functions that the generated code will use
• types.go which defines the Go structures that correspond to the Foo CRD
• register.go in the top-level directory of the API package which defines the name of the API group as a constant

All the other files are created by the Kubernetes API code generator. This generator accepts a few switches (“generators”) to generate various types of code artifacts – deep-copy routines, clientsets, listers and informers. We will see the generator in action in a later post. Instead of using the code generator directly, we could of course as well use the sample controller as a starting point for our own code and make the necessary modifications manually, the previous posts should contain all the information that we need for this.

This post completes our short mini-series on the machinery behind custom controllers. In the next series, we will actually apply all this to implement a controller that operators a small bitcoin test network in a Kubernetes cluster.

After understanding how an informer can be used to implement a custom controller, we will now learn more about the inner working of an informer.

We have seen that essentially, informers use the Kubernetes API to learn about changes in the state of a Kubernetes cluster and use that information to maintain a cache (the indexer) of the current cluster state and to inform clients about the changes by calling handler functions. To achieve this, an informer (more specifically: a shared informer) is again a composition of several components.

1. A reflector is the component which is actually talking to the Kubernetes API server
2. The reflector stores the information on state changes in a special queue, a Delta FIFO
3. A processor is used to distribute the information to all registered event handlers
4. Finally, a cache controller is reading from the queue and orchestrating the overall process

In the next few sections, we visit each of these components in turn.

Reflectors and the FIFO queue

In the last post, we have seen that the Kubernetes API offers mechanisms like resource versions and watches to allow a client to keep track of the cluster state. However, these mechanism require some logic – keeping track of the resource version, handling timeouts and so forth. Within the Kubernetes client package, this logic is built into an Reflector.

A reflector uses the API to keep track of changes for one specific type of resources and update an object store accordingly. To talk to the Kubernetes API, it uses an object implementing the ListWatch interface, which is simply a convenience interface with a default implementation

When a reflector is started by invoking its Run method, it will periodically run the ListAndWatch method, which contains the core logic. This function first lists all resources and retrieves the resource version. It then uses the retrieved list to rebuild the object store. It then creates a watch (which is an instance of the watch.Interface interface) and reads from the channel provided by this watch resource in a loop, while keeping the resource version up to date. Depending on the type of the event, it then calls either Add, Update or Delete on the store.

The store that is maintained by the reflector is not just an ordinary store, but a Delta FIFO. This is a special store that maintains deltas, i.e. instances of cache.Delta, which is a structure containing the changed object and the type of change. The store will, for each object that has been subject to at least one change, maintain a list of all changes that have been reported for this object, and perform a certain de-duplication. Thus a reader will get a complete list of all deltas for a specific resource.

The cache controller and the processor

So our reflector is feeding the delta FIFO queue, but who is reading from it? This is done by another component of the shared informer – the cache controller.

When the Run method of a cache controller is invoked, it creates a Reflector, starts the Reflectors Run method in a separate goroutine and starts its own main loop processLoop as a second goroutine. Within that loop, the controller will pop elements off the FIFO queue and invoke the handleDeltas method of the informer once for each element in the queue. This method represents the main loop of the informer that is executed once for each object in the delta store.

This function will do two things. First, it will update the indexer according to the detected change retrieved from the delta FIFO queue. Second, it will build notifications (using the indexer to get old versions of an object if needed) to create notifications and distribute them to all handler functions. This work is delegated to the processor.

Essentially, a processor is an object that maintains a list of listeners. A listeners has an add method to receive notifications. These notifications are then buffered and forwarded to a resource event handler which can be any object that has methods OnAdd, OnUpdate and OnDelete.

We now have a rather complete understanding of what happens in case the state of the Kubernetes cluster changes.

• The reflector picks up the change via the Kubernetes API and enqueues it in the delta FIFO queue, where deduplication is performed and all changes for the same object are consolidated
• The controller is eventually reading the change from the delta FIFO queue
• The controller updates the indexer
• The controller forwards the change to the processor, which in turns calls all handler functions
• The handler functions are methods of the custom controller who can then inspect the delta, use the updated indexer to retrieve additional information and adjust the cluster state if needed

Note that the same informer object can serve many different handler functions, while maintaining only one, shared indexer – as the name suggests. Throughout the code, synchronisation primitives are used to protect the shared data structures to make all this thread-safe.

Creating and using shared informers

Let us now see how we can create and use a shared informer in practice. Suppose we wanted to write a controller that is taking care of pods and, for instance, trigger some action whenever a pod is created. We can do this by creating an informer which will listen for events on pods and call a handler function that we define. Again, the sample controller is guiding us how to do this.

The recommended way to create shared informers is to use an informer factory. This factory maintains a list of informers per resource type, and if a new informer is requested, it first performs a lookup in that list and returns an existing informer if possible (this happens in the function InformerFor). It is this mechanism which actually makes the informers and therefore their caches shared and reduces both memory footprint and the load on the API server.

We can create a factory and use it to get a shared informer listening for pod events as follows.

factory := informers.NewSharedInformerFactory(clientset, 30*time.Second)
podInformer := factory.Core().V1().Pods().Informer()

Here clientset is a Kubernetes clientset that we can create as in our previous examples, and the second argument is the resync period. In our example, we instruct the informer to rebuild its cache every 30 seconds.

Once our informer is created, we need to start its main loop. However, we cannot simply call its Run method, as this informer might be shared and we do not want to call this method twice. Instead, we can again use a convenience function of the factory class – its method Start will start all informers created by the factory and keep track of their status.

We can now register event handlers with the informer. An event handler is an object implementing the ResourceHandler interface. Instead of building a class implementing this interface ourselves, we can use the existing class ResourceHandlerFuncs. Assuming that our handler functions for adding, updating and deleting pods are called onAdd, onUpdate and onDelete, the code to add these functions as an event handler would look as follows.

podInformer.AddEventHandler(
		&cache.ResourceEventHandlerFuncs{
			AddFunc:    onAdd,
			DeleteFunc: onDelete,
			UpdateFunc: onUpdate,
		})

Finally, we need to think about how we stop our controller again. At the end of our main function, we cannot simply exit, as this would stop the entire controller loop immediately. So we need to wait for something – most likely for a signal sent to us by the operating system, for instance because we have hit Ctrl-C. This can be achieved using the idea of a stop channel. This is a channel which is closed when we want to exit, for instance because a signal is received. To achieve this, we can create a dedicated goroutine which uses the os standard package to receive signals from the operating system and closes the channel accordingly. An example implementation is part of the sample controller.

Our event handlers are now very easy. They receive an object that we can convert to a Pod object using a type assertion. We can than use properties of the pod, for instance its name, print information or process that information further. A full working example can be found in my GitHub repository here.

This completes our post for today. In the next post in this mini-series on Go and Kubernetes, we will put everything that we have learned so far together and talk a short walk through the full source code of the sample controller to understand how a custom controller works. This will then finally put us in a position to start the implementation of our planned bitcoin controller.

In previous posts in my series on Kubernetes, we have stumbled across the concept of a controller. Essentially, a controller is a daemon that monitors the to-be state of components of a Kubernetes cluster against the as-is state and takes action to reach the to-be state.

A classical example is the Replica set controller which monitors replica sets and pods and is responsible for creating new pods or deleting existing pods if the number of replicas is out-of-sync with the defined value.

In this series, we will perform a deep dive into controllers. Specifically, we will take a tour through the sample controller that is provided by the Kubernetes project and try to understand how this controller works. We will also explain how this relates to customer resource definitions (CRDs) and what steps are needed to implement a customer controller for a given CRD.

Testing the sample controller

Let us now start with our analysis of the sample controller. To follow this analysis, I advise to download a copy of the sample controller into your Go workspace using go get github.com/kubernetes/sample-controller and then using an editor like Atom that offers plugins to navigate Go code.

To test the client and to have a starting point for debugging and tracing, let us follow the instructions in the README file that is located at the root of the repository. Assuming that you have a working kubectl config in $HOME/.kube/config, build and start the controller as follows.

$ cd $GOPATH/src/k8s.io/sample-controller
$ go build
$ kubectl create -f artifacts/examples/crd.yaml
$ ./sample-controller --kubeconfig=$HOME/.kube/config

This will create a custom resource definition, specifically a new resource type named “Foo” that we can use as any other resource like Pods or Deployments. In a separate terminal, we can now create a Foo resource.

$ kubectl create -f artifacts/examples/example-foo.yaml 
$ kubectl get deployments
$ kubectl get pods

You will find that our controller has created one deployment which in turn brings up one pod running nginx. If you delete the custom resource again using kubectl delete foo example-foo, both, the deployment and the pods disappear again. However, if you manually delete the deployment, it is is recreated by the controller. So apparently, our controller is able to detect changes to deployments and foo resources and to match them accordingly. How does this work?

Basically, a controller will periodically match the state of the system to the to-be state. For that purpose, several functionalities are required.

• We need to be able to keep track of the state of the system. This is done based on an event-driven processing and handled by informers that are able to subscribe to events and invoke specific handlers and listers that are able to list all resources in a given Kubernetes cluster
• We need to be able to keep track of the state of the system. This is done using object stores and their indexed variants indexer
• Ideally, we should be able to process larger volumes using multi-threading, coordinated by queues

In this and the next post, we will go through these elements one by one. We start with queues.

Queues and concurrency

Let us start by investigating threads and queues in the Kubernetes library. The ability to easily create threads (called go-routines) in Go and the support for managing concurrency and locking are one of the key differentiators of the Go programming language, and of course the Kubernetes client library makes use of these features.

Essentially, a queue in the Kubernetes client library is something that implements the interface k8s.io/client-go/util/workqueue/Interface. That interface contains (among others) the following methods.

• Add adds an object to the queue
• Get blocks until an item is available in the queue, and then returns the first item in the queue
• Done marks an item as processed

Internally, the standard implementation of a queue in queue.go uses Go maps. The keys of these maps are arbitrary objects, the items in the map are actually just placeholders (empty structures). One of these maps is called the dirty set, this map contains all elements that make up the actual queue, i.e. need to be processed. The second map is called the processing set, these are all items which have been retrieved using Get, but for which Done has not yet been called. As maps are unordered, there is also an array which holds the elements in the queue and is used to define the order of processing. Note that each of the maps can hold a specific object only once, whereas the queue can hold several copies of the object.

If we add something to the queue, it is added to the dirty set and appended to the queue array. If we call Get, the first item is retrieved from the queue, removed from the dirty set and added to the processing set. Calling Done will remove the element from the processing set as well, unless someone else has called Add in the meantime again on the same object – in this case it will be removed from the processing set, but also be added to the queue again.

Let us implement a little test program that works with queues. For that purpose, we will establish two threads aka goroutines. The first thread will call Add five times to add something to the queue and then complete. The second thread will sleep for three seconds and then read from the queue in a loop to retrieve the elements. Here are the functions to send and receive.

func fillQueue(queue workqueue.Interface) {
	time.Sleep(time.Second)
	queue.Add("this")
	queue.Add("is")
	queue.Add("a")
	queue.Add("complete")
	queue.Add("sentence")
}

func readFromQueue(queue workqueue.Interface) {
	time.Sleep(3 * time.Second)
	for {
		item, _ := queue.Get()
		fmt.Printf("Got item: %s\n", item)
		queue.Done(item)
	}
}

With these two functions in place, we can now easily create two goroutines that use the same queue to communicate (goroutines, being threads, share the same address space and can therefore communicate using common data structures).

myQueue := workqueue.New()
go fillQueue(myQueue)
go readFromQueue(myQueue)

However, if you run this, you will find that there is a problem. Our main thread completes after creating both worker threads, and this will cause the program to exit and kill both worker threads before the reader has done anything. To avoid this, we need to wait for the reader thread (which, reading from the queue, will in turn wait for the writer thread). One way to do this with Go language primitives is to use channels. So we change our reader function to receive a channel of integer elements

func readFromQueue(queue workqueue.Interface, stop chan int)

and in the main function, we create a channel, pass it to the function and then read from it which will block until the reader thread sends a confirmation that it is done reading.

stop := make(chan int)
myQueue := workqueue.New()
go fillQueue(myQueue)
go readFromQueue(myQueue, stop)
<-stop

Now, however, there is another problem – how does the reader know that no further items will be written to the queue? Fortunately, queues offers a way to handle this. When a writer is done using the queue, it will call Shutdown on the queue. This will change the queues behavior – reads will no longer be blocking, and the second return value of a Get will be true if the queue is empty. If a reader recognizes this situation, it can stop its goroutine.

A full example can be found here – of course this is made up to demonstrate goroutines, queues and channels and not the most efficient solution for the problem at hand.

The core controller loop

Armed with our understanding of concurrency and queues, we can now take a first look at the code of the sample controller. The main entry points for are the function handleObject and enqueueFoo – these are the functions invoked by the Informer, which we will discuss in one of the next posts, whenever either a Foo object or a Deployment is created, updated or deleted.

The function enqueueFoo is called whenever a Foo object is changed (i.e. added, updated or deleted). It simply determines a key for the object and adds that key to the workqueue.

The workqueue is read by worker threads, which are created in the Run function of the controller. This function creates a certain number of goroutines and then listens on a channel called stopCh, as we have done it in our simplified example before. This channel is created by main.go and used to be able to stop all workers and the main thread if a signal is received.

Each worker thread executes the method processNextItem of the controller in a loop. For each item in the queue, this method calls another method – syncHandler – passing the item retrieved from the queue, i.e. the key of the Foo resource. This method then uses a Lister to retrieve the current state of the Foo resource. It then retrieves the deployment behind the Foo resource, creates it if it could not be found, and updates the number of replicas if needed.

The function handleObject is similar. It is invoked by the informer with the new, updated state of the Deployment object. It then determines the owning Foo resource and simply enqueues that Foo resource. The rest of the processing will then be the same.

At this point, two open ends remain. First, we will have to understand how an Informer works and how it invokes the functions handleObject and enqueueFoo. And we will need to understand what a Lister is doing and where the lister and the data is uses comes from. This will be the topic of our next post.