ToB企服应用市场:ToB评测及商务社交产业平台

标题: client-go实战之九：手写一个kubernetes的controller [打印本页]
作者: 张春    时间: 2023-11-14 06:15
标题: client-go实战之九：手写一个kubernetes的controller
欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码)：https://github.com/zq2599/blog_demos
本篇概览

代码整体架构一览

对着架构细说流程



编码规划

编码之一：定义Controller数据结构(controller.go)

  1. type Controller struct {
  2.         indexer  cache.Indexer
  3.         queue    workqueue.RateLimitingInterface
  4.         informer cache.Controller
  5. }
复制代码
编码之二：编写业务逻辑代码(controller.go)


  1. func (c *Controller) processNextItem() bool {
  2.         // 阻塞等待，直到队列中有数据可以被取出，
  3.         // 另外有可能是多协程并发获取数据，此key会被放入processing中，表示正在被处理
  4.         key, quit := c.queue.Get()
  5.         // 如果最外层调用了队列的Shutdown，这里的quit就会返回true，
  6.         // 调用processNextItem的地方发现processNextItem返回false，就不会再次调用processNextItem了
  7.         if quit {
  8.                 return false
  9.         }
  11.         // 表示该key已经被处理完成(从processing中移除)
  12.         defer c.queue.Done(key)
  14.         // 调用业务方法，实现具体的业务需求
  15.         err := c.syncToStdout(key.(string))
  16.         // Handle the error if something went wrong during the execution of the business logic
  18.         // 判断业务逻辑处理是否出现异常，如果出现就重新放入队列，以此实现重试，如果已经重试过5次，就放弃
  19.         c.handleErr(err, key)
  21.         // 调用processNextItem的地方发现processNextItem返回true，就会再次调用processNextItem
  22.         return true
  23. }
复制代码
  1. func (c *Controller) syncToStdout(key string) error {
  2.         // 根据key从本地存储中获取完整的pod信息
  3.         // 由于有长连接与apiserver保持同步，因此本地的pod信息与kubernetes集群内保持一致
  4.         obj, exists, err := c.indexer.GetByKey(key)
  5.         if err != nil {
  6.                 klog.Errorf("Fetching object with key %s from store failed with %v", key, err)
  7.                 return err
  8.         }
  10.         if !exists {
  11.                 fmt.Printf("Pod %s does not exist anymore\n", key)
  12.         } else {
  13.                 // 这里就是真正的业务逻辑代码了，一般会比较spce和status的差异，然后做出处理使得status与spce保持一致，
  14.                 // 此处为了代码简单仅仅打印一行日志
  15.                 fmt.Printf("Sync/Add/Update for Pod %s\n", obj.(*v1.Pod).GetName())
  16.         }
  17.         return nil
  18. }
复制代码
编码之三：编写错误处理代码(controller.go)

  1. func (c *Controller) handleErr(err error, key interface{}) {
  2.         // 没有错误时的处理逻辑
  3.         if err == nil {
  4.                 // 确认这个key已经被成功处理，在队列中彻底清理掉
  5.                 // 假设之前在处理该key的时候曾报错导致重新进入队列等待重试，那么也会因为这个Forget方法而不再被重试
  6.                 c.queue.Forget(key)
  7.                 return
  8.         }
  10.         // 代码走到这里表示前面执行业务逻辑的时候发生了错误，
  11.         // 检查已经重试的次数，如果不操作5次就继续重试，这里可以根据实际需求定制
  12.         if c.queue.NumRequeues(key) < 5 {
  13.                 klog.Infof("Error syncing pod %v: %v", key, err)
  14.                 c.queue.AddRateLimited(key)
  15.                 return
  16.         }
  18.         // 如果重试超过了5次就彻底放弃了，也像执行成功那样调用Forget做彻底清理（否则就没完没了了）
  19.         c.queue.Forget(key)
  20.         // 向外部报告错误，走通用的错误处理流程
  21.         runtime.HandleError(err)
  22.         klog.Infof("Dropping pod %q out of the queue: %v", key, err)
  23. }
复制代码
编码之四：编写Controller主流程(controller.go)

  1. func NewController(queue workqueue.RateLimitingInterface, indexer cache.Indexer, informer cache.Controller) *Controller {
  2.         return &Controller{
  3.                 informer: informer,
  4.                 indexer:  indexer,
  5.                 queue:    queue,
  6.         }
  7. }
复制代码
  1. func (c *Controller) runWorker() {
  2.         for c.processNextItem() {
  3.         }
  4. }
复制代码
[code]func (c *Controller) Run(workers int, stopCh chan struct{}) {        defer runtime.HandleCrash()        // 只要工作队列的ShutDown方法被调用，processNextItem方法就会返回false，runWorker的无限循环就会结束        defer c.queue.ShutDown()        klog.Info("Starting Pod controller")        // informer的Run方法执行后，就开始接受apiserver推送的资源变更事件，并更新本地存储        go c.informer.Run(stopCh)        // 等待本地存储和apiserver完成同步        if !cache.WaitForCacheSync(stopCh, c.informer.HasSynced) {                runtime.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))                return        }        // 启动worker，并发从工作队列取数据，然后执行业务逻辑        for i := 0; i < workers; i++ {                go wait.Until(c.runWorker, time.Second, stopCh)        }



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4