Kubernetes - controller-runtime

Last Updated: 2024-03-23

https://github.com/kubernetes-sigs/controller-runtime

controller-runtime: a set of go libraries for building Controllers.

Controllers are one kind of clients, they also talk to API Server, so controller-runtime wraps client-go and provide a Client to easily perform CRUD.

Unlike http handlers, Controllers DO NOT handle events directly, but enqueue Requests to eventually reconcile the object.

Controllers require a Reconciler to be provided to perform the work pulled from the work queue.

Controller will read and write to API server; webhooks are only called by API Server. controllers and webhooks may live in the same binary.

Create your own controllers

From a high level:

manifest -> image -> main() -> manager -> controller -> reconciler
                                     | -> webhook server -> webhooks

You will need the following pieces:

Reconcilers and / or Webhooks: the logic.
- Reconcilers implements reconcile.Reconciler, which has a method Reconcile(ctx context.Context, req ctrl.Request) (Result, error)
  - Request does not include any specifics about an action, it is just the namespace and the name, to identify the object that needs to be reconciled; "It does NOT contain information about any specific Event or the object contents itself."
  - Result does not tell what's changed in the reconciliation , but only has Requeue (a bool) and RequeueAfter (a time)
Controller Manager: host the reconcilers and or webhooks.
Binary: a main() func to start the controller manager.
- main() calls Start() of the controller manager sigs.k8s.io/controller-runtime/pkg/manager/manager.go
Image: build as an image.
Chart / Manifest: some yaml files on how to deploy the new controller.

Code Example

Create a new manager:

import ctrl "sigs.k8s.io/controller-runtime"
manager, err := ctrl.NewManager(cfg.RESTConfig, ctrlOptions)

ctrl.NewControllerManagedBy(manager).For(monitoredObj).Owns(generatedObj).complete(reconciler)

validator := v1.NewPodValidator(manager.GetClient())
manager.GetWebhookServer().Register("/validate-foo", &webhook.Admission{Handler: validator})

Start the manager:

manager.Start(ctx)

Get an object by Client

var foo Foo
key := client.ObjectKey{
		Name: version,
	}
if err := cl.Get(ctx, key, &foo); err != nil {

}

Get ObjectKey from Object

object => ObjectKey (i.e. Name+Namespace)

import "sigs.k8s.io/controller-runtime/pkg/client"

d := appsv1.Deployment{
			ObjectMeta: name,
		}
client.ObjectKeyFromObject(&d)

kubeconfigPath => client.Client

Create a client.Client from a path of the kubeconfig file:

// kubeconfigPath => rest.Config
loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
loadingRules.ExplicitPath = kubeconfigPath
restConfig=clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, nil).ClientConfig()

// rest.Config => client.Client
c, err := client.New(restConfig, client.Options{})

How to get the client?

You need a client to talk to the k8s cluster.

In controller-runtime: client.Client is defined in sigs.k8s.io/controller-runtime/pkg/client

controller-runtime depends on client-go. There 2 types of clients:

typed: k8s.io/client-go/kubernets.ClientSet
dynamic: k8s.io/client-go/dynamic.Interface

How to update status?

client.Status().Update(ctx, obj)

How to create new obj?

import "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

obj = &foo.Bar{
    ObjectMeta: metav1.ObjectMeta{
        Name:      "name",
        Namespace:  "namespace",
    },
}
controllerutil.CreateOrUpdate(ctx, client, obj, func () error {
    // mutation logic, i.e. modify obj based on existing states.
})

controllerutil provides CreateOrUpdate / CreateOrPatch

restconfig => (controller-runtime) Client

import (
  "k8s.io/apimachinery/pkg/runtime"
  "sigs.k8s.io/controller-runtime/pkg/client"
)

c, err := client.New(restConfig, client.Options{
  Scheme: runtime.NewScheme(),
})

AddToScheme

scheme is defined in apimachinery.

AddToScheme is in sigs.k8s.io/controller-runtime/pkg/scheme.

// Create a scheme
scheme *runtime.Scheme = runtime.NewScheme()

// Add to the scheme
utilruntime.Must(myapigroupv1.AddToScheme(scheme))

Note about AddToScheme: Avoid adding to scheme on a running process. This will cause concurrent read and write for the schemes. Consider adding it during registration or in init().

For vs Owns vs Watches

For: to declare the target resources that this reconciler is responsible for reconciling.
Owns: to declare the resources whose lifecycles are controlled by the target resources; these resources are usually created/updated in a reconciliation, and they will have an owner reference in their metadata pointing to the target resource. When these resources get changed, the owner will get requeued for a re-reconciliation. Example: A Deployment owns a ReplicaSet, and a ReplicaSet owns Pods.
Watches: to declare resources that are "inputs" to the target resources. Since there's no explicit metadata associating these input resources with target resources, you need to provide a function that has the business logic to locate the target resource to requeue from any given input resource.

Flags / Options

flags for controllers: previously use flags, now use component config (ctrl.Options).

options := ctrl.Options{Scheme: scheme}
mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), options)

Manager

controllers
webhook manager (mgr.GetWebhookServer())
client (mgr.GetClient())

Controller

Controller-Runtime controllers use a cache to subscribe to events from Kubernetes objects and to read those objects more efficiently by avoiding to call out to the API.

watch: what should we monitor?
reconcile: what to do if there's a difference?

Controller Implementation

You can implement an a Controller using any language / runtime that can act as a client for the Kubernetes API. E.g. kubebuilder for golang.

controller-runtime is a spin-off from the kubebuilder project. kubebuilder provides some additional scalfolding toolings on top of controller-runtime to further make it easy for beginners. Advanced users can use controller-runtime directly without using kubebuilder.

controller-runtime is built on top of client-go which is the official Kubernetes golang client for interacting with the API server.

Architecture

https://book.kubebuilder.io/architecture.html

Hierarchy: main -> Manager -> Controller -> Reconciler.

main.go: the entry point, calls the controller manager, packaged into a container.
Manager:
- sigs.k8s.io/controller-runtime/pkg/manager
Controller
- sigs.k8s.io/controller-runtime/pkg/controller
- one per Kind / CRD, calls a Reconciler each time it gets an event, after filtering by Predicates
- implemented as worker queues: unlike http handlers, Controllers DO NOT handle events directly, but enqueue Requests to eventually reconcile the object.
  - requires Watches to be configured to enqueue reconcile.Requests in response to events.
  - requires a Reconciler to be provided to perform the work pulled from the work queue.
Reconciler:
- sigs.k8s.io/controller-runtime/pkg/reconcile
- Reconciler is a function provided to a Controller that may be called at anytime with the Name and Namespace of an object.
- Reconcile() actually performs the reconciling for a single named object.
- Reconciler contains all of the business logic of a Controller.
- Reconciler typically works on a single object type.
- Reconciler does not care about the event contents or event type responsible for triggering the reconcile.
- Request: just has a name, but we can use the client to fetch that object from the cache.
- Result: return an empty result and no error, to indicates to controller-runtime that we’ve successfully reconciled this object and don’t need to try again until there’s some changes.

Other noteable pkg:

Predicate:
- sigs.k8s.io/controller-runtime/pkg/predicate
- filters a stream of events, pass those require action to reconciler
Webhook:
- sigs.k8s.io/controller-runtime/pkg/webhook

Deployment

The Controller will normally run outside of the control plane, much as you would run any containerized application. For example, you can run the controller in your cluster as a Deployment.

CRD in the crds folder of a Helm Chart.

Scheme

If create a client with empty Scheme, it has no scheme at all, you may get errors like no kind is registered for the type v1.NodeList in scheme when using the client.

If use an empty option, it will use a default Scheme which has includes some types in core.

import "sigs.k8s.io/controller-runtime/pkg/client"

// Create with an empty Scheme.
c, err := client.New(restConfig, client.Options{
  Scheme: runtime.NewScheme(),
})

// Create with an empty Options, but default Scheme.
c, err := client.New(restConfig, client.Options{})

IgnoreNotFound

If the error is NotFound, return nil, otherwise return the err:

import "sigs.k8s.io/controller-runtime/pkg/client"

func ... error {
	if err = ...; err != nil {
		return client.IgnoreNotFound(err)
	}
}

Ensure the namespace exists

ns := &corev1.Namespace{
  ObjectMeta: metav1.ObjectMeta{
    Name: "foo-system",
  },
}
if err := cl.Create(ctx, ns); client.IgnoreAlreadyExists(err) != nil {
  return fmt.Errorf("failed to create namespace: %w", err)
}

Scale up or down a deployment from code

deployment := appsv1.Deployment{}
deploymentKey := client.ObjectKey{
  Namespace: "foo-system",
  Name:      "foo",
}
// Read
if err := adminC.Get(ctx, deploymentKey, &deployment); err != nil {
  return fmt.Errorf("failed to get the Deployment %s: %w", deploymentKey, err)
}
// Set the new num of replicas
deployment.Spec.Replicas = ptr.To[int32](0)
// Write
if err := adminC.Update(ctx, &deployment); err != nil {
  return fmt.Errorf("failed to scale the Deployment %s: %w", deploymentKey, err)
}

Wait for Pod Running

func waitForPodRunning(ctx context.Context, cl client.Client, podKey client.ObjectKey) error {
	isReady := func() error {
		pod := &v1.Pod{}

		if err := cl.Get(ctx, podKey, pod); err != nil {
			return err
		}
		if pod.Status.Phase == v1.PodRunning {
			return nil
		}
		return err
	}
	if err := backoff.Retry(isReady, defaultBackoff); err != nil {
		return fmt.Errorf("timed out waiting for pod to be running: %v", err)
	}
	return nil
}