自学内容网 自学内容网

Kubebuilder 架构介绍

1、Kubebuilder 介绍

Kubebuilder 是一个用 Go 语言构建 Kubernetes APIs 的框架,通过使用 Kubebuilder,用户可以遵循一套简单的编程框架,使用 CRD 构建API、Controllers 和 Admission WebHooks,实现对 k8s 的扩展。

Kubebuilder 中的主要组件包含 Manager、Cache、Client 与 Finalizers。

  • Manager组件主要实现管理外层,负责初始化 Controller、Cache、Client的工作;
  • Cache 组件负责生成 ShareInformer,Watch关注的GVK下的GVR的变化(增、删、改),以触发Controller 的Reconcile逻辑;
  • Client 组件在工作中实现对资源进行 CURD 操作,CURD操作封装到Client中进行,其中的写操作(增、删、改)直接访问APIServer,读操作(查)对接的是本地的Cache;
  • Finzlizers组件主要用于处理Kubernetes资源的预删除逻辑,保障资源被删除后能够从Cache中读取到,清理相关的其它资源。

Kubebuilder 是个脚手架中间,是将 Kubernetes 的可扩展能力CRD进行了简化封装。kubebuilder 大致划分为四大块:User Defined、API Scaffolds、Controller Runtime、Kubernetes集群。

当自定义好的 CRD 结构,想要在 Kubernetes 集群中实现这样的 CRD 结构定义, 这时候需要 Reconcile 去协调。定义好的 CRD 需要安装然后运行 Controller,这个 Controller 的运行是通过 Controller Runtime 库实现。

在 Controller Runtime 模块中,k8s 构建出来的 CRD 会注册到 Scheme 模块,它会提供 Kinds 与对应的 Go Type 的映射,就能知道它的 GKV(Group Kind Version)。

2、Kubebuilder 模块分析

2.1、CRD 创建命令
kubebuilder create api --group demo --version v1 --kind Demo
kubebuilder create api --group ship --version v1beta1 --kind Test1
kubebuilder create api --group ship --version v1beta1 --kind Test2
  • 自定义 CRD,Group 表示 CRD 所属的组,它可以支持多种不同版本、不同类型的资源构建;

  • Version 表示 CRD 的版本号;

  • Kind 表示 CRD 的类型;

上面创建了 1 个 v1 版本的 Demo 类型的资源,它会自动生成了 {kind}types.go 的文件, 即 demo_types.go ;同时创建了 2 个 v1beta1 版本的不 同类型的资源, 可以看到生成了 2 个资源文件, 分别是 test1_types.go、test2_types.go。目录如下:

➤ tree api/                                                                                                                                                    
api/
├── v1
│   ├── demo_types.go
│   ├── groupversion_info.go
│   └── zz_generated.deepcopy.go
└── v1beta1
    ├── groupversion_info.go
    ├── test1_types.go
    ├── test2_types.go
    └── zz_generated.deepcopy.go

demo_types.go 文件里面内容

type Demo struct {  
   metav1.TypeMeta   `json:",inline"`  
   metav1.ObjectMeta `json:"metadata,omitempty"`  
  
   Spec   DemoSpec   `json:"spec,omitempty"`  
   Status DemoStatus `json:"status,omitempty"`  
}

type DemoList struct {  
   metav1.TypeMeta `json:",inline"`  
   metav1.ListMeta `json:"metadata,omitempty"`  
   Items           []Demo `json:"items"`  
}

func init() {  
   SchemeBuilder.Register(&Demo{}, &DemoList{})  
}

Demo 资源结构体包含:Metadata、Spec、Status、及继承的 Kubernetes 资源属性, 如 kind、 apiVersion 等

DemoList 表明资源的列表结构体,当用户查询这一类资源时,各 demo 的内容放在了 Items 键的下面。

init() 初始化函数的作用是将资源的类型注册到 Scheme 对应的 demo 组的 v1 版本下。

每一个 CRD, 默认会创建对应的 {kind}controller.go 文件, 如 demo_controller.go, 这 就是 CRD Controller 逻辑构造的了。

➤ tree controllers/                                                                                                                                            
controllers/
├── demo_controller.go
├── suite_test.go
├── test1_controller.go
└── test2_controller.go

demo_controller.go 代码里面,自动生成的 Reconciler 的对象名称是 {kind}Reconciler,它的主方法是 Reconcile(),即通过在这个函数的空白处填入逻辑完成对应的 CRD 构造工作。SetupWithManager 方法作用是用于 CRD Controller 的安装,这样CRD Controller 才能运行。、


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

// 它用于 CRD Controller 的安装。安 装完成后,CRD Controller 才能运行,  
func (r *DemoReconciler) SetupWithManager(mgr ctrl.Manager) error {  
   return ctrl.NewControllerManagedBy(mgr).  
      For(&demov1.Demo{}).  
      Complete(r)  
}
2.2、Manager 初始化

在 main 文件,Manager 初始化是借助于 ctrl.NewManager 方法实现,进去原来是 controller-runtime 包的 manager.New方法。


mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{  
   Scheme:                 scheme, // 将 crd 加入到scheme 中  
   MetricsBindAddress:     metricsAddr,  
   Port:                   9443,  
   HealthProbeBindAddress: probeAddr,  
   LeaderElection:         enableLeaderElection,  
   LeaderElectionID:       "ecaf1259.my.domain",  
})
if err != nil {  
   setupLog.Error(err, "unable to start manager")  
   os.Exit(1)  
}

在 New 方法中,实际是根据传入的参数 进行 Manager 对象的 Scheme、Cache、Client 等模块的初始化构建。

// sigs.k8s.io/controller-runtime@v0.13.0/pkg/manager/manager.go
func New(config *rest.Config, options Options) (Manager, error) {  
   // Set default values for options fields  
   options = setOptionsDefaults(options)  
  
   cluster, err := cluster.New(config, func(clusterOptions *cluster.Options) {  
      clusterOptions.Scheme = options.Scheme  
      clusterOptions.MapperProvider = options.MapperProvider  
      clusterOptions.Logger = options.Logger  
      clusterOptions.SyncPeriod = options.SyncPeriod  
      clusterOptions.Namespace = options.Namespace  
      clusterOptions.NewCache = options.NewCache  
      clusterOptions.NewClient = options.NewClient  
      clusterOptions.ClientDisableCacheFor = options.ClientDisableCacheFor  
      clusterOptions.DryRunClient = options.DryRunClient  
      clusterOptions.EventBroadcaster = options.EventBroadcaster //nolint:staticcheck  
   })  
   if err != nil {  
      return nil, err  
   }  
    
   recorderProvider, err := options.newRecorderProvider(config, cluster.GetScheme(), options.Logger.WithName("events"), options.makeBroadcaster)  
   if err != nil {  
      return nil, err  
   }  
  
   // Create the resource lock to enable leader election)  
   var leaderConfig *rest.Config  
   var leaderRecorderProvider *intrec.Provider  
  
   if options.LeaderElectionConfig == nil {  
      leaderConfig = rest.CopyConfig(config)  
      leaderRecorderProvider = recorderProvider  
   } else {  
      leaderConfig = rest.CopyConfig(options.LeaderElectionConfig)  
      leaderRecorderProvider, err = options.newRecorderProvider(leaderConfig, cluster.GetScheme(), options.Logger.WithName("events"), options.makeBroadcaster)  
      if err != nil {  
         return nil, err  
      }  
   }  
  
   resourceLock, err := options.newResourceLock(leaderConfig, leaderRecorderProvider, leaderelection.Options{  
      LeaderElection:             options.LeaderElection,  
      LeaderElectionResourceLock: options.LeaderElectionResourceLock,  
      LeaderElectionID:           options.LeaderElectionID,  
      LeaderElectionNamespace:    options.LeaderElectionNamespace,  
   })  
   if err != nil {  
      return nil, err  
   }  
  
   // Create the metrics listener. This will throw an error if the metrics bind  
   // address is invalid or already in use.   
   metricsListener, err := options.newMetricsListener(options.MetricsBindAddress)  
   if err != nil {  
      return nil, err  
   }  
  
   // By default we have no extra endpoints to expose on metrics http server.  
   metricsExtraHandlers := make(map[string]http.Handler)  
  
   // Create health probes listener. This will throw an error if the bind  
   // address is invalid or already in use.   
   healthProbeListener, err := options.newHealthProbeListener(options.HealthProbeBindAddress)  
   if err != nil {  
      return nil, err  
   }  
  
   errChan := make(chan error)  
   runnables := newRunnables(options.BaseContext, errChan)  
  
   return &controllerManager{...}, nil  
}
2.3、Manager 启动

启动 Cache

// sigs.k8s.io/controller-runtime@v0.13.0/pkg/cache/multi_namespace_cache.go
func (c *multiNamespaceCache) Start(ctx context.Context) error {  
   // start global cache  
   go func() {  
      err := c.clusterCache.Start(ctx)  
      if err != nil {  
         log.Error(err, "cluster scoped cache failed to start")  
      }  
   }()  
  
   // start namespaced caches  
   for ns, cache := range c.namespaceToCache {  
      go func(ns string, cache Cache) {  
     // namespaceToCache 存储每个 ns 的 cache,默认是 InformersMap 类型
         err := cache.Start(ctx)  // chach 入口
         if err != nil {  
            log.Error(err, "multinamespace cache failed to start namespaced informer", "namespace", ns)  
         }  
      }(ns, cache)  
   }  
  
   <-ctx.Done()  
   return nil  
}

InformersMap 抽象出 3 个 Map 结构,存储不同的 Informer

func (m *InformersMap) Start(ctx context.Context) error {  
   go m.structured.Start(ctx)  
   go m.unstructured.Start(ctx)  
   go m.metadata.Start(ctx)  
   <-ctx.Done()  
   return nil  
}

启动每个 informer


func (ip *specificInformersMap) Start(ctx context.Context) {  
   func() {  
      ip.mu.Lock()  
      defer ip.mu.Unlock()  
  
      // Set the stop channel so it can be passed to informers that are added later  
      ip.stop = ctx.Done()  
  
      // Start each informer  
      for _, informer := range ip.informersByGVK {  
         go informer.Informer.Run(ctx.Done())  
      }  
  
      // Set started to true so we immediately start any informers added later.  
      ip.started = true  
      close(ip.startWait)  
   }()  
   <-ctx.Done()  
}

Cache 的核心逻辑是初始化内部所有的 Informer, 初始化 Informer 后就创建了 Reflector 和内部 Controller,Reflector 和 Controller 两个组件是一个“生产者—消费者” 模型,Reflector 负责监听 APIServer 上指定的 GVK 资源的变化,然后将变更写入 delta 队列中,Controller 负责消费这些变更的事件,然后更新本地 Indexer,最后计算出是创建、 更新,还是删除事件,推给我们之前注册的 Watch Handler。

2.4、Controller 初始化

Kubebuilder 脚手架生成后 controller 文件里,CRD 的 Controller 初始化的核心代码是 SetupWithManager 方法。

// 它用于 CRD Controller 的安装。安装完成后,CRD Controller 才能运行,  
func (r *DemoReconciler) SetupWithManager(mgr ctrl.Manager) error {  
   return ctrl.NewControllerManagedBy(mgr).  
      For(&demov1.Demo{}).  
      Complete(r)  
}

完成 CRD 在 Manager 对象中的安装,最后通过 Manager 对象的 start 方法来完成 CRD Controller 的运行。

使用 Controller-runtime 包初始化 Builder 对象,当它完成 Complete 方法时,实 际完成了 CRD Reconciler 对象的初始化,而这个对象是一个接口方法,它必须实现 Reconcile 方法。


type Builder struct {  
   forInput         ForInput  
   ownsInput        []OwnsInput  
   watchesInput     []WatchesInput  
   mgr              manager.Manager  
   globalPredicates []predicate.Predicate  
   ctrl             controller.Controller  
   ctrlOptions      controller.Options  
   name             string  
}

func ControllerManagedBy(m manager.Manager) *Builder {  
   return &Builder{mgr: m}  
}

func (blder *Builder) Complete(r reconcile.Reconciler) error {  
   _, err := blder.Build(r)  
   return err  
}  
  
// Build builds the Application Controller and returns the Controller it created.
func (blder *Builder) Build(r reconcile.Reconciler) (controller.Controller, error) {  
   if r == nil {  
      return nil, fmt.Errorf("must provide a non-nil Reconciler")  
   }  
   if blder.mgr == nil {  
      return nil, fmt.Errorf("must provide a non-nil Manager")  
   }  
   if blder.forInput.err != nil {  
      return nil, blder.forInput.err  
   }  
   // Checking the reconcile type exist or not  
   if blder.forInput.object == nil {  
      return nil, fmt.Errorf("must provide an object for reconciliation")  
   }  
  
   // Set the ControllerManagedBy  
   if err := blder.doController(r); err != nil {  
      return nil, err  
   }  
  
   // Set the Watch  
   if err := blder.doWatch(); err != nil {  
      return nil, err  
   }  
  
   return blder.ctrl, nil  
}

原文地址:https://blog.csdn.net/realize_dream/article/details/144322216

免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!