【kubernetes/k8s源码分析】kube-controller-manager之PVProtection controller源码分析

清疚 2022-01-22 04:27 195阅读 0赞

# #

# **0. 入口NewControllerInitializers函数** #

**    注册 **pv-protection** **controllers\["pv-protection"\] = startPVProtectionController

// NewControllerInitializers is a public map of named controller groups (you can start more than one in an init func)
    // paired to their InitFunc.  This allows for structured downstream composition and subdivision.
    func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
    	controllers := map[string]InitFunc{}
    
    	controllers["persistentvolume-binder"] = startPersistentVolumeBinderController
    	controllers["attachdetach"] = startAttachDetachController
    	controllers["persistentvolume-expander"] = startVolumeExpandController
    	controllers["clusterrole-aggregation"] = startClusterRoleAggregrationController
    	controllers["pvc-protection"] = startPVCProtectionController
    	controllers["pv-protection"] = startPVProtectionController
    
    	return controllers
    }

Controller 结构，比较简单，关心的是 pv，队列

>     // Controller is controller that removes PVProtectionFinalizer
>     // from PVs that are not bound to PVCs.
>     type Controller struct {
>        client clientset.Interface
>     
>        pvLister       corelisters.PersistentVolumeLister
>        pvListerSynced cache.InformerSynced
>     
>        queue workqueue.RateLimitingInterface
>     
>        // allows overriding of StorageObjectInUseProtection feature Enabled/Disabled for testing
>        storageObjectInUseProtectionEnabled bool
>     }

# 1. NewPVProtectionController 函数 #

实例化 Controller，添加 pv informer机制，主要就是一个函数 pvAddedUpdated

// NewPVProtectionController returns a new *Controller.
    func NewPVProtectionController(pvInformer coreinformers.PersistentVolumeInformer, cl clientset.Interface, storageObjectInUseProtectionFeatureEnabled bool) *Controller {
    	e := &Controller{
    		client:                              cl,
    		queue:                               workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pvprotection"),
    		storageObjectInUseProtectionEnabled: storageObjectInUseProtectionFeatureEnabled,
    	}
    	if cl != nil && cl.CoreV1().RESTClient().GetRateLimiter() != nil {
    		metrics.RegisterMetricAndTrackRateLimiterUsage("persistentvolume_protection_controller", cl.CoreV1().RESTClient().GetRateLimiter())
    	}
    
    	e.pvLister = pvInformer.Lister()
    	e.pvListerSynced = pvInformer.Informer().HasSynced
    	pvInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
    		AddFunc: e.pvAddedUpdated,
    		UpdateFunc: func(old, new interface{}) {
    			e.pvAddedUpdated(new)
    		},
    	})
    
    	return e
    }

###     1.1 pvAddedUpdated ###

只有这两种情况加入队列去处理，比较简单

a. 已经被删除，而且finalizers 包含 kubernetes.io/pv-protection

>   finalizers:  
>   - kubernetes.io/pv-protection

b. 没有删除的， finalizers 不包含 kubernetes.io/pv-protection

// pvAddedUpdated reacts to pv added/updated events
    func (c *Controller) pvAddedUpdated(obj interface{}) {
    	pv, ok := obj.(*v1.PersistentVolume)
    	if !ok {
    		utilruntime.HandleError(fmt.Errorf("PV informer returned non-PV object: %#v", obj))
    		return
    	}
    	klog.V(4).Infof("Got event on PV %s", pv.Name)
    
    	if needToAddFinalizer(pv) || isDeletionCandidate(pv) {
    		c.queue.Add(pv.Name)
    	}
    }
    
    func isDeletionCandidate(pv *v1.PersistentVolume) bool {
    	return pv.ObjectMeta.DeletionTimestamp != nil && slice.ContainsString(pv.ObjectMeta.Finalizers, volumeutil.PVProtectionFinalizer, nil)
    }
    
    func needToAddFinalizer(pv *v1.PersistentVolume) bool {
    	return pv.ObjectMeta.DeletionTimestamp == nil && !slice.ContainsString(pv.ObjectMeta.Finalizers, volumeutil.PVProtectionFinalizer, nil)
    }

Run

-->  runWorker

-->  processPV

-->  removeFinalizer

-->  addFinalizer

# 2. Run 函数 #

所有一个模式克隆出来的，同步cache，工作线程处理，直奔主题 runWorker

// Run runs the controller goroutines.
    func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
    	defer utilruntime.HandleCrash()
    	defer c.queue.ShutDown()
    
    	klog.Infof("Starting PV protection controller")
    	defer klog.Infof("Shutting down PV protection controller")
    
    	if !controller.WaitForCacheSync("PV protection", stopCh, c.pvListerSynced) {
    		return
    	}
    
    	for i := 0; i < workers; i++ {
    		go wait.Until(c.runWorker, time.Second, stopCh)
    	}
    
    	<-stopCh
    }

###     2.1 processPV 函数 ###

已经被删除，而且finalizers 包含 kubernetes.io/pv-protection 这种情况，没有被使用的pv，调用 removeFinalizer，判断 pv有没有被使用只看 status.Phase是否为 Bound

没有删除的， finalizers 不包含 kubernetes.io/pv-protection 调用 addFinalizer

func (c *Controller) processPV(pvName string) error {
    	pv, err := c.pvLister.Get(pvName)
    
    	if isDeletionCandidate(pv) {
    		// PV should be deleted. Check if it's used and remove finalizer if
    		// it's not.
    		isUsed := c.isBeingUsed(pv)
    		if !isUsed {
    			return c.removeFinalizer(pv)
    		}
    	}
    
    	if needToAddFinalizer(pv) {
    		// PV is not being deleted -> it should have the finalizer. The
    		// finalizer should be added by admission plugin, this is just to add
    		// the finalizer to old PVs that were created before the admission
    		// plugin was enabled.
    		return c.addFinalizer(pv)
    	}
    	return nil
    }

# 3. removeFinalizer #

简单粗暴，直接删除 kubernetes.io/pv-protection，调用 client-go 更新 pv

func (c *Controller) removeFinalizer(pv *v1.PersistentVolume) error {
    	pvClone := pv.DeepCopy()
    	pvClone.ObjectMeta.Finalizers = slice.RemoveString(pvClone.ObjectMeta.Finalizers, volumeutil.PVProtectionFinalizer, nil)
    	_, err := c.client.CoreV1().PersistentVolumes().Update(pvClone)
    	if err != nil {
    		klog.V(3).Infof("Error removing protection finalizer from PV %s: %v", pv.Name, err)
    		return err
    	}
    	klog.V(3).Infof("Removed protection finalizer from PV %s", pv.Name)
    	return nil
    }

# 4. addFinalizer #

添加 kubernetes.io/pv-protection 并更新 pv

func (c *Controller) addFinalizer(pv *v1.PersistentVolume) error {
    	// Skip adding Finalizer in case the StorageObjectInUseProtection feature is not enabled
    	if !c.storageObjectInUseProtectionEnabled {
    		return nil
    	}
    	pvClone := pv.DeepCopy()
    	pvClone.ObjectMeta.Finalizers = append(pvClone.ObjectMeta.Finalizers, volumeutil.PVProtectionFinalizer)
    	_, err := c.client.CoreV1().PersistentVolumes().Update(pvClone)
    	if err != nil {
    		klog.V(3).Infof("Error adding protection finalizer to PV %s: %v", pv.Name, err)
    		return err
    	}
    	klog.V(3).Infof("Added protection finalizer to PV %s", pv.Name)
    	return nil
    }

### 总结： ###

watch pv

已经被删除，而且finalizers 包含 kubernetes.io/pv-protection 这种情况，而且没有被bound的，删除 kubernetes.io/pv-protection

没有删除的， finalizers 不包含 kubernetes.io/pv-protection 调用 addFinalizer 添加 kubernetes.io/pv-protection