Lock in $30 Savings on PRO—Offer Ends Soon! ⏳

It's all about reconciliation - Anatomy of a ku...

It's all about reconciliation - Anatomy of a kubernetes controller

Developing a native kubernetes app goes around two simple concepts: defining new data structures (custom resources) and delivering a kubernetes component that handles those structures.

Despite that, doing the first steps can be intimidating:
• How do we define (and consume) new resources?
• How about interacting with existing resources defined by somebody else?
• How do I even find their Go definition?

In this talk, I'll introduce:
• How to interact with kubernetes custom resource using non type safe clients
• How to leverage the code generation to avoid casting errors and reduce the proliferation of boilerplate code
• How informers, workqueues and other powerful tools make it possible to implement a bullet proof event loop
• Finally, I'll describe the operator-sdk, a framework that makes the implementation of a controller even simpler and more automated

Federico Paolinelli

July 12, 2020
Tweet

More Decks by Federico Paolinelli

Other Decks in Programming

Transcript

  1. About me • Red Hatter • Doing distributed systems for

    more than 10 years • OpenShift CNF networking • Passionate about open source @fedepaol [email protected] [email protected]
  2. Kubernetes “Kubernetes is an open-source container-orchestration system for automating application

    deployment, scaling, and management. It was originally designed by Google, and is now maintained by the Cloud Native Computing Foundation”. Wikipedia
  3. Declarative over imperative apiVersion: apps/v1 kind: ReplicaSet metadata: name: frontend

    spec: replicas: 3 selector: matchLabels: tier: frontend template: metadata: labels: tier: frontend spec: containers: - name: php-redis image: gcr.io/google_samples/gb-frontend:v3 Pod 1 Pod 3 Pod 2
  4. Declarative over imperative apiVersion: apps/v1 kind: ReplicaSet metadata: name: frontend

    spec: replicas: 2 selector: matchLabels: tier: frontend template: metadata: labels: tier: frontend spec: containers: - name: php-redis image: gcr.io/google_samples/gb-frontend:v3 Pod 1 Pod 3 Pod 2
  5. Knock! Knock! Race Condition! Who’s There? • Optimistic Concurrency •

    Conflicts are resolved with failures • The client must implement the extra logic of retrying ETCD API Server Controller 1 Controller 2
  6. Missed Events • Network is not reliable • Pods can

    die • Nodes can die! https://flic.kr/p/ofjycj
  7. Edge Driven vs Level Driven 0 -> 1 1 ->

    0 0 -> 2 With edges, the behaviour depends on the variation of the data observed at edge location.
  8. Edge Driven vs Level Driven 0 1 0 2 With

    levels, we care about the level, in our case the state of the system at a given time.
  9. Edge Triggered, Level Driven It’s fine to use the events

    (edges) as waking up points, but we must use the system state (the level) to implement our logic.
  10. Common traits of a K8s kind https://flic.kr/p/c7yUpJ type Pod struct

    { metav1.TypeMeta metav1.ObjectMeta Spec PodSpec Status PodStatus }
  11. Type Meta type TypeMeta struct { Kind string APIVersion string

    } type ObjectMeta struct { Name string GenerateName string Namespace string SelfLink string UID types.UID ResourceVersion string Generation int64 . . Object Meta
  12. Client-Go config, _ := clientcmd.BuildConfigFromFlags("", "/home/fede/kubeconfig") clientset, _ := kubernetes.NewForConfig(config)

    pods, _ := clientset.CoreV1().Pods("testnamespace").List(v1.ListOptions{}) ds, _ := clientset.AppsV1().DaemonSet("testnamespace").Get("myset", v1.GetOptions{})
  13. Client-Go config, _ := clientcmd.BuildConfigFromFlags("", "/home/fede/kubeconfig") clientset, _ := kubernetes.NewForConfig(config)

    pods, _ := clientset.CoreV1().Pods("testnamespace").List(v1.ListOptions{}) ds, _ := clientset.AppsV1().DaemonSet("testnamespace").Get("myset", v1.GetOptions{}) Get Delete Create Update List Patch Watch
  14. Watch returns a watch.Interface type Interface interface { Stop() ResultChan()

    <-chan Event } type Event struct { Type EventType Object runtime.Object }
  15. Shared Informers import kubeinformers "k8s.io/client-go/informers" kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30) podsInformer

    := kubeInformerFactory.Core().V1().Pods() podsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { }, ... }) podsInformer.Informer().Lister().Pods("namespace").Get("podname")
  16. Work Queues type Interface interface { Add(item interface{}) Len() int

    Get() (item interface{}, shutdown bool) Done(item interface{}) ShutDown() ShuttingDown() bool }
  17. Rate Limiting Work Queue func NewRateLimitingQueue(rateLimiter RateLimiter) RateLimitingInterface type RateLimitingInterface

    interface { DelayingInterface AddRateLimited(item interface{}) Forget(item interface{}) NumRequeues(item interface{}) int }
  18. Setting the informer and the queue kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)

    informer := kubeInformerFactory.Core().V1().Pods() informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { pod := obj.(*v1.Pod) key, _ = cache.MetaNamespaceKeyFunc(obj) workqueue.Add(key) }, ... cache.WaitForCacheSync(stopCh, informer.Informer().HasSynced)
  19. Setting the informer and the queue kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)

    informer := kubeInformerFactory.Core().V1().Pods() informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { pod := obj.(*v1.Pod) key, _ = cache.MetaNamespaceKeyFunc(obj) workqueue.Add(key) }, ... cache.WaitForCacheSync(stopCh, informer.Informer().HasSynced)
  20. Setting the informer and the queue kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)

    informer := kubeInformerFactory.Core().V1().Pods() informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { pod := obj.(*v1.Pod) key, _ = cache.MetaNamespaceKeyFunc(obj) workqueue.Add(key) }, ... cache.WaitForCacheSync(stopCh, informer.Informer().HasSynced)
  21. Setting the informer and the queue kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)

    informer := kubeInformerFactory.Core().V1().Pods() informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { pod := obj.(*v1.Pod) key, _ = cache.MetaNamespaceKeyFunc(obj) workqueue.Add(key) }, ... cache.WaitForCacheSync(stopCh, informer.Informer().HasSynced)
  22. Setting the informer and the queue kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)

    informer := kubeInformerFactory.Core().V1().Pods() informer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { pod := obj.(*v1.Pod) key, _ = cache.MetaNamespaceKeyFunc(obj) workqueue.Add(key) }, ... cache.WaitForCacheSync(stopCh, informer.Informer().HasSynced)
  23. Reading from the queue obj, shutdown := c.workqueue.Get() if shutdown

    { return false } err := func(obj interface{}) error { defer workqueue.Done(obj) if key, ok := obj.(string); !ok { // should never happen } if err := handle(key); err != nil { workqueue.AddRateLimited(key) return fmt.Errorf(“Help”) } c.workqueue.Forget(obj) return nil }(obj)
  24. Reading from the queue obj, shutdown := c.workqueue.Get() if shutdown

    { return false } err := func(obj interface{}) error { defer workqueue.Done(obj) if key, ok := obj.(string); !ok { // should never happen } if err := handle(key); err != nil { workqueue.AddRateLimited(key) return fmt.Errorf(“Help”) } c.workqueue.Forget(obj) return nil }(obj)
  25. Reading from the queue obj, shutdown := c.workqueue.Get() if shutdown

    { return false } err := func(obj interface{}) error { defer workqueue.Done(obj) if key, ok := obj.(string); !ok { // should never happen } if err := handle(key); err != nil { workqueue.AddRateLimited(key) return fmt.Errorf(“Help”) } c.workqueue.Forget(obj) return nil }(obj)
  26. Reading from the queue obj, shutdown := c.workqueue.Get() if shutdown

    { return false } err := func(obj interface{}) error { defer workqueue.Done(obj) if key, ok := obj.(string); !ok { // should never happen } if err := handle(key); err != nil { workqueue.AddRateLimited(key) return fmt.Errorf(“Help”) } c.workqueue.Forget(obj) return nil }(obj)
  27. Reading from the queue obj, shutdown := c.workqueue.Get() if shutdown

    { return false } err := func(obj interface{}) error { defer workqueue.Done(obj) if key, ok := obj.(string); !ok { // should never happen } if err := handle(key); err != nil { workqueue.AddRateLimited(key) return fmt.Errorf(“Help”) } c.workqueue.Forget(obj) return nil }(obj)
  28. Reading from the queue obj, shutdown := c.workqueue.Get() if shutdown

    { return false } err := func(obj interface{}) error { defer workqueue.Done(obj) if key, ok := obj.(string); !ok { // should never happen } if err := handle(key); err != nil { workqueue.AddRateLimited(key) return fmt.Errorf(“Help”) } c.workqueue.Forget(obj) return nil }(obj) ns, name, _ := cache.SplitMetaNamespaceKey(key) pod, err := informer.Lister().Pods(namespace).Get(name) ….
  29. Reading from the queue obj, shutdown := c.workqueue.Get() if shutdown

    { return false } err := func(obj interface{}) error { defer workqueue.Done(obj) if key, ok := obj.(string); !ok { // should never happen } if err := handle(key); err != nil { workqueue.AddRateLimited(key) return fmt.Errorf(“Help”) } c.workqueue.Forget(obj) return nil }(obj)
  30. Custom Resource Definition (CRD) • A kubernetes type that describes

    other types • Let us enhance the capabilities of a k8s cluster • From great power comes great responsibility
  31. Custom Resource Definition (CRD) apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: name:

    crontabs.stable.example.com spec: group: stable.example.com versions: - name: v1 served: true storage: true schema: openAPIV3Schema: type: object properties: spec: type: object properties: cronSpec: type: string image: type: string replicas: type: integer scope: Namespaced names: plural: crontabs singular: crontab kind: CronTab shortNames: - ct
  32. Code Generation • Deep Copy • ClientSet • Informers •

    Listers k8s.io/code-generator/generate-groups.sh all \ github.com/myorg/myprj/pkg/client \ github.com/myorg/myprj/pkg/apis \ example.com:v1
  33. Code Generation tree pkg pkg └── apis └── example ├──

    register.go └── v1 ├── doc.go └── types.go
  34. Code Generation tree pkg pkg └── apis └── example ├──

    register.go └── v1 ├── doc.go └── types.go // +k8s:deepcopy-gen=package // +k8s:defaulter-gen=TypeMeta // +groupName=foo.com package v1
  35. Code Generation tree pkg pkg └── apis └── example ├──

    register.go └── v1 ├── doc.go └── types.go // +genclient // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object // TestType is a top-level type. A client is created for it. type TestType struct { metav1.TypeMeta `json:",inline"` // +optional metav1.ObjectMeta `json:"metadata,omitempty"` // +optional Status TestTypeStatus `json:"status,omitempty"` Spec TestTypeSpec `json:"spec,omitempty"` }
  36. Dynamic Client • No need for additional dependencies • Part

    of client-go (k8s.io/client-go/dynamic) • Objects as map[string]interface{} • Arrays as []interface{}
  37. Dynamic Client gvr := schema.GroupVersionResource{Group: "", Resource: "pods", Version: "v1"}

    pod, err := dynamic.Interface.Resource(gvr).Get(context.Background(), "mypod", metav1.GetOptions{}) meta := pod["meta"] metaMap, ok := meta.(map[string]interface{}) name, ok := metaMap["name"] name, found, err := unstructured.NestedString(pod.Object, "metadata", "name")
  38. Operators operator-sdk new goway-operator --repo github.com/fedepaol/goway-operator operator-sdk add api --api-version=goway.com/v1

    --kind=Sample operator-sdk add controller --api-version=goway.com/v1 --kind=Sample . ├── build │ ├── bin │ │ ├── entrypoint │ │ └── user_setup │ └── Dockerfile ├── cmd │ └── manager │ └── main.go ├── deploy │ ├── crds │ │ ├── goway.com_samples_crd.yaml │ │ └── goway.com_v1_sample_cr.yaml │ ├── operator.yaml │ ├── role_binding.yaml │ ├── role.yaml │ └── service_account.yaml ├── go.mod ├── go.sum ├── pkg │ ├── apis │ │ ├── addtoscheme_goway_v1.go │ │ ├── apis.go │ │ └── goway │ │ ├── group.go │ │ └── v1 │ │ ├── doc.go │ │ ├── register.go │ │ ├── sample_types.go │ │ └── zz_generated.deepcopy.go │ └── controller │ ├── add_sample.go │ ├── controller.go │ └── sample │ └── sample_controller.go ├── tools.go └── version └── version.go github.com/operator-framework/operator-sdk
  39. Operators err = c.Watch(&source.Kind{Type: &gowayv1.Sample{}}, &handler.EnqueueRequestForObject{}) if err != nil

    { return err } err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ IsController: true, OwnerType: &gowayv1.Sample{}, })
  40. Operators err = c.Watch(&source.Kind{Type: &gowayv1.Sample{}}, &handler.EnqueueRequestForObject{}) if err != nil

    { return err } err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ IsController: true, OwnerType: &gowayv1.Sample{}, })
  41. Operators func (r *ReconcileSample) Reconcile(request reconcile.Request) (reconcile.Result, error) { instance

    := &gowayv1.Sample{} err := r.client.Get(context.TODO(), request.NamespacedName, instance) pod := newPodForCR(instance) controllerutil.SetControllerReference(instance, pod, r.scheme); err != nil { found := &corev1.Pod{} err = r.client.Get(context.TODO(), types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}, found) if err != nil && errors.IsNotFound(err) { r.client.Create(context.TODO(), pod) return reconcile.Result{}, nil } else if err != nil { return reconcile.Result{}, err } return reconcile.Result{}, nil }
  42. Operators func (r *ReconcileSample) Reconcile(request reconcile.Request) (reconcile.Result, error) { instance

    := &gowayv1.Sample{} err := r.client.Get(context.TODO(), request.NamespacedName, instance) pod := newPodForCR(instance) controllerutil.SetControllerReference(instance, pod, r.scheme); err != nil { found := &corev1.Pod{} err = r.client.Get(context.TODO(), types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}, found) if err != nil && errors.IsNotFound(err) { r.client.Create(context.TODO(), pod) return reconcile.Result{}, nil } else if err != nil { return reconcile.Result{}, err } return reconcile.Result{}, nil }
  43. Operators func (r *ReconcileSample) Reconcile(request reconcile.Request) (reconcile.Result, error) { instance

    := &gowayv1.Sample{} err := r.client.Get(context.TODO(), request.NamespacedName, instance) pod := newPodForCR(instance) controllerutil.SetControllerReference(instance, pod, r.scheme); err != nil { found := &corev1.Pod{} err = r.client.Get(context.TODO(), types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}, found) if err != nil && errors.IsNotFound(err) { r.client.Create(context.TODO(), pod) return reconcile.Result{}, nil } else if err != nil { return reconcile.Result{}, err } return reconcile.Result{}, nil }
  44. Operators func (r *ReconcileSample) Reconcile(request reconcile.Request) (reconcile.Result, error) { instance

    := &gowayv1.Sample{} err := r.client.Get(context.TODO(), request.NamespacedName, instance) pod := newPodForCR(instance) controllerutil.SetControllerReference(instance, pod, r.scheme); err != nil { found := &corev1.Pod{} err = r.client.Get(context.TODO(), types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}, found) if err != nil && errors.IsNotFound(err) { r.client.Create(context.TODO(), pod) return reconcile.Result{}, nil } else if err != nil { return reconcile.Result{}, err } return reconcile.Result{}, nil }
  45. Operators func (r *ReconcileSample) Reconcile(request reconcile.Request) (reconcile.Result, error) { instance

    := &gowayv1.Sample{} err := r.client.Get(context.TODO(), request.NamespacedName, instance) pod := newPodForCR(instance) controllerutil.SetControllerReference(instance, pod, r.scheme); err != nil { found := &corev1.Pod{} err = r.client.Get(context.TODO(), types.NamespacedName{Name: pod.Name, Namespace: pod.Namespace}, found) if err != nil && errors.IsNotFound(err) { r.client.Create(context.TODO(), pod) return reconcile.Result{}, nil } else if err != nil { return reconcile.Result{}, err } return reconcile.Result{}, nil }
  46. Operator Lifecycle Manager • Dependency Management • Upgrades Management •

    “Marketplace” (operatorhub.io) • Multitenancy