【k8s】——自定义的k8s controller

川长思鸟来 2022-10-22 05:25 268阅读 0赞

### 一、前言 ###

在 K8s 中当我们需要监控某个资源的变化并作一系列操作时，使用 K8s 提供的 controller 机制来实现，同时 K8s 官方提供了一个通用库 `client-go`，通过它可以很容易实现自定义controller.`Custom Resource`是扩展Kubernetes的一种方式（另外一种就是通过聚合层API `apiserver-aggregation`），而controller对指定的resource进行监听和执行对应的动作(watch,diff,action)。这里引申一点，我们一般来说要写controller，大多数时候是写operator，但operator和controller不能画等号，区别如下：

*  所有的Operator都是用了Controller模式，但并不是所有Controller都是Operator。只有当它满足: controller模式 + API扩展 + 专注于某个App/中间件时，才是一个Operator。
 *  Operator就是使用CRD实现的定制化的Controller. 它与内置K8S Controller遵循同样的运行模式(比如 watch, diff, action)
 *  Operator是特定领域的Controller实现

### 二、原理 ###

在编写controller 之前，我们需要了解 client-go 对资源监控的整个架构和流程，并需要知道我们所需要自定义的是哪部分组件。在 client-go 中包含了编写自定义 controller 可以使用的各种机制，这些机制在库的`k8s.io/client-go/tools` `k8s.io/client-go/util`中定义。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ0OTMwMA_size_16_color_FFFFFF_t_70][]

这张架构图以虚线为界限分为上下两部分，上边的部分是client-go来实现的，下边的部分就是我们的coustom controller来完成的工作，下边来介绍一下这两部分：

**1. client-go**

在client-go 中主要包含了以下组件：

*  Reflector：通过调用 `LISTWATCH` 接口与 ApiServer 通信，监听特定资源（此处监听的资源需要我们指定），并把资源的更新动态存入 Delta FIFO 队列
 *  Informer：从Delta FIFO队列拿出对象，完成此操作的函数是processLoop。
 *  Indexer: 提供线程级别安全来存储对象和key。

**2. custom controller**

*  Informer reference: Informer对象引用
 *  Indexer reference: Indexer对象引用
 *  Resource Event Handlers: 被Informer调用的回调函数，这些函数的作用通常是获取对象的key，并把key放入Work queue，以进一步做处理。
 *  Work queue: 工作队列，用于将对象的交付与其处理分离，编写Resource event handler functions以提取传递的对象的key并将其添加到工作队列。**此处可以过滤掉我们不关心的信息。**
 *  Process Item: 用于处理Work queue中的对象，可以有一个或多个其他函数一起处理；这些函数通常使用Indexer reference或Listing wrapper来检索与该键对应的对象。**这里就是我们需要自定义的业务逻辑**

**流程说明：**

黄色的部分是controller相关的框架，包括workqueue。蓝色部分是client-go的相关内容，包括informer, reflector(其实就是informer的封装), indexer。从流程上看：reflector从apiserver中通过list&watch机制接收事件变化；进入Delta FIFO队列中，由informer进行处理。informer会将delta FIFO队列中的事件交给indexer组件，indexer组件会将事件持久化存储在本地的缓存中。之后，由于用户事先将为informer注册各种事件的回调函数，这些回调函数将针对不同的组件做不同的处理。例如在controller中，将把object放入workqueue中，之后由controller的业务逻辑中进行处理。处理的时候将从缓存中获取object的引用。即各组件对资源的处理仅限于本地缓存中，直到update资源的时候才与apiserver交互。

**总结：简单来讲通过list/watch机器提供了本地缓存避免每次去请求apiserver。并且提供了Event Handler方法，在将数据保存进入cache时，通过调用自定义handler方法，增加自定义处理。**

### 三、牛刀小试 ###

编写一个简易的 Controller， 用于监听 pod 创建、删除信息，并将信息打印出来。

**1. Controller 逻辑**

1.1 定义一个这样的 Controller 结构体

type Controller struct {
        indexer  cache.Indexer  // Indexer 的引用
        queue    workqueue.RateLimitingInterface  //workqueue 的引用
        informer cache.Controller // Informer 的引用
    }

1.2 定义 Controller 的工作流

func (c *Controller) Run(threadiness int, stopCh chan struct{}) {
        defer runtime.HandleCrash()
    
        defer c.queue.ShutDown()
    
        klog.Info("Starting pod controller")
    
        go c.informer.Run(stopCh)   // 启动 informer
    
        if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {
            runtime.HandleError(fmt.Errorf("Time out waitng for caches to sync"))
            return
        }
    
        // 启动多个 worker 处理 workqueue 中的对象
        for i := 0; i < threadiness; i++ {
            go wait.Until(c.runWorker, time.Second, stopCh)
        }
    
        <-stopCh
        klog.Info("Stopping Pod controller")
    }

1.3 具体处理 worker queue 中对象的流程

func (c *Controller) runWorker() {
        // 启动无限循环，接收并处理消息
        for c.processNextItem() {
    
        }
    }
    // 从 workqueue 中获取对象，并打印信息。
    func (c *Controller) processNextItem() bool {
        key, shutdown := c.queue.Get()
        // 退出
        if shutdown {
            return false
        }
    
        // 标记此key已经处理
        defer c.queue.Done(key)
    
        // 将key对应的 object 的信息进行打印
        err := c.syncToStdout(key.(string))
    
        c.handleError(err, key)
        return true
    }
    
    // 获取 key 对应的 object，并打印相关信息
    func (c *Controller) syncToStdout(key string) error {
        obj, exists, err := c.indexer.GetByKey(key)
        if err != nil {
            klog.Errorf("Fetching object with key %s from store failed with %v", key, err)
            return err
        }
        if !exists {
            fmt.Printf("Pod %s does not exist")
        } else {
            fmt.Printf("Sync/Add/Update for Pod %s\n", obj.(*core_v1.Pod).GetName())
        }
        return nil
    }

**2. Main 函数逻辑**

func main() {
        var kubeconfig string
        var master string
        // 从外部获取集群信息(kube.config)
        flag.StringVar(&kubeconfig, "kubeconfig", "", "kubeconfig file")
        // 获取集群master 的url
        flag.StringVar(&master, "master", "", "master url")
    
        // 读取构建 config
        config, err := clientcmd.BuildConfigFromFlags(master, kubeconfig)
        if err != nil {
            klog.Fatal(err)
        }
    
        // 创建 k8s client
        clientset, err := kubernetes.NewForConfig(config)
        if err != nil {
            klog.Fatal(err)
        }
    
        // 指定 ListWatcher 在所有namespace下监听 pod 资源
        podListWatcher := cache.NewListWatchFromClient(clientset.CoreV1().RESTClient(), "pods", v1.NamespaceAll, fields.Everything())
    
        // 创建 workqueue
        queue := workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())

### 四、参考资料 ###

1. [https://zhuanlan.zhihu.com/p/141877334][https_zhuanlan.zhihu.com_p_141877334]

2. [https://segmentfault.com/a/1190000021934073][https_segmentfault.com_a_1190000021934073]

3. [https://blog.csdn.net/shida\_csdn/article/details/87102801][https_blog.csdn.net_shida_csdn_article_details_87102801]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDQ0OTMwMA_size_16_color_FFFFFF_t_70]: /images/20221022/261e46065898484cab6ed707b7e56e87.png
[https_zhuanlan.zhihu.com_p_141877334]: https://zhuanlan.zhihu.com/p/141877334
[https_segmentfault.com_a_1190000021934073]: https://segmentfault.com/a/1190000021934073
[https_blog.csdn.net_shida_csdn_article_details_87102801]: https://blog.csdn.net/shida_csdn/article/details/87102801