云原生｜详解Kubernetes Operator在项目中的开发应用

使用场景

controller-runtime是基于kubernetes控制器模式衍生出来的一套框架。控制回路（Control Loop）是一个非终止回路，用于调节系统状态。一个控制器至少追踪一种类型的 Kubernetes 资源。这些 对象 有一个代表期望状态的 spec 字段。该资源的控制器负责确保其当前状态接近期望状态。

从中，我们可以得出，controller-runtime开发需要遵循以下几点。

1. controller-runtime需要持续监听kubernetes对象。（通过list/watch机制实现）

   
   

2. controller-runtime需要提供某些机制，Kubernetes对象的status（实际状态）维持到与spec（期望状态）一致。（通过reconcile实现）

一、client-go中处理逻辑

在client-go中，会将一种Kubernetes对象封装成一个对的informer类型。informer类型中包含Reflector，DeltaFIFO，Indexer，EventHandler。

• Reflector：用于调用Kubernetes的API接口。informer会通过两个Kubernetes的API接口来实现kubernetes对象的监听。list/watch失败时，会自动重新list/watch

  
  

• list接口：通过http请求，获取informer中定义的全量kubernetes对象数据到deltaFIFO中

  
  

• watch接口：从本地存储的最大resourceVersion开始监听。当发送 watch 请求时，API 服务器会响应更改流。这些更改逐项列出了在你指定为 watch 请求参数的 resourceVersion 之后发生的操作（例如 ADDED、MOD-IFIED、DELETED、BOOKMARK、ERROR）的结果。

  
  

• DeltaFIFO：存储每个待处理的Kubernetes对象。DeltaFIFO中有两个存储kubernetes对象的数据结构，一个是存储对象key值的一维数组，一个存储数据流的map结构。

  
  

• queue: 一维数组，每个kubernetes对象对应一个key（namespace/name）。多次入队的变更数据会合并存储，出队时，处理整个对象。

  
  

• items: 存储kubernetes对象的变更流。（Added，Updated，Deleted，Replaced，Sync）启动informer时，持续从deltaFIFO中读取数据进行处理。

  
  

• Indexer：从DeltaFIFO中获取的Kubernetes对象，存储到内存map中。

  
  

• EventHandler：从DeltaFIFO中获取的Kubernetes对象，根据对象的变更事件类型，来进行不同的处理。一般是将key值传递给工作队列。然后通过custom controller进行处理。

二、controller-runtime中处理逻辑

根据要监听的Kubernetes对象，定义Informer

定义EventHandler处理逻辑。将informer eventhandler中的key值存入工作队列。该工作队列中会对key值进行去重处理。以避免在一个工作队列出现并发处理同一个对象的情况。

从工作队列中取出key值，在reconcile进行处理。

自定义结构体实现reconcile方法。在该方法中，协调kubernetes对象的状态，使其向期望状态靠拢。

controller-runtime框架简介

controller-runtime顶层使用manager进行封装。manager包含以下信息

• cluster：用于与kubernetes集群通信。包含kubernetes的kubeconfig配置信息，连接kuberenetes的apiserver的客户端信息，注册到operator的kubernetes对象scheme信息，用于获取gvk和gvr的RESTMapper信息，用于informer存储的indexer信息，用于和kubernetes对象读，写和事件发布的客户端信息。

  
  

• controller：用于informer的配置与启动，reconcile的定义

  
  

• cache：缓存informer信息。包含informer的定义和indexer数据读取。

  
  

• webhook：用于发布kubernetes中的三种类型的webhook服务。

  
  

• election：用于operator的主备配置

  
  

• prometheus：用于发布prometheus的数据服务

  
  

• probe：定义operator的探针服务，包含两种探针数据：健康探针和存活探针

使用controller-runtime开发operator项目

mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
   Scheme: scheme,
   MetricsBindAddress:     metricsAddr,
   Port:                   9443,
   HealthProbeBindAddress: probeAddr,
   LeaderElection: enableLeaderElection,
   LeaderElectionID: "aefd3536.demo",
})

if err = (&controllers.GuestbookReconciler{
   Client: mgr.GetClient(),
   Scheme: mgr.GetScheme(),
}).SetupWithManager(mgr); err != nil {
   setupLog.Error(err, "unable to create controller", "controller", "Guestbook")
   os.Exit(1)
}

if err = (&webappv1.Guestbook{}).SetupWebhookWithManager(mgr); err != nil {
   setupLog.Error(err, "unable to create webhook", "webhook", "Guestbook")
   os.Exit(1)
}

if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
   setupLog.Error(err, "unable to set up health check")
   os.Exit(1)
}
if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
   setupLog.Error(err, "unable to set up ready check")
   os.Exit(1)
}

if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
   setupLog.Error(err, "problem running manager")
   os.Exit(1)
}

func (r *GuestbookReconciler) SetupWithManager(mgr ctrl.Manager) error {
   return ctrl.NewControllerManagedBy(mgr).
      For(&webappv1.Guestbook{}).
      WithEventFilter(predicate.Funcs{
         CreateFunc: func(_ event.CreateEvent) bool {
            return false
         },
      }).
      WithOptions(controller.Options{MaxConcurrentReconciles: 2}).
      Complete(r)
}

在main.go文件中调用，用于定义informer并将informer配置到manager中

• informer中kubernetes对象

  
  

框架对对象类型的定义提供了三种方法：

· Watches(source.Source, handler.EventHandler, ...WatchesOption)

· For(client.Object,...ForOption)

· Owns(client.Object,...OwnsOption)

其中For和Owns是等同与Watches。

For的第二个参数默认为EnqueueRequestForObject。

Owns的第二个参数默认为EnqueueRequestForOwner

方法参数说明

Source：第一个参数，kubernetes对象类型

EventHandler：第二个参数，从DeltaFIFO中取出来的数据，在进入工作队列前进行的操作。EnqueueRequestForObject表示不做任何处理，直接进入工作队列。EnqueueRequestForOwner需要和For方法配合使用，Owns中的对象中ownerReference引用的对象类型需要和For中定义的对象类型相同，且ownerReference中的controller为true。

Predicate：第三个参数，从工作队列取出来的数据，在进行reconcile处理前进行的操作。通过builder的WithEventFilter可以给所有的对象添加Predicate。

EventHandler和Predicate方法说明

Create：kubernetes对象新增时调用

Update：kubernetes对象更新时调用

Delete：kubernetes对象删除时调用

Generic：未知的操作。非kubernetes集群的变更事件。在operator中自行使用

• controller配置

  
  

通过builder.WithOptions定义配置。controller中有以下配置可定义。

MaxConcurrentReconciles：从工作队列取出的数据，可使用几个协程进行处理。默认一个

Reconciler：协调器，需要实现Reconciler方法。默认为guest-book_controller.go文件中定义的Reconcile

RateLimiter：工作队列的限流策略。默认有两种限流策略。令牌桶策略（BucketRateLimiter）：默认100个令牌数量。每秒可取10个令牌。退避策略（ItemExponentialFailureRateLimiter）：reconcile返回失败时，重新入队。下次出队时间按2的指数次方增长。初始5ms，最大1000s。

LogConstructor：reconcile中从context中取到的日志logr。默认是mgr中定义的日志。

CacheSyncTimeout：informer同步超时时间。在informer进行list/watch时，只有当list接口调用的数据全部同步到informer的indexer后，才算informer同步成功。

RecoverPanic：reconcile异常时是否自动恢复。

func (r *GuestbookReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
   _ = log.FromContext(ctx)
 
   // TODO(user): your logic here
 
   return ctrl.Result{}, nil
}

func (r *Guestbook) Default() {
   guestbooklog.Info("default", "name", r.Name)
 
   // TODO(user): fill in your defaulting logic.
}

func (r *Guestbook) ValidateCreate() error {
   guestbooklog.Info("validate create", "name", r.Name)
 
   // TODO(user): fill in your validation logic upon object creation.
   return nil
}
 
// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
func (r *Guestbook) ValidateUpdate(old runtime.Object) error {
   guestbooklog.Info("validate update", "name", r.Name)
 
   // TODO(user): fill in your validation logic upon object update.
   return nil
}
 
// ValidateDelete implements webhook.Validator so a webhook will be registered for the type
func (r *Guestbook) ValidateDelete() error {
   guestbooklog.Info("validate delete", "name", r.Name)
 
   // TODO(user): fill in your validation logic upon object deletion.
   return nil
}

convertion.Webhook

对应crd中的spec.convertion中定义的webhook

中心版本

// Hub 标注为中心版本
func (r *Guestbook) Hub() {
 
}

func (r *Guestbook) ConvertTo(dst conversion.Hub) error {
   // todo 当前版本向中心版本转换
   return nil
}
func (r *Guestbook) ConvertFrom(src conversion.Hub) error {
   // todo 中心版本向当前版本转换
   return nil
}