pv-pvc | 南风未起｜云原生/Go技术分享博客

# 1.简介

# 1.1.功能

pv controller是kcm组件之一，负责监听PV/PVC/Storageclass资源，驱动卷创建回收及同步存储状态，实现PV动态加载及PV/PVC绑定。
注意

pv controller和in-tree配合管理in-tree volume，out-tree volume由external-provisioner和out-tree管理

# 1.2.状态

pvc会经历pending-->bound状态，pv会经历pending-->available-->bound-->released-->failed状态，转换由绑定或回收状态决定。
注意

PV/PVC状态根据两者绑定关系及删除状态转换，以影响PV Controller执行策略

# 2.入口

# 2.1.启动

startPersistentVolumeBinderController()负责加载plugin及实例化PV Controller，调用pv.Run()激活worker处理任务。

// NewController creates a new PersistentVolume controller
func NewController(ctx context.Context, p ControllerParameters) (*PersistentVolumeController, error) {
	...

    // pv controller实例化
	controller := &PersistentVolumeController{
		volumes:                       newPersistentVolumeOrderedIndex(),		// PV索引缓存
		claims:                        cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),	// PVC缓存
		...
		runningOperations:             goroutinemap.NewGoRoutineMap(true /* exponentialBackOffOnError */),
		...
		enableDynamicProvisioning:     p.EnableDynamicProvisioning,
		clusterName:                   p.ClusterName,
		createProvisionedPVRetryCount: 5,										// PV创建重试次数
		createProvisionedPVInterval:   10s,		// PV创建间隔
		claimQueue:                    workqueue.NewNamed("claims"),			// PVC队列
		volumeQueue:                   workqueue.NewNamed("volumes"),			// PVC队列
		resyncPeriod:                  5s,										// 重试周期
		...
	}

    // 初始化plugin prober(这里不感知flex volume)
	controller.volumePluginMgr.InitPlugins(p.VolumePlugins, nil /* prober */, controller)
    ...

    // PV Informer回调
	p.VolumeInformer.Informer().AddEventHandler(
		cache.ResourceEventHandlerFuncs{
			AddFunc:    func(obj interface{}) { controller.enqueueWork(ctx, controller.volumeQueue, obj) },
			UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueWork(ctx, controller.volumeQueue, newObj) },
			DeleteFunc: func(obj interface{}) { controller.enqueueWork(ctx, controller.volumeQueue, obj) },
		},
	)
	controller.volumeLister = p.VolumeInformer.Lister()
	...

    // PVCInformer回调
	p.ClaimInformer.Informer().AddEventHandler(
		cache.ResourceEventHandlerFuncs{
			AddFunc:    func(obj interface{}) { controller.enqueueWork(ctx, controller.claimQueue, obj) },
			UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueWork(ctx, controller.claimQueue, newObj) },
			DeleteFunc: func(obj interface{}) { controller.enqueueWork(ctx, controller.claimQueue, obj) },
		},
	)
	controller.claimLister = p.ClaimInformer.Lister()
	...

    // StorageClass/Pod/Node Informer缓存
	controller.classLister = p.ClassInformer.Lister()
	...
	controller.podLister = p.PodInformer.Lister()
	controller.podIndexer = p.PodInformer.Informer().GetIndexer()
	...
	controller.NodeLister = p.NodeInformer.Lister()
	...

	// pod<-->pvc name索引
	if err := common.AddPodPVCIndexerIfNotPresent(controller.podIndexer); err != nil {
		return nil, fmt.Errorf("could not initialize attach detach controller: %w", err)
	}

    // in-tree-->csi转换器
	csiTranslator := csitrans.New()
	controller.translator = csiTranslator
    
    // 维护迁移到csi的plugin
	controller.csiMigratedPluginManager = csimigration.NewPluginManager(csiTranslator, DefaultFeatureGate)

    // 网段访问/Loopback限制
	controller.filteredDialOptions = p.FilteredDialOptions

	return controller, nil
}

func startPersistentVolumeBinderController(...) (controller.Interface, bool, error) {
    // plugin加载(in-tree/csi)
	plugins, err := ProbeControllerVolumePlugins(...)
	...
    
	params := persistentvolumecontroller.ControllerParameters{
		...
		SyncPeriod:            	   30s,
		VolumePlugins:             plugins,
		...
		VolumeInformer:            controllerContext.InformerFactory.Core().V1().PersistentVolumes(),
		ClaimInformer:             controllerContext.InformerFactory.Core().V1().PersistentVolumeClaims(),
		ClassInformer:             controllerContext.InformerFactory.Storage().V1().StorageClasses(),
		PodInformer:               controllerContext.InformerFactory.Core().V1().Pods(),
		NodeInformer:              controllerContext.InformerFactory.Core().V1().Nodes(),
		EnableDynamicProvisioning: volumeConfiguration.EnableDynamicProvisioning,
		FilteredDialOptions:       filteredDialOptions, 	// 拒绝的网段/允许loopback
	}
    
    
	volumeController, volumeControllerErr := persistentvolumecontroller.NewController(ctx, params)
	...
    
	go volumeController.Run(ctx)
	return nil, true, nil
}

注意

csiTranslator将迁移到csi的in-tree插件进行转换注册到volumeMgr，可用plugin加载由ProbeControllerVolumePlugins构造

# 2.2.运行

pc.Run()阻塞至缓存同步完成，激活resync worker、volume worker和claim worker协程进行资源同步及PV/PVC周期处理。

// fills all controller caches with initial data from etcd.
func (ctrl *PersistentVolumeController) initializeCaches(...) {
    // 获取PV缓存
	volumeList, err := volumeLister.List(labels.Everything())
	...
    
    // 向volumes补充或更新PV缓存
	for _, volume := range volumeList {
		ctrl.storeVolumeUpdate(logger, volume.DeepCopy())
        ...
	}

    // 获取PVC缓存
	claimList, err := claimLister.List(labels.Everything())
	...
    
     // 向claims补充或更新PV缓存
	for _, claim := range claimList {
		ctrl.storeClaimUpdate(logger, claim.DeepCopy()
        ...
	}
}
                              
// updates given cache with a new object version from Informer callback.
func storeObjectUpdate(...) (bool, error) {
	objName, err := controller.KeyFunc(obj)
	...
    
	oldObj, found, err := store.Get(obj)
	...

	objAccessor, err := meta.Accessor(obj)
	...
    
    // controller缓存没有
	if !found {
		// 补充
		store.Add(obj)
        ...
		return true, nil
	}

	oldObjAccessor, err := meta.Accessor(oldObj)
	...

	objResourceVersion, err := strconv.ParseInt(objAccessor.GetResourceVersion(), 10, 64)
	...
	oldObjResourceVersion, err := strconv.ParseInt(oldObjAccessor.GetResourceVersion(), 10, 64)
	...

	// oldRV更新
	if oldObjResourceVersion > objResourceVersion {
		return false, nil
	}
	...
    
    // 更新PV缓存
	store.Update(obj)
    ...
    
	return true, nil
}

// Run starts all of this controller's control loops
func (ctrl *PersistentVolumeController) Run(ctx context.Context) {
	...

	if !cache.WaitForNamedCacheSync("pv", ctx.Done(), pvSynced, pvcSynced, classSynced, podSynced, nodeSynced) {
		return
	}

    // 基于informer初始化pv controller缓存
	ctrl.initializeCaches(logger, ctrl.volumeLister, ctrl.claimLister)

    // 激活worker
	go wait.Until(func() { ctrl.resync(ctx) }, ctrl.resyncPeriod, ctx.Done())
	go wait.UntilWithContext(ctx, ctrl.volumeWorker, time.Second)
	go wait.UntilWithContext(ctx, ctrl.claimWorker, time.Second)
	...

	<-ctx.Done()
}

注意

initializeCaches本质上是将PV/PVC informer缓存转到persistentVolumeOrderedIndex，基于accessMode索引及容量排序

# 2.3.同步

pc.resync()周期循环(5s)查询PV/PVC列表，将PV/PVC放入volumeQueue/claimQueue队列，供volumeworker/cliamworker消费。

// resync supplements short resync period of shared informers - we don't want
// all consumers of PV/PVC shared informer to have a short resync period,
// therefore we do our own.
func (ctrl *PersistentVolumeController) resync(ctx context.Context) {
	pvcs, err := ctrl.claimLister.List(labels.NewSelector())
	...
    
    // pvc入队
	for _, pvc := range pvcs {
		ctrl.enqueueWork(ctx, ctrl.claimQueue, pvc)
	}

	pvs, err := ctrl.volumeLister.List(labels.NewSelector())
	...
    
    // PV入队
	for _, pv := range pvs {
		ctrl.enqueueWork(ctx, ctrl.volumeQueue, pv)
	}
}

// enqueueWork adds volume or claim to given work queue.
func (ctrl *PersistentVolumeController) enqueueWork(...) {
	// Beware of "xxx deleted" events
	if unknown, ok := obj.(cache.DeletedFinalStateUnknown); ok && unknown.Obj != nil {
		obj = unknown.Obj
	}
    
	objName, err := controller.KeyFunc(obj)
	...
    
	queue.Add(objName)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

补充

volumeQueue和claimQueue队列数据供volumeworker和claimworker消费

# 3.PV处理

# 3.1.volumeWorker

pc.volumeWorker()不断获取volumeQueue队列数据执行updateVolume，无法找到匹配的PV会调用plugin清理存储及删除PV对象。

// deleteVolume runs in worker thread and handles "volume deleted" event.
func (ctrl *PersistentVolumeController) deleteVolume(ctx context.Context, volume *v1.PersistentVolume) {
	// 由缓存清理
	ctrl.volumes.store.Delete(volume)
    ...

	if volume.Spec.ClaimRef == nil {
		return
	}
	...
    
    // 引用的PVC入队
	ctrl.claimQueue.Add(claimKey)
}

// processes items from volumeQueue. It must run only once, syncVolume is not assured to be reentrant.
func (ctrl *PersistentVolumeController) volumeWorker(ctx context.Context) {
	// 处理函数
	workFunc := func(ctx context.Context) bool {
		keyObj, quit := ctrl.volumeQueue.Get()
		...
        
		defer ctrl.volumeQueue.Done(keyObj)
		key := keyObj.(string)
		
		// 获取PV名称
		_, name, err := cache.SplitMetaNamespaceKey(key)
		...
        
        // 获取PV对象
		volume, err := ctrl.volumeLister.Get(name)
		if err == nil {
			// 执行同步
			ctrl.updateVolume(ctx, volume)
			return false
		}
		...

		// PV已清理,获取volumes缓存
		volumeObj, found, err := ctrl.volumes.store.GetByKey(key)
		...
        
		// 转为PV对象
		volume, ok := volumeObj.(*v1.PersistentVolume)
		...
        
        // 回收存储及删除PV对象
		ctrl.deleteVolume(ctx, volume)
		return false
	}
    
	for {
		if quit := workFunc(ctx); quit {
			return
		}
	}
}

// runs in worker thread and handles "volume added", "volume updated" and "periodic sync" events.
func (ctrl *PersistentVolumeController) updateVolume(ctx context.Context, volume *v1.PersistentVolume) {
	...
    
    // 更新volumes缓存
	new, err := ctrl.storeVolumeUpdate(logger, volume)
	...
    
    // 无变化或处理错误
	if !new {
		return
	}

    // PV同步
	ctrl.syncVolume(ctx, volume)
	...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

注意

volumes缓存会清理未关联存储资源的PV，因此informer缓存找不到PV会检查volumes缓存决定走不走存储回收

# 3.2.updateVolume

updateMigrationAnnotations()基于CSI迁移情况及PV回收策略，修正migration Annotation和deletion finalizer。

// takes an Annotations map and checks for a provisioner name using the provisionerKey.
func updateMigrationAnnotations(...) bool {
	...

    // annotation为空
	if ann == nil {
		return false
	}
	...
    
    // PV提供者
    provisionerKey = storagehelpers.AnnDynamicallyProvisioned
    
    // 获取提供者
	provisioner, ok := ann[provisionerKey]
	if !ok {
        // Volume Statically provisioned.
        return false
	}

    // 基于迁移注解获取driver名称
	migratedToDriver := ann[storagehelpers.AnnMigratedTo]
    // plugin迁移至csi
	if cmpm.IsMigrationEnabledForPlugin(provisioner) {
        // 基于提供者获取driver名称
		csiDriverName, err = translator.GetCSINameFromInTreeName(provisioner)
		...
        
        // csi迁移注解纠正
		if migratedToDriver != csiDriverName {
			ann[storagehelpers.AnnMigratedTo] = csiDriverName
			return true
		}
    // 未迁移至csi
	} else {
		if migratedToDriver != "" {
			// 清理迁移注解
			delete(ann, storagehelpers.AnnMigratedTo)
			return true
		}
	}
	return false
}

// updates the finalizers based on the reclaim policy and if it is a in-tree volume or not.
func modifyDeletionFinalizers(...) ([]string, bool) {
	...
    
    // 未启用PV删除保护
	if !utilfeature.DefaultFeatureGate.Enabled(features.HonorPVReclaimPolicy) {
		return volume.Finalizers, false
	}
    
    // 静态供应
	if !metav1.HasAnnotation(volume.ObjectMeta, storagehelpers.AnnDynamicallyProvisioned) {
		return volume.Finalizers, false
	}
    
	if volume.Finalizers != nil {
		outFinalizers = append(outFinalizers, volume.Finalizers...)
	}
    
    // 获取提供者
	provisioner := volume.Annotations[storagehelpers.AnnDynamicallyProvisioned]
    
    // Plugin迁移至CSI
	if cmpm.IsMigrationEnabledForPlugin(provisioner) {
		// 移除in-tree删除保护finalizer,由CSI Driver管理删除
		if slice.ContainsString(outFinalizers, storagehelpers.PVDeletionInTreeProtectionFinalizer, nil) {
			outFinalizers = slice.RemoveString(outFinalizers, storagehelpers.PVDeletionInTreeProtectionFinalizer, nil)
			modified = true
		}
		return outFinalizers, modified
	}
    
	// 非in-tree plugin
	if !strings.HasPrefix(provisioner, "kubernetes.io/") {
		return volume.Finalizers, false
	}
    ...
    
    // 删除回收+无删除保护
	if reclaimPolicy == v1.PersistentVolumeReclaimDelete && !slice.ContainsString(outFinalizers, storagehelpers.PVDeletionInTreeProtectionFinalizer, nil) {
		// 补充删除保护finalizer
		outFinalizers = append(outFinalizers, storagehelpers.PVDeletionInTreeProtectionFinalizer)
		modified = true
    // 保留或复用+有删除保护finalizer
	} else if (reclaimPolicy == v1.PersistentVolumeReclaimRetain || reclaimPolicy == v1.PersistentVolumeReclaimRecycle) && slice.ContainsString(outFinalizers, storagehelpers.PVDeletionInTreeProtectionFinalizer, nil) {
        // 移除删除保护finalizer
		outFinalizers = slice.RemoveString(outFinalizers, storagehelpers.PVDeletionInTreeProtectionFinalizer, nil)
		modified = true
	}
    
	// 移除external-provisioner添加的删除保护finalizer
	if slice.ContainsString(outFinalizers, storagehelpers.PVDeletionProtectionFinalizer, nil) {
		outFinalizers = slice.RemoveString(outFinalizers, storagehelpers.PVDeletionProtectionFinalizer, nil)
		modified = true
	}
	return outFinalizers, modified
}

func (ctrl *PersistentVolumeController) updateVolumeMigrationAnnotationsAndFinalizers(...) (...) {
	volumeClone := volume.DeepCopy()
    // CSI迁移注解修正
	annModified := updateMigrationAnnotations(logger, ctrl.csiMigratedPluginManager, ctrl.translator, volumeClone.Annotations, false)
    
    // PV删除保护finalizer修正
	modifiedFinalizers, finalizersModified := modifyDeletionFinalizers(logger, ctrl.csiMigratedPluginManager, volumeClone)
    
    // 无更改
	if !annModified && !finalizersModified {
		return volumeClone, nil
	}
    
    // 更新finalizer
	if finalizersModified {
		volumeClone.ObjectMeta.SetFinalizers(modifiedFinalizers)
	}
    
    // patch到etcd
	newVol, err := ctrl.kubeClient.CoreV1().PersistentVolumes().Update(ctx, volumeClone, metav1.UpdateOptions{})
	...
    
    // 更新volumes缓存
	_, err = ctrl.storeVolumeUpdate(logger, newVol)
	...
    
	return newVol, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

注意

updateVolumeMigrationAnnotationsAndFinalizers基于PV及CSI供应者更新迁移annotation及deletion finalizer

# 3.3.syncVolume

// syncVolume is the main controller method to decide what to do with a volume.
func (ctrl *PersistentVolumeController) syncVolume(ctx context.Context, volume *v1.PersistentVolume) error {
	// 更新迁移annotation和deletion finalizer
	newVolume, err := ctrl.updateVolumeMigrationAnnotationsAndFinalizers(ctx, volume)
	...
	volume = newVolume

	// PV未使用
	if volume.Spec.ClaimRef == nil {
		// 更新phase为可用,同步到volumes缓存
		ctrl.updateVolumePhase(ctx, volume, v1.VolumeAvailable, "")
        ...
        
		return nil
    // 已开始使用
	} else /* pv.Spec.ClaimRef != nil */ {
		// PVC未绑定PV
		if volume.Spec.ClaimRef.UID == "" {
			// 更新phase为可用,同步到volumes缓存
			ctrl.updateVolumePhase(ctx, volume, v1.VolumeAvailable, "")
            ...
            
			return nil
		}
        ...
        
		// 基于claimRef获取PVC
		obj, found, err := ctrl.claims.GetByKey(claimName)
		...
        
        // PVC没找到
		if !found {
			// released/failed状态的PV略过
			if volume.Status.Phase != v1.VolumeReleased && volume.Status.Phase != v1.VolumeFailed {
                // 由informer获取PVC
				obj, err = ctrl.claimLister.PersistentVolumeClaims(claimRef.Namespace).Get(claimRef.Name)
				...
                
                // 未找到
				if !found {
                    // 由底层获取
					obj, err = ctrl.kubeClient.PVClaims(claimRef.Namespace).Get(ctx, claimRef.Name, ...)
					...
                    
					found = !apierrors.IsNotFound(err)
				}
			}
		}
        
        // 匹配到PVC
		if found {
			...
			claim, ok = obj.(*v1.PersistentVolumeClaim)
			...
		}
        
        // PVC和PV引用的UID不一致
		if claim != nil && claim.UID != volume.Spec.ClaimRef.UID {
			// 重新获取一次
			claim, err = ctrl.kubeClient.PVClaims(claimRef.Namespace).Get(ctx, claimRef.Name, ...)
            // 还是不匹配,PV引用PVC缺失
			if claim != nil && claim.UID != volume.Spec.ClaimRef.UID {
                claim = nil
			}
		}

        // PVC缺失
		if claim == nil {
			// 还没走到released/failed phase
			if volume.Status.Phase != v1.VolumeReleased && volume.Status.Phase != v1.VolumeFailed {
				// 更新phase为released,同步到volumes缓存
				ctrl.updateVolumePhase(ctx, volume, v1.VolumeReleased, "")
                ...
			}
            
            // 基于回收策略进行相应处理
			ctrl.reclaimVolume(ctx, volume)
            ...
            
			return nil
        // PVC引用的PV为空
		} else if claim.Spec.VolumeName == "" {
            // volumeMode未匹配
			if storagehelpers.CheckVolumeModeMismatches(&claim.Spec, &volume.Spec) {
				return nil
			}

			// PVC重新推到claimQueue处理,由claimworker进行绑定
			ctrl.claimQueue.Add(claimToClaimKey(claim))
			return nil
        // PVC绑定PV
		} else if claim.Spec.VolumeName == volume.Name {
			// 更新phase为bound
			ctrl.updateVolumePhase(ctx, volume, v1.VolumeBound, "")
            ...
            
			return nil
        // PVC绑定到其它PV
		} else {
            // PV动态供应+删除回收策略
			if metav1.HasAnnotation(volume.ObjectMeta, storagehelpers.AnnDynamicallyProvisioned) && volume.Spec.PersistentVolumeReclaimPolicy == v1.PersistentVolumeReclaimDelete {
				// PV未走到released/failed
				if volume.Status.Phase != v1.VolumeReleased && volume.Status.Phase != v1.VolumeFailed {
					// 更新PV Phase为released
					ctrl.updateVolumePhase(ctx, volume, v1.VolumeReleased, "")
                    ...
				}
                
                // 基于回收策略进行相关处理
				ctrl.reclaimVolume(ctx, volume)
                ...
                
				return nil
            // 静态供应+保留/复用策略
			} else {
				// 由controller/user绑定,摘除bound annotation,更新phase为可用,更新volumes缓存
                ctrl.unbindVolume(ctx, volume)
                ...

                return nil
			}
		}
	}
}

注意

syncVolume会更新PV状态及volumes回收，基于关联PVC和PV关系进行策略回收

# 3.4.reclaimVolume

pc.reclaimVolume()基于PV设置的回收策略进行相关处理，支持的策略有retain、delete及recycle，对应是否级联PVC删除。

// reclaimVolume implements volume.Spec.PersistentVolumeReclaimPolicy and starts appropriate reclaim action.
func (ctrl *PersistentVolumeController) reclaimVolume(ctx context.Context, volume *v1.PersistentVolume) error {
	// 供应者已迁移至CSI,由CSI驱动处理
	if migrated := volume.Annotations[storagehelpers.AnnMigratedTo]; len(migrated) > 0 {
		return nil
	}
	
    switch volume.Spec.PersistentVolumeReclaimPolicy {
    // 保留现状
	case v1.PersistentVolumeReclaimRetain:
	
    // 回收复用
	case v1.PersistentVolumeReclaimRecycle:
        // 执行回收(涉及重试限制)
		ctrl.scheduleOperation(logger, opName, func() error {
			ctrl.recycleVolumeOperation(ctx, volume)
			return nil
		})
	// 级联删除
	case v1.PersistentVolumeReclaimDelete:
		...
        // 执行删除(涉及重试限制)
		ctrl.scheduleOperation(logger, opName, func() error {
			_, err := ctrl.deleteVolumeOperation(ctx, volume)
			...
            
			return err
		})

	default:
		// 未知策略,更新phase为failed
		ctrl.updateVolumePhaseWithEvent(ctx, volume, v1.VolumeFailed, v1.EventTypeWarning, ...)
	}
	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

注意

recycle仅清理内容复用存储资源，delete会删除底层的存储资源

# 3.5.recycleVolume

pc.recycleVolumeOperation()检查PV回收条件，匹配volumeMgr支持volume回收的plugin及执行plugin.Recycle回收存储资源。

// recycles a volume. This method is running in standalone goroutine and already has all necessary locks.
func (ctrl *PersistentVolumeController) recycleVolumeOperation(...) {
	// 获取底层PV
	newVolume, err := ctrl.kubeClient.CoreV1().PersistentVolumes().Get(ctx, volume.Name, metav1.GetOptions{})
	...
    
    // 回收检查
	needsReclaim, err := ctrl.isVolumeReleased(logger, newVolume)
	...
    
    // 未引用PVC/匹配PVC UID和Name
	if !needsReclaim {
		return
	}
    
    // 检查PV使用情况
	pods, used, err := ctrl.isVolumeUsed(newVolume)
	...

	_, claimCached, err := ctrl.claims.GetByKey(claimName)
	...

    // PV正在使用&claims缓存未查到PVC
	if used && !claimCached {
		return
	}

    // PV未使用或PVC绑定其它PV,可以安全回收
	volume = newVolume

	// 构造volume plugin查询条件
	spec := vol.NewSpecFromPersistentVolume(volume, false)
    // 查找支持volume spec的recycle plugin(hostpath/nfs),同时触发一次volumeMgr再加载
	plugin, err := ctrl.volumePluginMgr.FindRecyclablePluginBySpec(spec)
	if err != nil {
		// PV没有支持回收插件,更新phase为failed
		ctrl.updateVolumePhaseWithEvent(ctx, volume, v1.VolumeFailed, ...)
        ...
        
		return
	}
	...

    // 调用csi plugin执行回收
	if err = plugin.Recycle(volume.Name, spec, recorder); err != nil {
		// 回收失败,PV更新phase为failed
        ctrl.updateVolumePhaseWithEvent(ctx, volume, v1.VolumeFailed, ...)
        ...
        
		return
	}
	...
    
	// 回收完成,解绑PVC,更新volumes缓存及更新PV为可用
	ctrl.unbindVolume(ctx, volume)
    ...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

注意

in-tree plugin仅hostpath和nfs实现recycle，支持清理volume存储数据实现资源复用

# 3.6.deleteVolume

pc.deleteVolumeOperation()检查PV和PVC关联状态，匹配支持volume delete的插件执行plugin.Deleter().Delete()删除存储。

// finds a deleter plugin for a given volume.
func (ctrl *PersistentVolumeController) findDeletablePlugin(volume *v1.PersistentVolume) (...) {
	...
    
    // PV动态供应
	if metav1.HasAnnotation(volume.ObjectMeta, storagehelpers.AnnDynamicallyProvisioned) {
        // 获取提供者
		provisionPluginName := volume.Annotations[storagehelpers.AnnDynamicallyProvisioned]
		if provisionPluginName != "" {
            // 基于提供者查询实现删除接口的plugin
			plugin, err := ctrl.volumePluginMgr.FindDeletablePluginByName(provisionPluginName)
			...
            
			return plugin, nil
		}
	}

	// in-tree plugin,基于volume定义查询plugin
	spec := vol.NewSpecFromPersistentVolume(volume, false)
	plugin, err := ctrl.volumePluginMgr.FindDeletablePluginBySpec(spec)
	...
    
	return plugin, nil
}

// finds appropriate delete plugin and deletes given volume, returning the volume plugin name.
func (ctrl *PersistentVolumeController) doDeleteVolume(...) (string, bool, error) {
	...

    // 查询支持volume删除的插件
	plugin, err := ctrl.findDeletablePlugin(volume)
	...
    
    // 三方实现处理
	if plugin == nil {
		return "", false, nil
	}

	// Plugin found
	pluginName := plugin.GetPluginName()
	spec := vol.NewSpecFromPersistentVolume(volume, false)
    // 构造plugin deleter
	deleter, err := plugin.NewDeleter(logger, spec)
	...

    // 执行删除
	deleter.Delete()
	...
    
	// 启用回收策略
	if utilfeature.DefaultFeatureGate.Enabled(features.HonorPVReclaimPolicy) {
        // 移除删除保护finalizer
		ctrl.removeDeletionProtectionFinalizer(ctx, volume)
        ...
	}
    
	return pluginName, true, nil
}

// deletes a volume. This method is running in standalone goroutine and already has all necessary locks.
func (ctrl *PersistentVolumeController) deleteVolumeOperation(...) (string, error) {
	// 获取底层volume
	newVolume, err := ctrl.kubeClient.CoreV1().PersistentVolumes().Get(ctx, volume.Name, metav1.GetOptions{})
	...

    // 未启用回收策略+PV正在删除
	if !DefaultFeatureGate.Enabled(features.HonorPVReclaimPolicy) &&newVolume.GetDeletionTimestamp() != nil {
        return "", nil
	}
    
    // 回收检查
	needsReclaim, err := ctrl.isVolumeReleased(logger, newVolume)
	...
    
    // 未引用PVC/匹配PVC UID和Name
	if !needsReclaim {
		return "", nil
	}

    // 调用plugin执行清理
	pluginName, deleted, err := ctrl.doDeleteVolume(ctx, volume)
	if err != nil {
		// 非清理正在使用PV报错
		if !volerr.IsDeletedVolumeInUse(err) {
			// 更新phase为failed
			ctrl.updateVolumePhaseWithEvent(ctx, volume, v1.VolumeFailed, ...)
            ...
		}

		return pluginName, err
	}
    
    // 还未执行删除
	if !deleted {
		return pluginName, nil
	}

	// 删除PV对象
	ctrl.kubeClient.CoreV1().PersistentVolumes().Delete(ctx, volume.Name, metav1.DeleteOptions{})
    ...
    
	return pluginName, nil
}

注意

deleteVolumeOperation会检查删除条件，引用未匹配PVC的PV会匹配delete plugin进行存储资源删除

# 4.PVC处理

# 4.1.claimWorker

pc.claimWorker()是一个无限循环，不断获取claimQueue队列的PVC，调用pc.updateClaim方法执行syncClaim完成PVC协调任务。

// runs in worker thread and handles "claim deleted" event.
func (ctrl *PersistentVolumeController) deleteClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) {
	// 清理claims缓存
	ctrl.claims.Delete(claim)
    ...
    
    // 关联的PV推到volumeQueue队列
	volumeName := claim.Spec.VolumeName
	if volumeName == "" {
		return
	}
	
	ctrl.volumeQueue.Add(volumeName)
}

// processes items from claimQueue. It must run only once, syncClaim is not reentrant.
func (ctrl *PersistentVolumeController) claimWorker(ctx context.Context) {

	workFunc := func() bool {
        // 获取claimQueue数据
		keyObj, quit := ctrl.claimQueue.Get()
		...
		defer ctrl.claimQueue.Done(keyObj)
		...

        // 获取informer缓存的PVC
		namespace, name, err := cache.SplitMetaNamespaceKey(key)
		...
		claim, err := ctrl.claimLister.PersistentVolumeClaims(namespace).Get(name)
		if err == nil {
			// 执行同步
			ctrl.updateClaim(ctx, claim)
			return false
		}
		...

		// 获取claims缓存PVC
		claimObj, found, err := ctrl.claims.GetByKey(key)
		...
        
        // 未找到
		if !found {
			return false
		}
        
        // 执行清理
		claim, ok := claimObj.(*v1.PersistentVolumeClaim)
		...
		ctrl.deleteClaim(ctx, claim)
		return false
	}
	for {
		if quit := workFunc(); quit {
			return
		}
	}
}

// runs in worker thread and handles "claim added", "claim updated" and "periodic sync" events.
func (ctrl *PersistentVolumeController) updateClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) {
	// 更新claims缓存
	new, err := ctrl.storeClaimUpdate(logger, claim)
	...
    
    // 无变化
	if !new {
		return
	}
    
    // 执行同步
	ctrl.syncClaim(ctx, claim)
	...
}

// syncClaim is the main controller method to decide what to do with a claim.
func (ctrl *PersistentVolumeController) syncClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) error {
	// 迁移Annotation纠正,更新PVC及claims缓存
	newClaim, err := ctrl.updateClaimMigrationAnnotations(ctx, claim)
	...
    
	claim = newClaim
	
    // 未绑定PV的PVC处理
	if !metav1.HasAnnotation(claim.ObjectMeta, storagehelpers.AnnBindCompleted) {
		return ctrl.syncUnboundClaim(ctx, claim)
    // 绑定PV的PVC处理
	} else {
		return ctrl.syncBoundClaim(ctx, claim)
	}
}

注意

claimworker基于底层PVC创建存储资源及关联PV，底层不存在PVC会由claims缓存清理及将关联PV推入volumeQueue

# 4.2.syncUnbound

pc.syncUnboundClaim()会匹配条件满足的PV及执行bind绑定PV，PV未找到则provisionClaim创建存储及对应的PV对象。

// main controller method to decide what to do with an unbound claim.
func (ctrl *PersistentVolumeController) syncUnboundClaim(...) error {
	// 未关联PV
	if claim.Spec.VolumeName == "" {
        // 延迟绑定检查(关联storageClass.Mode=waitForFirstConsumer)
		delayBinding, err := storagehelpers.IsDelayBindingMode(claim, ctrl.classLister)
		...

        // 获取最匹配的PV(容量刚好够/StorageClass/AccessMode/Selector)
		volume, err := ctrl.volumes.findBestMatchForClaim(claim, delayBinding)
		...
        
        // 没有匹配的PV
		if volume == nil {
			// 启用default storageClass
			if utilfeature.DefaultFeatureGate.Enabled(features.RetroactiveDefaultStorageClass) {
                // PVC未设置storageClass,基于default storageClass补充
				updated, err := ctrl.assignDefaultStorageClass(ctx, claim)
				...
                
                // 更新后等待下一轮调度
				if updated {
					return nil
				}
			}

			switch {
            // 延迟绑定+Pod未引用(未调度)
			case delayBinding && !storagehelpers.IsDelayBindingProvisioning(claim):
                // 发布事件
				ctrl.emitEventForUnboundDelayBindingClaim(claim)
                ...
            // PVC设置了storageClass
			case storagehelpers.GetPersistentVolumeClaimClass(claim) != "":
                // 交互csi plugin进行存储创建
				ctrl.provisionClaim(ctx, claim)
                ...
                
				return nil
			...
			}

			// 调度条件不充分,PVC设为Pending,更新claims缓存
			ctrl.updateClaimStatus(ctx, claim, v1.ClaimPending, nil)
            ...
            
			return nil
		} else /* pv != nil */ {
			...
            
            // 绑定PVC和PV
            // 1.PV关联PVC,设置bound-by-controller注解和Bound状态,更新volumes缓存
            // 2.PVC关联PV,设置bound-by-controller和bind-completed注解和Bound状态,设置status容量及Mode,更新claims缓存
			ctrl.bind(ctx, volume, claim)
            ...
            
			return nil
		}
	} else /* pvc.Spec.VolumeName != nil */ {
        // 获取volumes缓存PV
		obj, found, err := ctrl.volumes.store.GetByKey(claim.Spec.VolumeName)
		...
        
        // 未找到
		if !found {
            // PVC设置Pending
			ctrl.updateClaimStatus(ctx, claim, v1.ClaimPending, nil)
            ...
            
			return nil
		} else {
			volume, ok := obj.(*v1.PersistentVolume)
			...
            
            // PV还未绑定PVC
			if volume.Spec.ClaimRef == nil {
                // 容量/storageClass/volumeMode/accessMode匹配
				if !checkVolumeSatisfyClaim(volume, claim) {
					// PV不匹配PVC,PVC设为Pending
					ctrl.updateClaimStatus(ctx, claim, v1.ClaimPending, nil)
                    ...
                // PVC和PV绑定
				} else {
                    ...
					ctrl.bind(ctx, volume, claim)
				}

				return nil
            // PV绑定到当前PVC
			} else if storagehelpers.IsVolumeBoundToClaim(volume, claim) {
				ctrl.bind(ctx, volume, claim)
                ...
                
				return nil
            // PV绑定到其它PVC
			} else {
				// PVC期望的PV是用户指定的
				if !metav1.HasAnnotation(claim.ObjectMeta, storagehelpers.AnnBoundByController) {
					// PV设为Pending
					ctrl.updateClaimStatus(ctx, claim, v1.ClaimPending, nil)
                    ...
                    
					return nil
				}
                ...
			}
		}
	}
}

注意

syncUnboundClaim基于PVC和PV匹配情况更新状态及绑定，比较重要的是provisionClaim动态供应模块

# 4.3.provisionClaim

pc.provisionClaim()基于PVC定义匹配csi plugin，执行存储及PV创建，没有匹配的plugin会标记由external-provisioner处理。

// provisions a volume using external provisioner.
func (ctrl *PersistentVolumeController) provisionClaimOperationExternal(...) (string, error) {
	claimClass := storagehelpers.GetPersistentVolumeClaimClass(claim)
	...
    
    // 供应者
	provisionerName := storageClass.Provisioner
    
    // 迁移到CSI
	if ctrl.csiMigratedPluginManager.IsMigrationEnabledForPlugin(storageClass.Provisioner) {
		// provisioner转为迁移后的csi provisioner
		provisionerName, err = ctrl.translator.GetCSINameFromInTreeName(storageClass.Provisioner)
		...
	}
    
	// 更新PVC的provisioner annotation,更新claims缓存
	newClaim, err := ctrl.setClaimProvisioner(ctx, claim, provisionerName)
	...
	
	return provisionerName, nil
}

// provisions a volume. This method is running in standalone goroutine and already has all necessary locks.
func (ctrl *PersistentVolumeController) provisionClaimOperation(...) (string, error) {
	claimClass := storagehelpers.GetPersistentVolumeClaimClass(claim)
	...
    
	provisionerName := storageClass.Provisioner

	// 更新PVC的provisioner annotation,更新claims缓存
	newClaim, err := ctrl.setClaimProvisioner(ctx, claim, provisionerName)
	...
    
	claim = newClaim

    // 获取PVC关联PV——pvc-claim.UID
	pvName := ctrl.getProvisionedVolumeNameForClaim(claim)
	volume, err := ctrl.kubeClient.CoreV1().PersistentVolumes().Get(ctx, pvName, metav1.GetOptions{})
	...
    
    // 存储已创建
	if err == nil && volume != nil {
		// Volume has been already provisioned, nothing to do.
		return pluginName, err
	}

	// 准备claimRef
	claimRef, err := ref.GetReference(scheme.Scheme, claim)
	...

	// plugin不支持PV Mount方式
	if !plugin.SupportsMountOption() && len(options.MountOptions) > 0 {
		...
		return pluginName, fmt.Errorf("provisioner %q doesn't support mount options", plugin.GetPluginName())
	}

	// 创建provisioner实例
	provisioner, err := plugin.NewProvisioner(logger, options)
	...

    // scheduler已经为PVC选中节点
	if nodeName, ok := claim.Annotations[storagehelpers.AnnSelectedNode]; ok {
		selectedNode, err = ctrl.NodeLister.Get(nodeName)
		...
	}
    ...
    
    // 创建存储卷及生成PV对象定义
	volume, err = provisioner.Provision(selectedNode, allowedTopologies)
	...

	// 填充PV属性
	if volume.Name == "" {
		volume.Name = pvName
	}
	// Bind it to the claim
	volume.Spec.ClaimRef = claimRef
	volume.Status.Phase = v1.VolumeBound
	volume.Spec.StorageClassName = claimClass

	// Add AnnBoundByController (used in deleting the volume)
	metav1.SetMetaDataAnnotation(&volume.ObjectMeta, storagehelpers.AnnBoundByController, "yes")
	metav1.SetMetaDataAnnotation(&volume.ObjectMeta, storagehelpers.AnnDynamicallyProvisioned, plugin.GetPluginName())

    // 严格遵守PV ReclaimPolicy
	if utilfeature.DefaultFeatureGate.Enabled(features.HonorPVReclaimPolicy) {
        // 删除回收策略
		if volume.Spec.PersistentVolumeReclaimPolicy == v1.PersistentVolumeReclaimDelete {
			// PV补充删除保护finalizer,避免存储泄漏
			volume.SetFinalizers([]string{storagehelpers.PVDeletionInTreeProtectionFinalizer})
		}
	}

	// 重试5次
	for i := 0; i < ctrl.createProvisionedPVRetryCount; i++ {
		...
        
        // 尝试创建PV
		if newVol, err = ctrl.kubeClient.CoreV1().PV().Create(ctx, volume, metav1.CreateOptions{}); err == nil || apierrors.IsAlreadyExists(err) {
			// Save succeeded.
			if err != nil {
				err = nil
			} else {
				// 更新volumes缓存
				_, updateErr := ctrl.storeVolumeUpdate(logger, newVol)
			}
			break
		}
        // 5s重试一次
		time.Sleep(ctrl.createProvisionedPVInterval)
	}

    // PV创建错误,回收存储
	if err != nil {
		...
        
        // 重试5次
		for i := 0; i < ctrl.createProvisionedPVRetryCount; i++ {
            // 匹配deleter plugin删除存储
			_, deleted, deleteErr = ctrl.doDeleteVolume(ctx, volume)
			if deleteErr == nil && deleted {
				break
			}
            
            // plugin找不到、plugin deleter创建失败、delete失败都返回false
            // in-tree plugin接受存储泄漏,CSI基于对账GC规避了该风险
			if !deleted {
				break
			}
			time.Sleep(ctrl.createProvisionedPVInterval)
		}
		...
	}
    ...
    
	return pluginName, nil
}

// starts new asynchronous operation to provision a claim if provisioning is enabled.
func (ctrl *PersistentVolumeController) provisionClaim(...) error {
    // 未启用动态供应
	if !ctrl.enableDynamicProvisioning {
		return nil
	}
	...
    
    // 基于PVC定义匹配csi provisionable plugin
    // volumeMgr会注册csiPlugin,仅作识别作用,不负责provision
	plugin, storageClass, err := ctrl.findProvisionablePlugin(claim)
	...
    
    // 执行创建
	ctrl.scheduleOperation(logger, opName, func() error {
		...
        
        // 无匹配plugin
		if plugin == nil {
            // PVC Annotation设置,驱动external-provisioner创建
			_, err = ctrl.provisionClaimOperationExternal(ctx, claim, storageClass)
        // in-tree plugin创建
		} else {
            // 调用插件创建存储
			_, err = ctrl.provisionClaimOperation(ctx, claim, plugin, storageClass)
		}
		...
        
		return err
	})
	return nil
}

注意

provisionClaim根据in-tree plugin匹配情况决定是否负责volume及PVC创建

# 4.4.syncBound

pc.syncBoundClaim()负责处理PVC已绑定的各种异常情况，维护PVC和PV的绑定关系及状态，避免多个PVC绑定到同一个PV。

// main controller method to decide what to do with a bound claim.
func (ctrl *PersistentVolumeController) syncBoundClaim(...) error {
	
	// PVC标记bindComplete Annotation,关联PV为空
	if claim.Spec.VolumeName == "" {
        // 状态更新为Lost,更新claims缓存
		ctrl.updateClaimStatusWithEvent(ctx, claim, v1.ClaimLost, nil, v1.EventTypeWarning, ...)
		return nil
	}
    
    // 查询volumes缓存的PV
	obj, found, err := ctrl.volumes.store.GetByKey(claim.Spec.VolumeName)
	...
    
    // 未找到
	if !found {
		// PVC绑定不存在的PV
		ctrl.updateClaimStatusWithEvent(ctx, claim, v1.ClaimLost, nil, v1.EventTypeWarning, ...)
		return nil
	} else {
		volume, ok := obj.(*v1.PersistentVolume)
		...

		// PV关联PVC为空
		if volume.Spec.ClaimRef == nil {
			// 当前PVC和PV绑定,更新状态及volules/claims缓存
			ctrl.bind(ctx, volume, claim)
            ...
            
			return nil
        // PV绑定的就是当前PVC
		} else if volume.Spec.ClaimRef.UID == claim.UID {
            // 当前PVC和PV绑定,更新状态及volules/claims缓存
			ctrl.bind(ctx, volume, claim)
            ...
            
			return nil
        // PV绑定到其它PVC
		} else {
			// 状态更新为Lost,更新claims缓存
			ctrl.updateClaimStatusWithEvent(ctx, claim, v1.ClaimLost, nil, v1.EventTypeWarning, ...)
            ...
            
			return nil
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

注意

syncBoundClaim用于解决PVC绑定的各种异常情况，重置PVC状态或找到目标PV尝试修复绑定关系。