controller-runtime是通过informer监控apiserver来获取资源的变动。
接下来我会使用kubebuilder构建一个demo，根据这个demo进行深入解析controller-runtime
Demo创建kubebuilder官网文档 （具体操作还请参考官网文档，我只写出几个我认为需要说明一下的地方）
安装kubebuilder 一定要给kubebuilder 运行权限
创建并初始化项目
kubebuilder init --domain my.domain --repo my.domain/guestbook # 这里官网有注意事项说明，请移步官网 # 这里一般不需要配置东西直接下一步即可
创建API
kubebuilder create api --group webapp --version v1 --kind Guestbook # kind 后面 第一个字母要大写否则报错 # 调用这一步后，除了crd YAMl文件没有生成以外其余基本上都已生成，这时候如果想自定义crd，那么仅修改api/v1/ 下面的 *_type.go文件以及相关联的go文件即可
生成CRD
make manifests
将CRD安装到集群
make install
运行生成的控制器
make run # 如果想实现控制器的功能，需要修改controllers/*_controller.go 文件里面的 Reconcile 方法即可 # 如果想在方法里面操控别的资源需要添加权限，在该文件里面添加注释即可 # 比如//+kubebuilder:rbac:groups=webapp.my.domain,resources=guestbooks,verbs=get;list;watch;create;update;patch;delete 就是对guestbooks 资源查询创建等权限
那么接下来让我们讲解一下控制器run后运行吧
controller-runtime启动步骤大体如下
创建Manager
创建controller
如果想使用webhook，则创建webhook。
添加心跳检查 （这里可以可以实现自己的函数）
添加就绪检查
开始启动manager
注意：这里有一个类非常重要 Scheme，该类提供了GVK到TYPE，TYPE到GVK的功能，可以调用runtime.NewScheme()进行获取(可以理解成，序列化GVK与反序列化)
Manager 的创建调用NewManager
ctrl.NewManager(config *rest.Config, options Options) // 第一参数记录了client所需要的配置信息比如Host,Username等。读取--kubeconfig 上的配置，如果没有会尝试集群里的kubconfig(Home/.kube/config),都没有会读取默认配置 // 第二个参数是创建Manager的配置信息 // 第二参数 在看生成的代码的时候我们会发现有这么一个配置LeaderElectionID: "ecaf1259.my.domain" 这个配置就是竞选时的资源锁名称，leader竞争时默认使用configmap资源作为资源锁，那么我们调用kubectl get configmap -n guestbook-system 时就可以看到这个ecaf1259.my.domain 资源
2.进入NewManager
func New(config *rest.Config, options Options) (Manager, error) {
// 对options进行填充工作，没有配置的会添加默认值
options = setOptionsDefaults(options)
// 创建cluser 并且创建 cache 绑定informer
// 由于内容过长该函数参数不展示
cluster, err := cluster.New()
// 获取leader配置文件 如果没有就使用默认配置 可以通过启动时配置
leaderConfig := options.LeaderElectionConfig
if leaderConfig == nil {
leaderConfig = rest.CopyConfig(config)
}
// 竞选锁
resourceLock, err := options.newResourceLock(leaderConfig, recorderProvider, leaderelection.Options{
//是否启用leader选举
LeaderElection: options.LeaderElection,
// 资源锁 类型 默认为 configmapsleases
LeaderElectionResourceLock: options.LeaderElectionResourceLock,
//决定使用那个资源锁进行选举
LeaderElectionID: options.LeaderElectionID,
// 资源命名空间
LeaderElectionNamespace: options.LeaderElectionNamespace,
})
if err != nil {
return nil, err
}
// 创建MetricsListener
metricsListener, err := options.newMetricsListener(options.MetricsBindAddress)
if err != nil {
return nil, err
}
// 当访问Metrics时调用的handlers
metricsExtraHandlers := make(map[string]http.Handler)
// 同理上
healthProbeListener, err := options.newHealthProbeListener(options.HealthProbeBindAddress)
if err != nil {
return nil, err
}
errChan := make(chan error)
runnables := newRunnables(errChan)
return &controllerManager
}
首先创建GuestbookReconciler ，里面两个参数在上面都已经创建完成，client是创建manager时创建，而Scheme，是定义的全局变量
调用SetupWithManager ，使用了Builder模式，所以让我们深入研究一下吧
// SetupWithManager sets up the controller with the Manager.
func (r *GuestbookReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&webappv1.Guestbook{}).
Complete(r)
}
调用NewControllerManagedBy 创建一个Builder
调用FOR 初始化 Builder
开始Complete进行Controller创建
该方法里我们只需要关注两个方法即可doController，doWatch。从字面意思行我们就能知道一个是控制器，一个是监视器。
func (blder *Builder) Complete(r reconcile.Reconciler) error {
_, err := blder.Build(r)
return err
}
func (blder *Builder) Build(r reconcile.Reconciler) (controller.Controller, error) {
if err := blder.doController(r); err != nil {
return nil, err
}
if err := blder.doWatch(); err != nil {
return nil, err
}
return blder.ctrl, nil
}
创建Controller，首先获取配置(配置信息在下面，这里不再讲解)然后将配置进行初始化操作如果配置没有进行设置的话。
func NewUnmanaged(name string, mgr manager.Manager, options Options) (Controller, error) {
***
return &controller.Controller{
//此处 do的函数就是我们配置的Reconciler
Do: options.Reconciler,
// 监听CRD资源的CUD事件缓冲队列
MakeQueue: func() workqueue.RateLimitingInterface {
return workqueue.NewNamedRateLimitingQueue(options.RateLimiter, name)
},
MaxConcurrentReconciles: options.MaxConcurrentReconciles,
CacheSyncTimeout: options.CacheSyncTimeout,
SetFields: mgr.SetFields,
Name: name,
Log: options.Log.WithName("controller").WithName(name),
RecoverPanic: options.RecoverPanic,
}, nil
}
doWatch添加CRD资源监视以及与Controller相关的监视。
if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
return err
}
// Watch 里面就是使用informer为资源添加监控事件
// allPredicates过滤器
在控制器创建完成与manager启动之前中间还有一些调用，主要是添加就绪handler与健康检查handler 有默认值
启动serveMetrics
启动serveHealthProbes
如果配置了Webhooks就启动
Controller启动，轮询事件队列，读取调用Reconcile
for i := 0; i < c.MaxConcurrentReconciles; i++ {
go func() {
defer wg.Done()
// Run a worker thread that just dequeues items, processes them, and marks them done.
// It enforces that the reconcileHandler is never invoked concurrently with the same object.
for c.processNextWorkItem(ctx) {
}
}()
}
func (c *Controller) processNextWorkItem(ctx context.Context) bool {
obj, shutdown := c.Queue.Get()
if shutdown {
// Stop working
return false
}
// We call Done here so the workqueue knows we have finished
// processing this item. We also must remember to call Forget if we
// do not want this work item being re-queued. For example, we do
// not call Forget if a transient error occurs, instead the item is
// put back on the workqueue and attempted again after a back-off
// period.
defer c.Queue.Done(obj)
ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(1)
defer ctrlmetrics.ActiveWorkers.WithLabelValues(c.Name).Add(-1)
c.reconcileHandler(ctx, obj)
return true
}
watch启动
//
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
}
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()
// 开始循环启动informer
for _, informer := range ip.informersByGVK {
go informer.Informer.Run(ctx.Done())
}
ip.started = true
close(ip.startWait)
}()
<-ctx.Done()
}
// 运行informer
func (s *sharedIndexInformer) Run(stopCh <-chan struct{}) {
defer utilruntime.HandleCrash()
if s.HasStarted() {
klog.Warningf("The sharedIndexInformer has started, run more than once is not allowed")
return
}
// 创建fifo
fifo := NewDeltaFIFOWithOptions(DeltaFIFOOptions{
KnownObjects: s.indexer,
EmitDeltaTypeReplaced: true,
})
// 创建配置
cfg := &Config{
Queue: fifo,
// list，watch
ListerWatcher: s.listerWatcher,
ObjectType: s.objectType,
FullResyncPeriod: s.resyncCheckPeriod,
RetryOnError: false,
ShouldResync: s.processor.shouldResync,
// 接受fifo队列进行处理的方法
Process: s.HandleDeltas,
WatchErrorHandler: s.watchErrorHandler,
}
func() {
s.startedLock.Lock()
defer s.startedLock.Unlock()
s.controller = New(cfg)
s.controller.(*controller).clock = s.clock
s.started = true
}()
// Separate stop channel because Processor should be stopped strictly after controller
processorStopCh := make(chan struct{})
var wg wait.Group
defer wg.Wait() // Wait for Processor to stop
defer close(processorStopCh) // Tell Processor to stop
wg.StartWithChannel(processorStopCh, s.cacheMutationDetector.Run)
wg.StartWithChannel(processorStopCh, s.processor.run)
defer func() {
s.startedLock.Lock()
defer s.startedLock.Unlock()
s.stopped = true // Don't want any new listeners
}()
// 开始运行CacheController
s.controller.Run(stopCh)
}
func (c *controller) Run(stopCh <-chan struct{}) {
// 创建NewReflector
r := NewReflector(
c.config.ListerWatcher,
c.config.ObjectType,
c.config.Queue,
c.config.FullResyncPeriod,
)
// 开始list watch 并将事件添加到fifo中 注意这个是上面创建的那个fifo
wg.StartWithChannel(stopCh, r.Run)
wait.Until(c.processLoop, time.Second, stopCh)
wg.Wait()
}
//循环取出fifo中的值并调用PopProcessFunc
func (c *controller) processLoop() {
for {
obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
}
}
// PopProcessFunc 实际上调用的是shared_informer.HandleDeltas
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
s.blockDeltas.Lock()
defer s.blockDeltas.Unlock()
// from oldest to newest
for _, d := range obj.(Deltas) {
switch d.Type {
case Sync, Replaced, Added, Updated:
s.cacheMutationDetector.AddObject(d.Object)
if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
// 对indexer 的操作
if err := s.indexer.Update(d.Object); err != nil {
return err
}
isSync := false
switch {
case d.Type == Sync:
// Sync events are only propagated to listeners that requested resync
isSync = true
case d.Type == Replaced:
if accessor, err := meta.Accessor(d.Object); err == nil {
if oldAccessor, err := meta.Accessor(old); err == nil {
// Replaced events that didn't change resourceVersion are treated as resync events
// and only propagated to listeners that requested resync
isSync = accessor.GetResourceVersion() == oldAccessor.GetResourceVersion()
}
}
}
// 通知监视者
s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
} else {
if err := s.indexer.Add(d.Object); err != nil {
return err
}
s.processor.distribute(addNotification{newObj: d.Object}, false)
}
case Deleted:
if err := s.indexer.Delete(d.Object); err != nil {
return err
}
s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
}
}
return nil
}
leader选举操作
对于Even重试
ctx = logf.IntoContext(ctx, log)
// 这里就是开始调用自定义Reconcile方法
result, err := c.Reconcile(ctx, req)
switch {
case err != nil:
// 这里开始进行重试，将请求重新添加到队列中，但是由于限速等机制，会放入一个等待队列，当等待时间结束会放进正式队列
c.Queue.AddRateLimited(req)
}
// 重试时间间隔计算方法是 0.05s * 2 ^ exp 其中exp是重试次数
backoff := float64(r.baseDelay.Nanoseconds()) * math.Pow(2，float64(exp))
// 这里就是将重试时间与当前时间相加后与当前请求封装传入等待队列
q.waitingForAddCh <- &waitFor{data: item, readyAt: q.clock.Now().Add(duration)}
配置介绍
Manager启动时的配置文件
ctrl.Options
type Options struct {
//将gvk转runtime.Object 使用 runtime.NewScheme() 可以获取
Scheme *runtime.Scheme
// MapperProvider provides the rest mapper used to map go types to Kubernetes APIs
MapperProvider func(c *rest.Config) (meta.RESTMapper, error)
//资源轮询的最小时间间隔
SyncPeriod *time.Duration
//日志 有默认
Logger logr.Logger
//是否启用leader选举
LeaderElection bool
//决定使用那个资源锁进行选举
LeaderElectionID string
//自定义选举配置
LeaderElectionConfig *rest.Config
//是否自愿变为从节点
LeaderElectionReleaseOnCancel bool
// 租约时间间隔
LeaseDuration *time.Duration
// leader持有锁时间
RenewDeadline *time.Duration
// 其它副本重试(竞争leader)时间间隔
RetryPeriod *time.Duration
// 命名空间限制管理器缓存观察的对象
Namespace string
// 配置Metrics 地址
MetricsBindAddress string
// 配置HealthProbe 地址
HealthProbeBindAddress string
// 就绪探测端点名称，默认为“readyz”
ReadinessEndpointName string
// Liveness 探测端点名称，默认为“healthz”
LivenessEndpointName string
// webhook 服务器服务的端口
Port int
// webhook 服务器绑定到的主机名
Host string
// CertDir 是包含服务器密钥和证书的目录
CertDir string
// WebhookServer 是一个外部配置的 webhook.Server
// 该配置优先于上面几个配置
WebhookServer *webhook.Server
// 创建缓存的函数 有默认值
NewCache cache.NewCacheFunc
// 创建client 的函数
NewClient cluster.NewClientFunc
// 对于一些资源关闭缓存
ClientDisableCacheFor []client.Object
// 指定客户端是否应配置为强制执行
DryRunClient bool
// 关闭前的缓冲时间 有默认值
GracefulShutdownTimeout *time.Duration
// Controller 全局配置
Controller v1alpha1.ControllerConfigurationSpec
{
//等待同步缓存的时间限制。 默认2分钟 可选
CacheSyncTimeout *time.Duration
}
}
创建Controller时配置
// 可以运行的最大并发协调数,默认为1 MaxConcurrentReconciles int // 资源修改触发函数，也是在这个里面进行逻辑编写 Reconciler reconcile.Reconciler // 限制请求的频率 RateLimiter ratelimiter.RateLimiter // 日志 Log logr.Logger // 同步缓存等待时间 默认2分钟 CacheSyncTimeout time.Duration // 是否恢复由异常引起的panic(应该是发生异常后是否继续) RecoverPanic bool
