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

# 1.简介

# 1.1.功能

AD Controller负责创建、删除及同步VolumeAttachment对象，交互csi plugin完成存储的attach/detach，实现存储和node附着分离。
注意

不同volume plugin的attach/detach行为不同，csi plugin由外部完成attach/detach，ad controller仅负责更新对象状态

# 1.2.原理

ad controller和PV/PVC/Pod密切绑定，Pod正常调度且关联PVC处于Bound状态，ad controller会执行卷的attach/detach操作。
注意

ad controller会维护desired state和actual state状态，基于实际与期望对比执行volume的attach/detach操作

# 2.入口

# 2.1.start

startAttachDetachController()会加载attachable plugin及实例化attachDetachController，执行ad.Run()激活处理协程。

// NewAttachDetachController returns a new instance of AttachDetachController.
func NewAttachDetachController(...) (AttachDetachController, error) {

	adc := &attachDetachController{
		...
		pvcLister:           pvcInformer.Lister(),					// pvc缓存
		...
		pvLister:            pvInformer.Lister(),					// pv缓存
		...
		podLister:           podInformer.Lister(),					// pod缓存
		...
		podIndexer:          podInformer.Informer().GetIndexer(),	// pod<-->pvc索引
		nodeLister:          nodeInformer.Lister(),					// node缓存
		...
		pvcQueue:            workqueue.NewNamedRateLimitingQueue(DefaultControllerRateLimiter(), "pvcs"),
		...
	}

	adc.csiNodeLister = csiNodeInformer.Lister()						// csiNode缓存
	...

	adc.csiDriverLister = csiDriverInformer.Lister()					// csiDriver缓存
	...

	adc.volumeAttachmentLister = volumeAttachmentInformer.Lister()		// volumeAttachment缓存
	...

    // plugin实例化及flex plugin监听
	adc.volumePluginMgr.InitPlugins(plugins, prober, adc)
    ...

    // dsw缓存初始化
	adc.desiredStateOfWorld = cache.NewDesiredStateOfWorld(&adc.volumePluginMgr)
    // asw缓存初始化
	adc.actualStateOfWorld = cache.NewActualStateOfWorld(&adc.volumePluginMgr)
    
    // attach/detach执行器
	adc.attacherDetacher = operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(...))
	// node状态更新器
    adc.nodeStatusUpdater = statusupdater.NewNodeStatusUpdater(...)

	// 协调器
	adc.reconciler = reconciler.NewReconciler(...)

    // plugin迁移csi检测相关
	csiTranslator := csitrans.New()
	adc.intreeToCSITranslator = csiTranslator
	adc.csiMigratedPluginManager = csimigration.NewPluginManager(csiTranslator, utilfeature.DefaultFeatureGate)

	// dsw填充器    
	adc.desiredStateOfWorldPopulator = populator.NewDesiredStateOfWorldPopulator(...)

    // pod监听回调
	podInformer.Informer().AddEventHandler(kcache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) { adc.podAdd(logger, obj) },
		UpdateFunc: func(oldObj, newObj interface{}) { adc.podUpdate(logger, oldObj, newObj) },
		DeleteFunc: func(obj interface{}) { adc.podDelete(logger, obj) },
	})

	// pod<-->pvc索引
	common.AddPodPVCIndexerIfNotPresent(adc.podIndexer)
    ...

    // node监听回调
	nodeInformer.Informer().AddEventHandler(kcache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) { adc.nodeAdd(logger, obj) },
		UpdateFunc: func(oldObj, newObj interface{}) { adc.nodeUpdate(logger, oldObj, newObj) },
		DeleteFunc: func(obj interface{}) { adc.nodeDelete(logger, obj) },
	})

    // PVC监听回调
	pvcInformer.Informer().AddEventHandler(kcache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) { adc.enqueuePVC(obj) },
		UpdateFunc: func(old, new interface{}) { adc.enqueuePVC(new) },
	})

	return adc, nil
}

func startAttachDetachController(...) (controller.Interface, bool, error) {
	...

    // 加载attachable plugin(fc/iscsi/csi)
	plugins, err := ProbeAttachableVolumePlugins(logger)
	...

	// 实例化ad controller
	attachDetachController, attachDetachControllerErr := attachdetach.NewAttachDetachController(...)
	...
    
	go attachDetachController.Run(ctx)
	return nil, true, 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

注意

ad controller会监听多种资源，维护PVC的预期及实际状态，必要时进行attach/detach协调处理

# 2.2.podInformer

adc controller基于PodInformer监听已调度Pod，已调度Pod关联PVC可处理，根据node-volume-pod格式缓存到dsw作为预期状态。

func (adc *attachDetachController) podAdd(logger klog.Logger, obj interface{}) {
	pod, ok := obj.(*v1.Pod)
	...
    
	if pod.Spec.NodeName == "" {
		// Ignore pods without NodeName, indicating they are not scheduled.
		return
	}

    // 行为检查
	volumeActionFlag := util.DetermineVolumeAction(
		pod, adc.desiredStateOfWorld, true /* default volume action */)
	
    // dsw/asw同步
	util.ProcessPodVolumes(...)
}

func (adc *attachDetachController) podUpdate(logger klog.Logger, oldObj, newObj interface{}) {
    adc.podAdd(logger,newObj)
}

func (adc *attachDetachController) podDelete(logger klog.Logger, obj interface{}) {
	pod, ok := obj.(*v1.Pod)
	...

	util.ProcessPodVolumes(...)
}

// processes the volumes in the given pod and adds them to the dsw if addVolumes is true, otherwise it removes.
func ProcessPodVolumes(...) {
	if pod == nil || len(pod.Spec.Volumes) <= 0 {
		return
	}

    // 获取pod调度节点
	nodeName := types.NodeName(pod.Spec.NodeName)
    
    // 未调度
	if nodeName == "" {
		return
    // node还未被dsw接管
	} else if !desiredStateOfWorld.NodeExists(nodeName) {
		return
	}

	// 依次处理Pod Volume
	for _, podVolume := range pod.Spec.Volumes {
        // pod volume转为plugin volume格式
		volumeSpec, err := CreateVolumeSpec(logger, podVolume, pod, nodeName, volumePluginMgr, pvcLister, pvLister, csiMigratedPluginManager, csiTranslator)
		...

        // 匹配attachable plugin
		attachableVolumePlugin, err := volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
        // 无法处理的跳过
		if err != nil || attachableVolumePlugin == nil {
			continue
		}

        // pod uid作为身份
		uniquePodName := util.GetUniquePodName(pod)
        
        // Pod正在运行/node支持保留terminated Pod卷
		if addVolumes {
			// node-volume-Pod缓存到dsw
			_, err := desiredStateOfWorld.AddPod(uniquePodName, pod, volumeSpec, nodeName)
			...
		} else {
			// pluginName-volumeName
			uniqueVolumeName, err := util.GetUniqueVolumeNameFromSpec(attachableVolumePlugin, volumeSpec)
			...
            
            // dsw的node-volume-pod缓存移除Pod
			desiredStateOfWorld.DeletePod(uniquePodName, uniqueVolumeName, nodeName)
		}
	}
	return
}

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

注意

dsw维护的pod volume按照node-volume-pod关联，以感知volume attach/detach目标及used情况

# 2.3.nodeInformer

adc controller基于nodeInformer监听node资源，利用node初始化dsw缓存作为接管标记，相关的volume会基于使用状态进行处理。

func (adc *attachDetachController) nodeAdd(logger klog.Logger, obj interface{}) {
	node, ok := obj.(*v1.Node)
	
	if node == nil || !ok {
		return
	}
    
	nodeName := types.NodeName(node.Name)
    // 走更新流程
	adc.nodeUpdate(logger, nil, obj)
    // asw标记更新
	adc.actualStateOfWorld.SetNodeStatusUpdateNeeded(logger, nodeName)
}

func (adc *attachDetachController) nodeUpdate(logger klog.Logger, oldObj, newObj interface{}) {
	node, ok := newObj.(*v1.Node)
	
	if node == nil || !ok {
		return
	}

	nodeName := types.NodeName(node.Name)
    // 基于node初始化dsw
	adc.addNodeToDswp(node, nodeName)
    // 扫描volume使用情况
	adc.processVolumesInUse(logger, nodeName, node.Status.VolumesInUse)
}

func (adc *attachDetachController) nodeDelete(logger klog.Logger, obj interface{}) {
	node, ok := obj.(*v1.Node)
    
	if node == nil || !ok {
		return
	}

	nodeName := types.NodeName(node.Name)
    // 清理dsw node缓存
	adc.desiredStateOfWorld.DeleteNode(nodeName)
    ...

    // 扫描volume使用情况
	adc.processVolumesInUse(logger, nodeName, node.Status.VolumesInUse)
}

func (adc *attachDetachController) addNodeToDswp(node *v1.Node, nodeName types.NodeName) {
    // node标记由adc controller负责attach/detach
	if _, exists := node.Annotations[volumeutil.ControllerManagedAttachAnnotation]; exists {
		...

        // 检测node标记保留terminated Pod volume
		if t, ok := node.Annotations[volumeutil.KeepTerminatedPodVolumesAnnotation]; ok {
			keepTerminatedPodVolumes = t == "true"
		}

		// node缓存到dsw
		adc.desiredStateOfWorld.AddNode(nodeName, keepTerminatedPodVolumes)
	}
}

// processes the list of volumes marked as "in-use" according to the specified Node's Status.VolumesInUse.
func (adc *attachDetachController) processVolumesInUse(...) {
	// node已经attach的volume
	for _, attachedVolume := range adc.actualStateOfWorld.GetAttachedVolumesForNode(nodeName) {
		mounted := false
        // node记录attach的volume
		for _, volumeInUse := range volumesInUse {
            // asw和node维护的volume attach状态一致
			if attachedVolume.VolumeName == volumeInUse {
				mounted = true
				break
			}
		}
        
        // 更新asw维护的volume attach状态
		adc.asw.SetVolumeMountedByNode(logger, attachedVolume.VolumeName, nodeName, mounted)
		...
	}
}

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

注意

nodeInformer监听的node向dsw注册一份期望，基于node attached volume更新asw实际挂载情况

# 2.4.adcrun

adc.Run()会调用5个方法，负责初始化dsw/asw内部状态，基于asw/dsw维护的volume attach情况执行挂载或卸载及更新Pod/PVC。

func (adc *attachDetachController) Run(ctx context.Context) {
	...

    // 同步完成
	if !kcache.WaitForNamedCacheSync("attach detach", ctx.Done(), synced...) {
		return
	}

    // 初始化asw
	adc.populateActualStateOfWorld(logger)
	...
	
    // 初始化dsw
    adc.populateDesiredStateOfWorld(logger)
	...
    
    // 基于asw/dsw协调volume attach/detach
	go adc.reconciler.Run(ctx)
    
    // 动态更新dsw
	go adc.dswp.Run(ctx)
    
    // PVC任务处理
	go wait.UntilWithContext(ctx, adc.pvcWorker, time.Second)
	...

	<-ctx.Done()
}

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

注意

前两个函数用于初始化asw/dsw缓存，adc.reconciler.Run()激活监听及协调，adc.dswp.Run动态更新dsw，pvcworker处理卷挂载

# 3.缓存

# 3.1.populateAsw

adc.populateActualStateOfWorld()基于nodeLister初始化asw/dsw缓存，同步volumeAttachment对象纠正asw维护的内部状态。

func (adc *attachDetachController) populateActualStateOfWorld(logger klog.Logger) error {
	nodes, err := adc.nodeLister.List(labels.Everything())
	...

	for _, node := range nodes {
		nodeName := types.NodeName(node.Name)
        // 已经attach到node的volume
		for _, attachedVolume := range node.Status.VolumesAttached {
            // volume身份
			uniqueName := attachedVolume.Name
			
            // asw标记volume attached
			adc.asw.MarkVolumeAsAttached(logger, uniqueName, nil, nodeName, attachedVolume.DevicePath)
			...
            
            // 基于node attached更新asw volume attach状态
			adc.processVolumesInUse(logger, nodeName, node.Status.VolumesInUse)
            
            // 注册node至dsw
			adc.addNodeToDswp(node, types.NodeName(node.Name))
		}
	}
    
    // 同步volumeAttachement
	adc.processVolumeAttachments(logger)
	...
    
	return err
}

// Process Volume-Attachment objects.
func (adc *attachDetachController) processVolumeAttachments(logger klog.Logger) error {
    // 获取volumeAttachment对象
	vas, err := adc.volumeAttachmentLister.List(labels.Everything())
	...
    
	for _, va := range vas {
        // nodeName
		nodeName := types.NodeName(va.Spec.NodeName)
        // PVName
		pvName := va.Spec.Source.PersistentVolumeName
		if pvName == nil {
			continue
		}
        
        // 获取关联PV
		pv, err := adc.pvLister.Get(*pvName)
		...

		// 生成plugin volume
		volumeSpec := volume.NewSpecFromPersistentVolume(pv, false)

		// 匹配volume对应in-tree plugin
		if pluginName, err := adc.csiMigratedPluginManager.GetInTreePluginNameFromSpec(pv, nil); err == nil {
            // in-tree plugin迁移至CSI
			if adc.csiMigratedPluginManager.IsMigrationEnabledForPlugin(inTreePluginName) {
				// 匹配CSI Plugin
				plugin, _ = adc.volumePluginMgr.FindAttachablePluginByName(csi.CSIPluginName)
				// volumeSpec转换
				volumeSpec, err = csimigration.TranslateInTreeSpecToCSI(volumeSpec, "" /* podNamespace */, adc.intreeToCSITranslator)
				...
			}
		}

        // in-tree plugin处理
		if plugin == nil {
            // 查询attachable in-tree plugin
			plugin, err = adc.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
			...
		}

        // 生成volume身份
		volumeName, err := volumeutil.GetUniqueVolumeNameFromSpec(plugin, volumeSpec)
		...
        
        // 获取asw缓存的volume attach状态
		attachState := adc.actualStateOfWorld.GetAttachState(volumeName, nodeName)
        // asw未缓存volume attach状态
		if attachState == cache.AttachStateDetached {
            // asw缓存volume attach状态及标记不确定
			adc.actualStateOfWorld.MarkVolumeAsUncertain(logger, volumeName, volumeSpec, nodeName)
			...
		}
	}
	return nil
}

注意

asw基于node.status.volumeAttached初始化，基于volumeAttachment维护的PV Attach状态纠正

# 3.2.populateDsw

adc.populateDesiredStateOfWorld()基于podLister初始化dsw缓存的预期状态，以对账asw执行后续的reconciler动作。

func (adc *attachDetachController) populateDesiredStateOfWorld(logger klog.Logger) error {
	// 获取Pod对象
    pods, err := adc.podLister.List(labels.Everything())
	...
    
    // 初次加载
	for _, pod := range pods {
		podToAdd := pod
        // dsw缓存更新
		adc.podAdd(logger, podToAdd)
		for _, podVolume := range podToAdd.Spec.Volumes {
			nodeName := types.NodeName(podToAdd.Spec.NodeName)
			// 生成plugin volumeSpec 
			volumeSpec, err := util.CreateVolumeSpec(logger, podVolume, podToAdd, nodeName, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister, adc.csiMigratedPluginManager, adc.intreeToCSITranslator)
			...
            
            // 匹配attachable plugin
			plugin, err := adc.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
			...
            
            // 生成volume身份
			volumeName, err := volumeutil.GetUniqueVolumeNameFromSpec(plugin, volumeSpec)
			...
            
            // 获取asw缓存的volume attach状态
			attachState := adc.actualStateOfWorld.GetAttachState(volumeName, nodeName)
            
            // 确认attached
			if attachState == cache.AttachStateAttached {
				devicePath, _ := adc.getNodeVolumeDevicePath(volumeName, nodeName)
				...
                // 更新asw缓存的volume attach状态
				adc.actualStateOfWorld.MarkVolumeAsAttached(logger,volumeName, volumeSpec, nodeName, devicePath)
				...
			}
		}
	}

	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

注意

dsw基于pod volumes初始化，进一步基于pod volumes更新asw attachedVolume的devicePath

# 3.3.dswPopulator

dswp.Run()会间隔1min同步deletedPod和activePod，基于同步检测结果更新dsw缓存，以对账asw进行存储回收或创建协调。

func (dswp *desiredStateOfWorldPopulator) Run(ctx context.Context) {
	wait.UntilWithContext(ctx, dswp.populatorLoopFunc(ctx), 1min)
}

func (dswp *desiredStateOfWorldPopulator) populatorLoopFunc(ctx context.Context) func(ctx context.Context) {
	return func(ctx context.Context) {
		// 基于deletedPod清理dsw缓存
		dswp.findAndRemoveDeletedPods(logger)

		// 间隔3min查询一次activePod
		if time.Since(dswp.timeOfLastListPods) < 3min {
			return
		}
        
        // 基于activePod更新dsw缓存
		dswp.findAndAddActivePods(logger)
	}
}

// Iterate through all pods in dsw, and remove if they no longer exist in the informer.
func (dswp *desiredStateOfWorldPopulator) findAndRemoveDeletedPods(logger klog.Logger) {
    // 遍历dsw已维护的Pod
	for dswPodUID, dswPodToAdd := range dswp.desiredStateOfWorld.GetPodToAdd() {
		...

		namespace, name, err := kcache.SplitMetaNamespaceKey(dswPodKey)
		...
        
        // 获取Pod对象
		informerPod, err := dswp.podLister.Pods(namespace).Get(name)
		switch {
		case errors.IsNotFound(err):
			// if we can't find the pod, we need to delete it below
		case err != nil:
			continue
		default:
            // Pod终止+卷保留 Pod正常均需继续维护
            volumeActionFlag := util.DetermineVolumeAction(informerPod, dswp.desiredStateOfWorld, true)

            // 继续维护
			if volumeActionFlag {
                // UID作为Pod身份
				informerPodUID := volutil.GetUniquePodName(informerPod)
				// 身份一致无需更新
				if informerPodUID == dswPodUID {
					continue
				}
			}
		}

		// Pod未找到/无需继续维护Pod,清理缓存
		dswp.desiredStateOfWorld.DeletePod(dswPodUID, dswPodToAdd.VolumeName, dswPodToAdd.NodeName)
	}

	// dsw维护的attaching volume
	for _, volumeToAttach := range dswp.desiredStateOfWorld.GetVolumesToAttach() {
        // 检查volume attachable
		volumeAttachable := volutil.IsAttachableVolume(volumeToAttach.VolumeSpec, dswp.volumePluginMgr)
        // 可attach-->不可attach
		if !volumeAttachable {
            // 清理相关Pod映射
			for _, scheduledPod := range volumeToAttach.ScheduledPods {
				podUID := volutil.GetUniquePodName(scheduledPod)
				dswp.desiredStateOfWorld.DeletePod(podUID, volumeToAttach.VolumeName, volumeToAttach.NodeName)
			}
		}
	}
}

// Iterate through all pods, and add dsw if not terminated.
func (dswp *desiredStateOfWorldPopulator) findAndAddActivePods(logger klog.Logger) {
	// 列出所有Pod
    pods, err := dswp.podLister.List(labels.Everything())
	...
    
    // 记录Pod查询时间
	dswp.timeOfLastListPods = time.Now()

    // 遍历Pod
	for _, pod := range pods {
        // Pod终止
		if volutil.IsPodTerminated(pod, pod.Status) {
			continue
		}
        
        // 注册到dsw的node-->volume-->Pod结构
		util.ProcessPodVolumes(logger, pod, true, dswp.desiredStateOfWorld, dswp.volumePluginMgr, dswp.pvcLister, dswp.pvLister, dswp.csiMigratedPluginManager, dswp.intreeToCSITranslator)
	}

}

注意

dswp间隔1min同步deletedPod触发dsw缓存清理，间隔3min同步activePod触发dsw缓存更新

# 3.4.pvcworker

pvcworker会由pvcQueue队列获取pvc对象，查询pvc关联的pod列表，基于pod列表更新dsw维护的期望状态。

// pvcWorker processes items from pvcQueue
func (adc *attachDetachController) pvcWorker(ctx context.Context) {
	for adc.processNextItem(klog.FromContext(ctx)) {
	}
}

func (adc *attachDetachController) processNextItem(logger klog.Logger) bool {
	keyObj, shutdown := adc.pvcQueue.Get()
	...
	defer adc.pvcQueue.Done(keyObj)

    // 执行同步
	if err := adc.syncPVCByKey(logger, keyObj.(string)); err != nil {
		adc.pvcQueue.AddRateLimited(keyObj)
		runtime.HandleError(fmt.Errorf("Failed to sync pvc %q, will retry again: %v", keyObj.(string), err))
		return true
	}
	
	adc.pvcQueue.Forget(keyObj)
	return true
}

func (adc *attachDetachController) syncPVCByKey(logger klog.Logger, key string) error {
	namespace, name, err := kcache.SplitMetaNamespaceKey(key)
	...
    
    // 获取pvc对象
	pvc, err := adc.pvcLister.PersistentVolumeClaims(namespace).Get(name)
	...

    // PVC还未绑定PV
	if pvc.Status.Phase != v1.ClaimBound || pvc.Spec.VolumeName == "" {
		// Skip unbound PVCs.
		return nil
	}

    // 基于podIndexer查询使用PVC的Pod列表
	objs, err := adc.podIndexer.ByIndex(common.PodPVCIndex, key)
	...
    
	for _, obj := range objs {
		pod, ok := obj.(*v1.Pod)
		...
        
		// Pod未调度或终止
		if len(pod.Spec.NodeName) == 0 || volumeutil.IsPodTerminated(pod, pod.Status) {
			continue
		}
        
        // Pod Volume接管条件检查(Pod正常||Pod终止+删除保留策略)
		volumeActionFlag := util.DetermineVolumeAction(pod, adc.desiredStateOfWorld, true)

        // 注册到dsw
		util.ProcessPodVolumes(logger, pod, volumeActionFlag, /* addVolumes */
			adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister, adc.csiMigratedPluginManager, adc.intreeToCSITranslator)
	}
	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
48
49
50
51
52
53
54
55
56
57
58

注意

pvcworker基于PVC关联的Pod列表更新dsw缓存，最终执行的还是processPodVolumes

# 4.协调

# 4.1.runloop

reconciler.Run()是ad controller的核心逻辑，根据dsw和asw维护的期望状态及实际状态计算attach/detach执行动作。

func (rc *reconciler) Run(ctx context.Context) {
	wait.UntilWithContext(ctx, rc.reconciliationLoopFunc(ctx), rc.loopPeriod)
}

// checks whether attached volumes from asw are still attached to node and update the status if they are not.
func (rc *reconciler) reconciliationLoopFunc(ctx context.Context) func(context.Context) {
	return func(ctx context.Context) {
		// 执行attach/detach
		rc.reconcile(ctx)

        // 信任asw状态
		if rc.disableReconciliationSync {
            logger.V(5).Info(...)
        // 同步间隔小于1s
		} else if rc.syncDuration < time.Second {
			logger.V(5).Info(...)
        // 距上次同步超过5s
		} else if time.Since(rc.timeOfLastSync) > rc.syncDuration {
            // 同步attach状态
			rc.sync()
		}
	}
}

func (rc *reconciler) sync() {
    // 更新同步时间
	defer rc.updateSyncTime()
    // 同步状态
	rc.syncStates()
}

func (rc *reconciler) syncStates() {
    // 获取node attachedVolume
	volumesPerNode := rc.actualStateOfWorld.GetAttachedVolumesPerNode()
    // 验证attach状态
	rc.attacherDetacher.VerifyVolumesAreAttached(volumesPerNode, rc.actualStateOfWorld)
}

func (oe *operationExecutor) VerifyVolumesAreAttached(...) {
	...

	for node, nodeAttachedVolumes := range attachedVolumes {
		needIndividualVerifyVolumes := []AttachedVolume{}
		for _, volumeAttached := range nodeAttachedVolumes {
            // 略过不完整的volume
			if volumeAttached.VolumeSpec == nil {
				continue
			}

            // 匹配volume plugin
			plugin, _ := oe.operationGenerator.GetVolumePluginMgr().FindPluginBySpec(volumeAttached.VolumeSpec)
			...
            
			if plugin == nil {
				// should never happen since FindPluginBySpec always returns error if volumePlugin = nil
				continue
			}

			pluginName := plugin.GetPluginName()

            // plugin支持批量验证
			if plugin.SupportsBulkVolumeVerification() {
				...

				// 统计到批量验证列表
				volumeSpecList = append(volumeSpecList, volumeAttached.VolumeSpec)
				pluginNodes[node] = volumeSpecList
				bulkVerifyPluginsByNode[pluginName] = pluginNodes
                ...
                
                // 更新volumeSpec<-->volumeName映射
				volumeSpecMap[volumeAttached.VolumeSpec] = volumeAttached.VolumeName
				volumeSpecMapByPlugin[pluginName] = volumeSpecMap
				continue
			}
            
			// 未支持批量验证,统计到单独验证
			needIndividualVerifyVolumes = append(needIndividualVerifyVolumes, volumeAttached)
		}
        
        // 基于operation交互plugin检查volume attach状态,更新asw detach部分
		oe.VerifyVolumesAreAttachedPerNode(needIndividualVerifyVolumes, node, actualStateOfWorld)
		...
	}

    // 批量验证
	for pluginName, pluginNodeVolumes := range bulkVerifyPluginsByNode {
        // operaton,负责交互plugin批量检查volume attach状态,更新asw detach部分
		generatedOperations, err := oe.operationGenerator.GenerateBulkVolumeVerifyFunc(
			pluginNodeVolumes,
			pluginName,
			volumeSpecMapByPlugin[pluginName],
			actualStateOfWorld)
		...

		// 相同plugin的批量检查
		oe.pendingOperations.Run(uniquePluginName, "" /* Pod Name */, "" /* nodeName */, generatedOperations)
		...
	}
}

注意

bulkVerify批量指的是plugin内批量，不同plugin间不会批量

# 4.2.reconcile

rc.reconcile()会对比asw/dsw缓存，优先执行detach行为释放存储，再执行attach行为供应Pod Volume，执行成功会创建VA对象。

func (rc *reconciler) reconcile(ctx context.Context) {
	// 遍历asw attachedVolume
	for _, attachedVolume := range rc.actualStateOfWorld.GetAttachedVolumes() {
        // dsw未接管
		if !rc.desiredStateOfWorld.VolumeExists(attachedVolume.VolumeName, attachedVolume.NodeName) {
			// volume支持attach到多节点,detach限制node级别并行
			if util.IsMultiAttachAllowed(attachedVolume.VolumeSpec) {
                // volume detach正在进行
				if !rc.attacherDetacher.IsOperationSafeToRetry(volumeName, "", nodeName, "volume_detach") {
					continue
				}
			} else {
                // volume仅支持attach到一个节点,detach限制cluster级别并行
				if !rc.attacherDetacher.IsOperationSafeToRetry(volumeName, "", nodeName, "volume_detach") {
					continue
				}
			}

			// 获取asw记录的volume attach状态
			attachState := rc.actualStateOfWorld.GetAttachState(volumeName, nodeName)
            
            // detached状态
			if attachState == cache.AttachStateDetached {
				continue
			}

			// 计算volume detach请求时间
			elapsedTime, err := rc.actualStateOfWorld.SetDetachRequestTime(logger, volumeName, nodeName)
			...
            
			// 超时检查
			timeout := elapsedTime > 6s

            // 节点健康检查
			isHealthy, err := rc.nodeIsHealthy(attachedVolume.NodeName)
			...
            
			// 超时+节点不健康会强制detach
			forceDetach := !isHealthy && timeout

            // node out-of-service taint检查
			hasOutOfServiceTaint, err := rc.hasOutOfServiceTaint(attachedVolume.NodeName)
			...

			// 节点正常+volume attached
			if attachedVolume.MountedByNode && !forceDetach && !hasOutOfServiceTaint {
				continue
			}

			// detach前清理asw缓存
			rc.actualStateOfWorld.RemoveVolumeFromReportAsAttached(volumeName, nodeName)
			...

			// 更新node attachedVolume
			if rc.nodeStatusUpdater.UpdateNodeStatusForNode(logger, attachedVolume.NodeName) != nil {
				// 更新出错恢复asw缓存
				rc.actualStateOfWorld.AddVolumeToReportAsAttached(logger, volumeName, nodeName)
				continue
			}
			...
            
            // 执行detach
			if rc.attacherDetacher.DetachVolume(logger, attachedVolume, verifySafeToDetach, rc.asw) != nil {
                // 出错恢复asw缓存
				rc.actualStateOfWorld.AddVolumeToReportAsAttached(logger, volumeName, nodeName)
				...
			}
		}
	}

    // 执行attach操作
	rc.attachDesiredVolumes(logger)

	// 更新node attachedVolume
	rc.nodeStatusUpdater.UpdateNodeStatuses(logger)
	...
}

注意

detach根据cluster/node粒度限制并发执行，detach前先清理asw缓存，再更新node attachedVolume

# 4.3.detach

rc.attacherDetacher.DetachVolume()负责执行volume detach行为，生成的operationHook基于node/cluster粒度并发执行。

func (og *operationGenerator) GenerateDetachVolumeFunc(...) (volumetypes.GeneratedOperations, error) {
	...

    // attachedVolume完整
	if volumeToDetach.VolumeSpec != nil {
        // 匹配detachable plugin
		attachableVolumePlugin, err = findDetachablePluginBySpec(volumeToDetach.VolumeSpec, og.volumePluginMgr)
		...

        // 获取volumeName
		volumeName, err = attachableVolumePlugin.GetVolumeName(volumeToDetach.VolumeSpec)
		...
	} else {
		// 解析detachable pluginName
		pluginName, volumeName, err = util.SplitUniqueName(volumeToDetach.VolumeName)
		...
	
        // 获取detachable plugin
		attachableVolumePlugin, err = og.volumePluginMgr.FindAttachablePluginByName(pluginName)
		...
	}

    // 补充pluginName
	if pluginName == "" {
		pluginName = attachableVolumePlugin.GetPluginName()
	}

    // 初始化detacher
	volumeDetacher, err := attachableVolumePlugin.NewDetacher()
	...

    // detachHook
	detachVolumeFunc := func() volumetypes.OperationContext {
		...
        
        // 开启安全验证
		if verifySafeToDetach {
            // 检查volume占用
			err = og.verifyVolumeIsSafeToDetach(volumeToDetach)
		}
        
        // volume未被Pod使用
		if err == nil {
            // 执行detach
			err = volumeDetacher.Detach(volumeName, volumeToDetach.NodeName)
		}
		...
        
        // detach失败
		if err != nil {
			// asw attachedVolume标记为不确定
			uncertainError := actualStateOfWorld.MarkVolumeAsUncertain(logger, volumeName, volumeSpec, nodeName)
			...
			return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
		}

		// 清理asw缓存
		actualStateOfWorld.MarkVolumeAsDetached(volumeToDetach.VolumeName, volumeToDetach.NodeName)

		return volumetypes.NewOperationContext(nil, nil, migrated)
	}

    // 返回封包对象
	return volumetypes.GeneratedOperations{
		OperationName:     DetachOperationName,
		OperationFunc:     detachVolumeFunc,
		CompleteFunc:      util.OperationCompleteHook(util.GetFullQualifiedPluginNameForVolume(pluginName, volumeToDetach.VolumeSpec), DetachOperationName),
		EventRecorderFunc: nil, // nil because we do not want to generate event on error
	}, nil
}

func (oe *operationExecutor) DetachVolume(...) error {
    // 构造operationHook
	operations, err := oe.operationGenerator.GenerateDetachVolumeFunc(logger, volumeToDetach, verifySafeToDetach, actualStateOfWorld)
	...

    // volume支持attach到多节点
	if util.IsMultiAttachAllowed(volumeToDetach.VolumeSpec) {
        // node粒度限流执行detach
		return oe.pendingOperations.Run(volumeToDetach.VolumeName, "" , volumeToDetach.NodeName, operations)
	}
    
    // cluster粒度限流执行detach
	return oe.pendingOperations.Run(volumeToDetach.VolumeName, "" /* podName */, "" /* nodeName */, operations)
}

注意

detachVolume基于volumeSpec/pluginName匹配对应插件，执行detacher.Detach()及清理ASW 缓存

# 4.4.attach

rc.attachDesiredVolumes()与detach行为相反，基于DSW队长ASW，ASW未接管的volume会交互Plugin执行Attach操作。

func (rc *reconciler) attachDesiredVolumes(logger klog.Logger) {
	// Ensure volumes that should be attached are attached.
	for _, volumeToAttach := range rc.desiredStateOfWorld.GetVolumesToAttach() {
        // volume支持attach到多节点
		if util.IsMultiAttachAllowed(volumeToAttach.VolumeSpec) {
			// node级别并发检查
			rc.attacherDetacher.IsOperationPending(volumeName, "" /* podName */, nodeName)
            ...
		} else {
			// cluster级别并发检查
			rc.attacherDetacher.IsOperationPending(olumeName, "" /* podName */, "" /* nodeName */)
		}

		// ASW缓存volume attach状态
		attachState := rc.actualStateOfWorld.GetAttachState(volumeToAttach.VolumeName, volumeToAttach.NodeName)
		// Attached
        if attachState == cache.AttachStateAttached {
            // 重置detach请求时间
			rc.actualStateOfWorld.ResetDetachRequestTime(logger, volumeName, nodeName)
			continue
		}

        // volume不支持attach多节点
		if !util.IsMultiAttachAllowed(volumeToAttach.VolumeSpec) {
            // volume attach的节点列表
			nodes := rc.actualStateOfWorld.GetNodesForAttachedVolume(volumeToAttach.VolumeName)
            // volume attach到其它节点
			if len(nodes) > 0 {
                // volumeToAttach未标记过多挂载错误
				if !volumeToAttach.MultiAttachErrorReported {
					...
                    // 更新DSW缓存
					rc.desiredStateOfWorld.SetMultiAttachError(volumeName, nodeName)
				}
				continue
			}
		}

		// 执行Attach
		rc.attacherDetacher.AttachVolume(logger, volumeToAttach.VolumeToAttach, rc.actualStateOfWorld)
		...
	}
}

func (og *operationGenerator) GenerateAttachVolumeFunc(...) volumetypes.GeneratedOperations {
	// attachHook
	attachVolumeFunc := func() volumetypes.OperationContext {
        // 匹配attachable plugin
		attachableVolumePlugin, err := og.volumePluginMgr.FindAttachablePluginBySpec(volumeToAttach.VolumeSpec)
		...

        // 初始化attacher
		volumeAttacher, newAttacherErr := attachableVolumePlugin.NewAttacher()
		...

        // 执行Attach(顺带创建volumeAttachment)
		devicePath, attachErr := volumeAttacher.Attach(volumeToAttach.VolumeSpec, volumeToAttach.NodeName)

		if attachErr != nil {
			...
            
            // 向ASW注册volume uncertain状态
			addErr := actualStateOfWorld.MarkVolumeAsUncertain(logger, volumeName, volumeSpec, uncertainNode)
			...

			return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
		}
		...

		// 向ASW注册volume attached状态
		actualStateOfWorld.MarkVolumeAsAttached(logger, v1.UniqueVolumeName(""),volumeSpec,nodeName, devicePath)
		...

		return volumetypes.NewOperationContext(nil, nil, migrated)
	}
	...

	// 匹配attachable plugin
	attachableVolumePlugin, err := og.volumePluginMgr.FindAttachablePluginBySpec(volumeToAttach.VolumeSpec)
	...
    
    // 获取pluginName
    attachableVolumePluginName = attachableVolumePlugin.GetPluginName()

	return volumetypes.GeneratedOperations{ ... }
}

func (oe *operationExecutor) AttachVolume(...) error {
    // operationHook
	operations := oe.operationGenerator.GenerateAttachVolumeFunc(logger, volumeToAttach, asw)

    // volume支持attach到多节点
	if util.IsMultiAttachAllowed(volumeToAttach.VolumeSpec) {
        // node级别限流
		return oe.pendingOperations.Run(volumeToAttach.VolumeName, "" /*podName*/, nodeName, operations)
	}

    // cluster级别限流
	return oe.pendingOperations.Run(volumeToAttach.VolumeName, "" /* podName */, "" /* nodeName */, operations)
}

注意

CSI Attach会同步创建VolumeAttachment对象，CSI Detach会同步删除VolumeAttachment对象