volumeManager
南风未起 2026-04-02 19:39:22 csi
# 1.umtvolume
# 1.1.umvolume
rc.unmountVolumes()负责对比asw/dsw维护的volume，多出的asw volume会unmount回收，解除subpath mount和mount挂载。func (rc *reconciler) unmountVolumes() { // Ensure volumes that should be unmounted are unmounted. for _, mountedVolume := range rc.asw.GetAllMountedVolumes() { // dsw未注册pod volume if !rc.dsw.PodExistsInVolume(mountedVolume.PodName, mountedVolume.VolumeName, SELinuxMountContext) { // Volume is mounted, unmount it rc.operationExecutor.UnmountVolume(mountedVolume.MountedVolume, rc.asw, rc.kubeletPodsDir) ... } } } func (oe *operationExecutor) UnmountVolume(...) error { // volumeMode检查 fsVolume, err := util.CheckVolumeModeFilesystem(volumeToUnmount.VolumeSpec) ... // filesystem if fsVolume { // Unmount a volume if a volume is mounted generatedOperations = oe.operationGenerator.GenerateUnmountFunc(volumeToUnmount, asw, podsDir) // block } else { // Unmap a volume if a volume is mapped generatedOperations = oe.operationGenerator.GenerateUnmapVolumeFunc(volumeToUnmount, asw) } ... // All volume plugins can execute unmount/unmap for multiple pods referencing the same volume in parallel podName := volumetypes.UniquePodName(volumeToUnmount.PodUID) // 执行 return oe.pendingOperations.Run(volumeToUnmount.VolumeName, podName, "" /* nodeName */, generatedOperations) }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
注意
unmount/unmap核心逻辑都在GenerateUnmountFunc和GenerateUnmapVolumeFunc
# 1.2.unmounter
oe.operationGenerator.GenerateUnmountFunc用于生成unmount volume相关的卸载函数，负责清理subpath mount和mount。func (og *operationGenerator) GenerateUnmountVolumeFunc(...) (volumetypes.GeneratedOperations, error) { // 匹配plugin volumePlugin, err := og.volumePluginMgr.FindPluginByName(volumeToUnmount.PluginName) ... // 初始化volume unmounter volumeUnmounter := volumePlugin.NewUnmounter(volumeToUnmount.InnerVolumeSpecName, volumeToUnmount.PodUID) ... unmountVolumeFunc := func() volumetypes.OperationContext { // 获取subpather subpather := og.volumePluginMgr.Host.GetSubpather() ... // Remove bind-mounts for subPaths subpather.CleanSubPaths(podDir, volumeToUnmount.InnerVolumeSpecName) ... // execute unmount(NodeUnpublishVolume) unmountErr := volumeUnmounter.TearDown() if unmountErr != nil { // mark asw volume as uncertain asw.MarkVolumeMountAsUncertain(opts) ... return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } // delete from asw volume actualStateOfWorld.MarkVolumeAsUnmounted(volumeToUnmount.PodName, volumeToUnmount.VolumeName) ... return volumetypes.NewOperationContext(nil, nil, migrated) } return volumetypes.GeneratedOperations{ OperationName: "volume_unmount", OperationFunc: unmountVolumeFunc, ... }, nil } // This implementation is shared between Linux and NsEnter func doCleanSubPaths(mounter mount.Interface, podDir string, volumeName string) error { ... // subPathDir:/var/lib/kubelet/pods/<uid>/volume-subpaths/<volume>/* containerDirs, err := ioutil.ReadDir(subPathDir) ... for _, containerDir := range containerDirs { if !containerDir.IsDir() { continue } ... // fullContainerDirPath:/var/lib/kubelet/pods/<uid>/volume-subpaths/<volume>/<container name>/* filepath.WalkDir(fullContainerDirPath, func(path string, info os.DirEntry, _ error) error { ... // pass through errors and let doCleanSubPath handle them doCleanSubPath(mounter, fullContainerDirPath, filepath.Base(path)) ... return nil }) ... // Whole container has been processed, remove its directory. os.Remove(fullContainerDirPath) ... } // pod volume subpaths have been cleaned up, remove its subpath directory. os.Remove(subPathDir) ... // podSubPathDir:/var/lib/kubelet/pods/<uid>/volume-subpaths podSubPathDir := filepath.Join(podDir, containerSubPathDirectoryName) os.Remove(podSubPathDir) ... 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
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
注意
subpath卸载仅是子目录部分，pod voume卸载才是重点，核心逻辑都在volumeUnmounter.TearDown
# 1.3.teardown
volumeUnmounter.TearDown()会执行volume unmount，不同的plugin处理模式不同，这里以emptyDir和csi实现分析。// TearDownAt simply discards everything in the directory. func (ed *emptyDir) TearDownAt(dir string) error { ... // metaDir:/var/lib/kubelet/pods/<podUID>/plugins/kubernetes.io~empty-dir/<volumeName>/ready os.RemoveAll(ed.getMetaDir()) ... // 检查mount dir目录 pathExists := mount.PathExists(dir) if !pathExists { return nil } // 介质检查 medium, isMnt, _, err := ed.mountDetector.GetMountMedium(dir, ed.medium) ... // mount point if isMnt { // mount -t tmpfs -o size=100M tmpfs dir if medium == v1.StorageMediumMemory { ed.medium = v1.StorageMediumMemory return ed.teardownTmpfsOrHugetlbfs(dir) // mount -t hugetlbfs -o pagesize=2M,size=10M hugetlbfs dir } else if medium == v1.StorageMediumHugePages { ed.medium = v1.StorageMediumHugePages return ed.teardownTmpfsOrHugetlbfs(dir) } } // assume StorageMediumDefault return ed.teardownDefault(dir) } // mountdir func (ed *emptyDir) teardownTmpfsOrHugetlbfs(dir string) error { ... // unmount volume:/var/lib/kubelet/pods/<podUID>/volumes/kubernetes.io~empty-dir/<volumeName> ed.mounter.Unmount(dir) ... os.RemoveAll(dir) ... return nil } // commondir func (ed *emptyDir) teardownDefault(dir string) error { // emptyDir quota清理 // chattr -p 1048577 dir 设置projectID // setquota -P 1048577 0 5G 0 0 dir 设置配额 // quota缓存 fsquota.ClearQuota(ed.mounter, dir) ... return os.RemoveAll(dir) } func (c *csiMountMgr) TearDownAt(dir string) error { ... // csiClient csi, err := c.csiClientGetter.Get() ... // 调用csiPlugin执行unmount csi.NodeUnpublishVolume(ctx, volID, dir) ... // dir:/var/lib/kubelet/pods/<podUID>/volumes/kubernetes.io~csi/<volumeName> // 删除volumeDir/vol_data.json/csiDir removeMountDir(c.plugin, dir) ... 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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
注意
teardown会先unmount volume再清理相关目录，以释放pod volume占用存储
# 1.4.unmapper
GenerateUnmapVolumeFunc()主要做的是unmapVolume和unmapPodDevice，释放及回收相关设备存储，清理相关的asw volume。// mapper func (og *operationGenerator) GenerateUnmapVolumeFunc(...) (volumetypes.GeneratedOperations, error) { // block mapper plugin plugin := og.volumePluginMgr.FindMapperPluginByName(volumeToUnmount.PluginName) ... // 实例化volume unmapper unmapper := plugin.NewBlockVolumeUnmapper(volumeToUnmount.InnerVolumeSpecName, volumeToUnmount.PodUID) ... unmapVolumeFunc := func() volumetypes.OperationContext { ... // /var/lib/kubelet/pods/<podUID>/volumeDevices/<plugin> podDeviceUnmapPath, volName := blockVolumeUnmapper.GetPodDeviceMapPath() // var/lib/kubelet/plugins/kubernetes.io/<plugin>/volumeDevices/<volumeHandle> globalUnmapPath := volumeToUnmount.DeviceMountPath ... // mark asw volume as uncertain asw.MarkVolumeMountAsUncertain(markVolumeOpts) ... // execute common unmap util.UnmapBlockVolume(og.blkUtil, globalUnmapPath, podDeviceUnmapPath, volName, volumeToUnmount.PodUID) ... // Call UnmapPodDevice if blockVolumeUnmapper implements CustomBlockVolumeUnmapper unmapper.(volume.CustomBlockVolumeUnmapper).UnmapPodDevice() ... // update asw volume asw.MarkVolumeAsUnmounted(volumeToUnmount.PodName, volumeToUnmount.VolumeName) ... return volumetypes.NewOperationContext(nil, nil, migrated) } return volumetypes.GeneratedOperations{ OperationName: "unmap_volume", OperationFunc: unmapVolumeFunc, ... }, nil } // utility function to provide a common way of unmapping block device path for a specified volume and pod. func UnmapBlockVolume(...) error { // 释放设备文件锁(losetup -d device) blkUtil.DetachFileDevice(filepath.Join(globalUnmapPath, string(podUID))) ... // unmap devicePath from pod volume path // remove podDevice syslink blkUtil.UnmapDevice(podDeviceUnmapPath, volumeMapName, false /* bindMount */) ... // unmap devicePath from global node path blkUtil.UnmapDevice(globalUnmapPath, string(podUID), true /* bindMount */) ... return nil } // takes a path to the attached block device and detach it from block device. func (v VolumePathHandler) DetachFileDevice(path string) error { // parse realPath from symlink. realPath, err := filepath.EvalSymlinks(path) ... // list loop device devices, err := filepath.Glob("/sys/block/loop*") ... for _, device := range devices { // read loop backing_file backingFile := fmt.Sprintf("%s/loop/backing_file", device) // The contents of this file is the absolute path of "path". data, err := ioutil.ReadFile(backingFile) ... // 格式化路径 backingFilePath := cleanBackingFilePath(string(data)) // 匹配globalUnmapPath if backingFilePath == path || backingFilePath == realPath { loopPath := fmt.Sprintf("/dev/%s", filepath.Base(device)), nil } } if len(loopPath) != 0 { // losetup -d /dev/loop0释放设备 removeLoopDevice(loopPath) ... } return nil } // UnmapDevice removes a symbolic link associated to block device under specified map path func (v VolumePathHandler) UnmapDevice(mapPath string, linkName string, bindMount bool) error { // podDevice if !bindMount { // check symlink exists linkPath := filepath.Join(mapPath, string(linkName)) ... if !v.IsSymlinkExist(linkPath) { return nil } return os.Remove(linkPath) } // globalDevice bind mount not exists if !v.IsDeviceBindMountExist(linkPath) { // Check if linkPath still exists if _, err := os.Stat(linkPath); err != nil { if !os.IsNotExist(err) { return fmt.Errorf("failed to check if path %s exists: %v", linkPath, err) } // linkPath has already been removed return nil } // Remove file os.Remove(linkPath) ... return nil } // unmount file(umount <linkPath>) mounter.Unmount(linkPath) ... // Remove file os.Remove(linkPath) ... return nil } // unmaps the block device path with custom. func (m *csiBlockMapper) UnmapPodDevice() error { ... csiClient, err := m.csiClientGetter.Get() ... // publishPath:/var/lib/kubelet/plugins/kubernetes.io/csi/volumeDevices/publish/<specName>/<podUID> // Call NodeUnpublishVolume. return m.unpublishVolumeForBlock(ctx, csiClient, publishPath) }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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
注意
unmap deivce根据losetup device-->del syslink-->unmount global-->unmount publish执行
# 2.mounter
# 2.1.mttachvolume
rc.mountOrAttachVolumes()会遍历dsw volume，基于volume和pod检查asw注册状态，未注册过的触发mount volume动作。func (rc *reconciler) mountOrAttachVolumes() { // dsw volumeToMount for _, vm := range rc.dsw.GetVolumesToMount() { mounted, devicePath := rc.asw.PodExistsInVolume(vm.PodName, vm.VolumeName, vm.PVSize, vm.SELinuxLabel) vm.DevicePath = devicePath // seLinuxLabel不匹配 if cache.IsSELinuxMountMismatchError(err) { // 注册err rc.dsw.AddErrorToPod(vm.PodName, err.Error()) continue // attaching } else if cache.IsVolumeNotAttachedError(err) { rc.waitForVolumeAttach(vm) // mounting||remounting } else if !mounted || cache.IsRemountRequiredError(err) { rc.mountAttachedVolumes(vm, err) // expanding } else if cache.IsFSResizeRequiredError(err) { fsResizeRequiredErr, _ := err.(cache.FsResizeRequiredError) rc.expandVolume(vm, fsResizeRequiredErr.CurrentSize) } } } // wait ad controller attach volume func (rc *reconciler) waitForVolumeAttach(volumeToMount cache.VolumeToMount) { ... // wait for controller to finish attaching volume. rc.operationExecutor.VerifyControllerAttachedVolume(logger,volumeToMount.VolumeToMount, rc.nodeName, rc.asw) ... } func (oe *operationExecutor) VerifyControllerAttachedVolume(...) error { operator := oe.operationGenerator.GenerateVerifyControllerAttachedVolumeFunc(logger, volumeToMount, nodeName, asw) ... return oe.pendingOperations.Run(volumeToMount.VolumeName, "" /* podName */, "" /* nodeName */, operator) } func (og *operationGenerator) GenerateVerifyControllerAttachedVolumeFunc(...) (GeneratedOperations, error) { // 匹配plugin volumePlugin := og.volumePluginMgr.FindPluginBySpec(volumeToMount.VolumeSpec) ... // For attachable volume types, lets check if volume is attached by reading from node lister. if volumeToMount.PluginIsAttachable { // node.Status.VolumesAttached cachedAttachedVolumes, _ := og.volumePluginMgr.Host.GetAttachedVolumesFromNodeStatus() if cachedAttachedVolumes != nil { _, volumeFound := cachedAttachedVolumes[volumeToMount.VolumeName] if !volumeFound { return volumetypes.GeneratedOperations{}, NewMountPreConditionFailedError(...) } } } verifyControllerAttachedVolumeFunc := func() volumetypes.OperationContext { // pvcSize claimSize := asw.GetClaimSize(volumeToMount.VolumeName) // only fetch claimSize if it was not set previously if volumeToMount.VolumeSpec.PersistentVolume != nil && claimSize == nil && !volumeToMount.VolumeSpec.InlineVolumeSpecForCSIMigration { pv := volumeToMount.VolumeSpec.PersistentVolume // 获取关联PVC pvc := og.kubeClient.CoreV1().PVC(pv.Spec.ClaimRef.Namespace).Get(..., pv.Spec.ClaimRef.Name, ...) ... // 设置pvcSize pvcStatusSize := pvc.Status.Capacity.Storage() if pvcStatusSize != nil { claimSize = pvcStatusSize } } // non-attachable if !volumeToMount.PluginIsAttachable { // 注册到asw attachedVolume asw.MarkVolumeAsAttached(logger, volumeToMount.VolumeName, volumeToMount.VolumeSpec, nodeName, "") ... asw.InitializeClaimSize(logger, volumeToMount.VolumeName, claimSize) return volumetypes.NewOperationContext(nil, nil, migrated) } // volume未使用 if !volumeToMount.ReportedInUse { return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } // fetch current node object node := og.kubeClient.CoreV1().Nodes().Get(context.TODO(), nodeName, metav1.GetOptions{}) ... // node.status.attachedVolume for _, attachedVolume := range node.Status.VolumesAttached { if attachedVolume.Name == volumeToMount.VolumeName { // 更新asw volume asw.MarkVolumeAsAttached(logger, v1.UniqueVolumeName(""), volumeToMount.VolumeSpec, nodeName, attachedVolume.DevicePath) ... actualStateOfWorld.InitializeClaimSize(logger, volumeToMount.VolumeName, claimSize) return volumetypes.NewOperationContext(nil, nil, migrated) } } return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } return volumetypes.GeneratedOperations{ OperationName: VerifyControllerAttachedVolumeOpName, OperationFunc: verifyControllerAttachedVolumeFunc, ... }, 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
注意
mountOrAttachVolumes会基于asw volume注册状态触发attach-->mount-->resize
# 2.2.mountvolume
rc.mountAttachedVolumes()负责生成mountVolumeFunc或mapVolumeFunc，处理volume/device的挂载操作及asw状态更新。func (rc *reconciler) mountAttachedVolumes(volumeToMount cache.VolumeToMount, podExistError error) { ... // Volume is not mounted, or is already mounted, but requires remounting rc.operationExecutor.MountVolume(rc.waitForAttachTimeout, volumeToMount.VolumeToMount, rc.asw, isRemount) ... } func (oe *operationExecutor) MountVolume(...) error { ... // Filesystem volume if fsVolume { // Mount/remount a volume when a volume is attached generatedOperations = oe.operationGenerator.GenerateMountVolumeFunc( waitForAttachTimeout, volumeToMount, asw, isRemount) // Block volume case } else { // Creates a map to device if a volume is attached generatedOperations, err = oe.operationGenerator.GenerateMapVolumeFunc( waitForAttachTimeout, volumeToMount, actualStateOfWorld) } ... // Avoid executing mount/map from multiple pods referencing the same volume in parallel podName := nestedpendingoperations.EmptyUniquePodName // volume plugins which are Non-attachable and Non-deviceMountable if !volumeToMount.PluginIsAttachable && !volumeToMount.PluginIsDeviceMountable { // execute mount for multiple pods referencing the same volume in parallel podName = util.GetUniquePodName(volumeToMount.Pod) } // TODO mount_device return oe.pendingOperations.Run(volumeToMount.VolumeName, podName, "" /* nodeName */, generatedOperations) }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注意
GenerateMountVolumeFunc用于filesystem volume，GenerateMapVolumeFunc用户block device volume
# 2.3.mountVFunc
oe.operationGenerator.GenerateMountVolumeFunc()负责生成挂载回调，涉及attachCheck、deviceMounter和mounter。func (og *operationGenerator) GenerateMountVolumeFunc(...) volumetypes.GeneratedOperations { ... mountVolumeFunc := func() volumetypes.OperationContext { // 匹配mounter plugin volumePlugin, err := og.volumePluginMgr.FindPluginBySpec(volumeToMount.VolumeSpec) ... // pv nodeAffinity匹配 affinityErr := checkNodeAffinity(og, volumeToMount) ... volumeMounter := volumePlugin.NewMounter(volumeToMount.VolumeSpec, volumeToMount.Pod, VolumeOptions{}) ... // plugin mount option检查 checkMountOptionSupport(og, volumeToMount, volumePlugin) ... // attacher attachableVolumePlugin := og.volumePluginMgr.FindAttachablePluginBySpec(volumeToMount.VolumeSpec) if attachableVolumePlugin != nil { attacher := attachableVolumePlugin.NewAttacher() } // deviceMounter deviceMountablePlugin := og.volumePluginMgr.FindDeviceMountablePluginBySpec(volumeToMount.VolumeSpec) if deviceMountableVolumePlugin != nil { volumeDeviceMounter := deviceMountableVolumePlugin.NewDeviceMounter() } ... devicePath := volumeToMount.DevicePath if volumeAttacher != nil { // wait attached devicePath = attacher.WaitForAttach(volumeToMount.VolumeSpec,devicePath, volumeToMount.Pod, timeout) ... } ... if volumeDeviceMounter != nil && asw.GetDeviceMountState(VolumeName) != DeviceGloballyMounted { // /var/lib/kubelet/plugins/kubernetes.io/csi/<driver>/<volumeHandle>/globalmount deviceMountPath := volumeDeviceMounter.GetDeviceMountPath(volumeToMount.VolumeSpec) ... // mount device to global mount path(NodeStageVolume) err = volumeDeviceMounter.MountDevice(volumeToMount.VolumeSpec, devicePath, deviceMountPath, ...) if err != nil { // 更新asw volume状态 og.markDeviceErrorState(volumeToMount, devicePath, deviceMountPath, err, actualStateOfWorld) // On failure, return error. Caller will log and retry. return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } // mark device mounted asw.MarkDeviceAsMounted(volumeToMount.VolumeName, devicePath, deviceMountPath, SELinuxLabel) ... // set staging path for volume expansion resizeOptions.DeviceStagePath = deviceMountPath } if volumeDeviceMounter != nil && resizeOptions.DeviceStagePath == "" { deviceStagePath, err := volumeDeviceMounter.GetDeviceMountPath(volumeToMount.VolumeSpec) ... resizeOptions.DeviceStagePath = deviceStagePath } // Execute mount(NodePublishVolume) // 由deviceMountPath mount至/var/lib/kubelet/pods/<podUID>/volumes/kubernetes.io~csi/<volName>/{mount} volumeMounter.SetUp(volume.MounterArgs{ FsUser: util.FsUserFrom(volumeToMount.Pod), FsGroup: fsGroup, DesiredSize: volumeToMount.DesiredSizeLimit, FSGroupChangePolicy: fsGroupChangePolicy, SELinuxLabel: volumeToMount.SELinuxLabel, }) ... // Update asw if mountErr != nil { // 更新asw volume状态 og.markVolumeErrorState(volumeToMount, markOpts, mountErr, asw) // On failure, return error. Caller will log and retry. return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } ... resizeOptions.DeviceMountPath = volumeMounter.GetPath() // 扩容(resize) _, resizeError = og.expandVolumeDuringMount(volumeToMount, actualStateOfWorld, resizeOptions) if resizeError != nil { // 更新asw volume状态 actualStateOfWorld.MarkVolumeMountAsUncertain(markOpts) ... return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } ... // mark volume mounted actualStateOfWorld.MarkVolumeAsMounted(markOpts) ... return volumetypes.NewOperationContext(nil, nil, migrated) } ... return volumetypes.GeneratedOperations{ OperationName: "volume_mount", OperationFunc: mountVolumeFunc, ... } }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
注意
mountVolumeFunc依次执行attacher.WaitForAttach-->deviceMounter.MountDevice-->volumeMounter.SetUp-->expandVolume
# 2.4.mapVolume
oe.operationGenerator.GenerateMapVolumeFunc()负责生成block device挂载函数，将设备文件及关联设备mount到pod volumes。// After setup is done, create symbolic links on both global map path and pod device map path. Once symbolic // links are created, take fd lock by loopback for the device to avoid silent volume replacement. func (og *operationGenerator) GenerateMapVolumeFunc(...) (volumetypes.GeneratedOperations, error) { // block volume mapper plugin blockPlugin, err := og.volumePluginMgr.FindMapperPluginBySpec(volumeToMount.VolumeSpec) ... // volume pv匹配nodeAffinity checkNodeAffinity(og, volumeToMount) ... // block volume mapper blockVolumeMapper := blockPlugin.NewBlockVolumeMapper(volumeToMount.VolumeSpec, volumeToMount.Pod, ...) ... // block volume attach plugin attachableVolumePlugin, _ := og.volumePluginMgr.FindAttachablePluginBySpec(volumeToMount.VolumeSpec) if attachableVolumePlugin != nil { attacher = attachableVolumePlugin.NewAttacher() } mapVolumeFunc := func() (operationContext volumetypes.OperationContext) { ... // var/lib/kubelet/plugins/kubernetes.io/<plugin>/volumeDevices/<volumeHandle>/dev globalMapPath := blockVolumeMapper.GetGlobalMapPath(volumeToMount.VolumeSpec) ... if volumeAttacher != nil { // Wait for attachable volumes to finish attaching devicePath = attacher.WaitForAttach(volumeToMount.VolumeSpec, volumeToMount.DevicePath, ...) ... } // Call SetUpDevice if blockVolumeMapper implements CustomBlockVolumeMapper if customeMapper, ok := blockVolumeMapper.(volume.CustomBlockVolumeMapper); ok && asw.GetDeviceMountState(volumeToMount.VolumeName) != DeviceGloballyMounted { // block mapper implement custom(NodeStageVolume) // stagingPath:/var/lib/kubelet/plugins/kubernetes.io/csi/volumeDevices/staging/{specName} // 匹配已附加设备-->stating目录创建设备节点 stagingPath, mapErr = customeMapper.SetUpDevice() if mapErr != nil { og.markDeviceErrorState(volumeToMount, devicePath, globalMapPath, mapErr, actualStateOfWorld) // On failure, return error. Caller will log and retry. return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } } // Update asw to reflect volume is globally mounted markedDevicePath := devicePath asw.MarkDeviceAsMounted(volumeToMount.VolumeName, markedDevicePath, globalMapPath, "") ... // Call MapPodDevice if blockVolumeMapper implements CustomBlockVolumeMapper if customBlockVolumeMapper, ok := blockVolumeMapper.(volume.CustomBlockVolumeMapper); ok { // Execute driver specific map(NodePublishVolume) // pluginDevicePath:/var/lib/kubelet/plugins/kubernetes.io/csi/volumeDevices/publish/{specName} // 绑定stating device file或创建设备节点 pluginDevicePath, mapErr := customBlockVolumeMapper.MapPodDevice() if mapErr != nil { // On failure, return error. Caller will log and retry. og.markVolumeErrorState(volumeToMount, markVolumeOpts, mapErr, actualStateOfWorld) return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } // From now on, the volume is mapped. Mark it as uncertain on error, // so it is is unmapped when corresponding pod is deleted. defer func() { if operationContext.EventErr != nil { og.markVolumeErrorState(volumeToMount, markVolumeOpts, errText, asw) } }() // if pluginDevicePath is provided, assume attacher may not provide device // or attachment flow uses SetupDevice to get device path if len(pluginDevicePath) != 0 { devicePath = pluginDevicePath } ... } kvh, ok := og.GetVolumePluginMgr().Host.(volume.KubeletVolumeHost) ... hu := kvh.GetHostUtil() devicePath = hu.EvalHostSymlinks(devicePath) ... // Update asw with the devicePath again, if devicePath has changed from markedDevicePath if markedDevicePath != devicePath { asw.MarkDeviceAsMounted(volumeToMount.VolumeName, devicePath, globalMapPath, "") ... } // var/lib/kubelet/pods/<podUID>/volumeDevices/<plugin>/{volName} volumeMapPath, volName := blockVolumeMapper.GetPodDeviceMapPath() // Execute common map util.MapBlockVolume(og.blkUtil, devicePath, globalMapPath, volumeMapPath, volName,volumeToMount.Pod.UID) ... // Device mapping for pod device map path succeeded resizeError := og.expandVolumeDuringMount(volumeToMount, actualStateOfWorld, resizeOptions) if resizeError != nil { asw.MarkVolumeMountAsUncertain(markVolumeOpts) ... return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } asw.MarkVolumeAsMounted(markVolumeOpts) ... return volumetypes.NewOperationContext(nil, nil, migrated) } ... return volumetypes.GeneratedOperations{ OperationName: "map_volume", OperationFunc: mapVolumeFunc, ... }, nil } // provide a common way of mapping block device path for a specified volume and pod. func MapBlockVolume(...) error { // map devicePath to global node path as bind mount blkUtil.MapDevice(devicePath, globalMapPath, string(podUID), true /* bindMount */) ... // map devicePath to pod volume path blkUtil.MapDevice(devicePath, podVolumeMapPath, volumeMapName, false /* bindMount */) ... // take file descriptor lock to keep a block device opened. blkUtil.AttachFileDevice(filepath.Join(globalMapPath, string(podUID))) ... return nil } // MapDevice creates a symbolic link to block device under specified map path func (v VolumePathHandler) MapDevice(devicePath string, mapPath string, linkName string, bindMount bool) error { ... os.MkdirAll(mapPath, 0750) ... // globalDevice if bindMount { // Check bind mount exists linkPath := filepath.Join(mapPath, string(linkName)) file, err := os.Stat(linkPath) if err != nil { if !os.IsNotExist(err) { return fmt.Errorf("failed to stat file %s: %v", linkPath, err) } // Create file newFile, err := os.OpenFile(linkPath, os.O_CREATE|os.O_RDWR, 0750) ... newFile.Close() ... } else { // Check if device file if file.Mode()&os.ModeDevice == os.ModeDevice { return nil } } ... // bind mount pluginDeviceFile globalDevicePath mounter.MountSensitiveWithoutSystemd(devicePath, linkPath, "" /* fsType */, []string{"bind"}, nil) ... return nil } // podDevice linkPath := filepath.Join(mapPath, string(linkName)) os.Remove(linkPath) ... return os.Symlink(devicePath, linkPath) } // takes a path to a regular file and makes it available as an attached block device. func (v VolumePathHandler) AttachFileDevice(path string) (string, error) { // 先匹配loop device,检查path绑定过没有 blockDevicePath := v.GetLoopDevice(path) ... // If no existing loop device for the path, create one if blockDevicePath == "" { // losetup -f path(激活globalPath设备文件) blockDevicePath = makeLoopDevice(path) ... } return blockDevicePath, 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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
注意
GenerateMapVolumeFunc依次执行waitForAttach-->NodeStageVolume-->NodePublishVolume-->MapBlockVolume-->resize
# 3.umtdevice
# 3.1.umtdevices
rc.unmountDetachDevices()会进一步清理deviceMount/nodeState动作相关的数据，确保volume/device未使用的存储及设备资源释放。func (rc *reconciler) unmountDetachDevices() { // asw unmounted volume for _, attachedVolume := range rc.asw.GetUnmountedVolumes() { // 非预期&非处理中 if !rc.dsw.VolumeExists(...) && !rc.operationExecutor.IsOperationPending(...) { // 触发过deviceMount if attachedVolume.DeviceMayBeMounted() { // Volume is globally mounted to device, unmount it rc.operationExecutor.UnmountDevice(attachedVolume.AttachedVolume, rc.asw, rc.hostutil) ... // 未触发过deviceMount } else { // 清理asw volume,由ad controller执行detach rc.asw.MarkVolumeAsDetached(attachedVolume.VolumeName, attachedVolume.NodeName) } } } } func (oe *operationExecutor) UnmountDevice(...) error { ... // filesystem volume if fsVolume { // unmount and detach device if a volume isn't referenced generatedOperations = oe.operationGenerator.GenerateUnmountDeviceFunc(deviceToDetach, asw, hostutil) // block volume } else { // detach device and remove loopback if a volume isn't referenced generatedOperations = oe.operationGenerator.GenerateUnmapDeviceFunc(deviceToDetach, asw, hostutil) } ... // Avoid executing unmount/unmap device from multiple pods referencing the same volume in parallel podName := nestedpendingoperations.EmptyUniquePodName return oe.pendingOperations.Run(deviceToDetach.VolumeName, podName, "" /* nodeName */, generatedOperations) }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
注意
unmountDetachDevices仅处理执行过deviceMount/deviceMapper的设备
# 3.2.unmountFunc
oe.operationGenerator.GenerateUnmountDeviceFunc()用于将执行过deviceMount volume的全局挂载点清理，确保回收不会泄漏。func (og *operationGenerator) GenerateUnmountDeviceFunc(...) (volumetypes.GeneratedOperations, error) { // deviceMounter plugin deviceMountableVolumePlugin := og.volumePluginMgr.FindDeviceMountablePluginByName(deviceToDetach.PluginName) ... // device unmounter volumeDeviceUnmounter, err := deviceMountableVolumePlugin.NewDeviceUnmounter() ... // device mounter volumeDeviceMounter, err := deviceMountableVolumePlugin.NewDeviceMounter() ... unmountDeviceFunc := func() volumetypes.OperationContext { ... // /var/lib/kubelet/plugins/kubernetes.io/csi/<driver>/<volumeHandle>/globalmount deviceMountPath := volumeDeviceMounter.GetDeviceMountPath(deviceToDetach.VolumeSpec) ... // 获取/proc/self/mountinfo引用 refs, err := deviceMountableVolumePlugin.GetDeviceMountRefs(deviceMountPath) // err或deviceMountPath仍被其它目录引用 if err != nil || util.HasMountRefs(deviceMountPath, refs) { return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } // unmount device and remove global mount path(NodeUnstageVolume) unmountDeviceErr := volumeDeviceUnmounter.UnmountDevice(deviceMountPath) if unmountDeviceErr != nil { // mark the device as uncertain asw.MarkDeviceAsUncertain(deviceToDetach.VolumeName, deviceToDetach.DevicePath, deviceMountPath, ..) ... return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } // device opened检查 deviceOpened, deviceOpenedErr := isDeviceOpened(deviceToDetach, hostutil) ... // The device is still in use elsewhere. Caller will log and retry. if deviceOpened { // Mark the device as uncertain, so MountDevice is called for new pods. asw.MarkDeviceAsUncertain(deviceToDetach.VolumeName, deviceToDetach.DevicePath, deviceMountPath, ..) ... return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } // Update asw actualStateOfWorld.MarkDeviceAsUnmounted(deviceToDetach.VolumeName) ... return volumetypes.NewOperationContext(nil, nil, migrated) } return volumetypes.GeneratedOperations{ OperationName: "unmount_device", OperationFunc: unmountDeviceFunc, ... }, nil } // isDeviceOpened checks the device status if the device is in use anywhere else on the system func isDeviceOpened(deviceToDetach AttachedVolume, hostUtil hostutil.HostUtils) (bool, error) { // device path isDevicePath, devicePathErr := hostUtil.PathIsDevice(deviceToDetach.DevicePath) ... if !isDevicePath && devicePathErr == nil || (devicePathErr != nil && devicePathErr.isNotExist) { deviceOpened = false } else if devicePathErr != nil { return false, deviceToDetach.GenerateErrorDetailed("PathIsDevice failed", devicePathErr) } else { // device open? deviceOpened, deviceOpenedErr = hostUtil.DeviceOpened(deviceToDetach.DevicePath) if deviceOpenedErr != nil { return false, deviceToDetach.GenerateErrorDetailed("DeviceOpened failed", deviceOpenedErr) } } return deviceOpened, nil } // ExclusiveOpenFailsOnDevice is shared with NsEnterMounter func DeviceOpened(pathname string) (bool, error) { ... finfo, err := os.Stat(pathname) ... // path refers to a device if finfo.Mode()&os.ModeDevice != 0 { isDevice = true } if !isDevice { return false, nil } // 独占打开 fd, errno := unix.Open(pathname, unix.O_RDONLY|unix.O_EXCL|unix.O_CLOEXEC, 0) // If the device is in use, open will return an invalid fd. // When this happens, it is expected that Close will fail and throw an error. defer unix.Close(fd) if errno == nil { // device not in use return false, nil } else if errno == unix.EBUSY { // device is in use return true, nil } // error during call to Open return false, errno }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
注意
GenerateUnmountDeviceFunc用于释放deviceMount的存储资源及目录
# 3.3.unmapFunc
oe.operationGenerator.GenerateUnmapDeviceFunc()用于将执行过nodeStete volume的全局设备节点清理，确保回收不会泄漏。// marks device as unmounted based on following steps. func (og *operationGenerator) GenerateUnmapDeviceFunc(...) (volumetypes.GeneratedOperations, error) { ... unmapDeviceFunc := func() volumetypes.OperationContext { // var/lib/kubelet/plugins/kubernetes.io/<plugin>/volumeDevices/<volumeHandle> globalMapPath := deviceToDetach.DeviceMountPath // device bind mount refs refs, err := og.blkUtil.GetDeviceBindMountRefs(deviceToDetach.DevicePath, globalMapPath) ... if len(refs) > 0 { return volumetypes.NewOperationContext(eventErr, detailedErr, migrated) } // Mark device as uncertain to make sure kubelet calls UnmapDevice again. asw.MarkDeviceAsUncertain(deviceToDetach.VolumeName, deviceToDetach.DevicePath, globalMapPath, "") ... // Call TearDownDevice if blockVolumeUnmapper implements CustomBlockVolumeUnmapper if customBlockVolumeUnmapper, ok := blockVolumeUnmapper.(volume.CustomBlockVolumeUnmapper); ok { // Execute tear down device(NodeUnstageVolume) // customBlockVolumeUnmapper.TearDownDevice(globalMapPath, deviceToDetach.DevicePath) ... } // globalMapPath dir and plugin's stored data on the dir are unnecessary, clean up it. og.blkUtil.RemoveMapPath(globalMapPath) ... // check if the path is a device and in use anywhere else on the system. Retry if it returns true. deviceOpened, deviceOpenedErr := isDeviceOpened(deviceToDetach, hostutil) if deviceOpenedErr != nil || deviceOpened { // The device is still in use elsewhere. Caller will log and retry. return volumetypes.NewOperationContext(nil, deviceOpenedErr, migrated) } ... // Update asw asw.MarkDeviceAsUnmounted(deviceToDetach.VolumeName) ... return volumetypes.NewOperationContext(nil, nil, migrated) } return volumetypes.GeneratedOperations{ OperationName: "unmap_device", OperationFunc: unmapDeviceFunc, ... }, nil } // GetDeviceBindMountRefs searches bind mounts under global map path func (v VolumePathHandler) GetDeviceBindMountRefs(devPath string, mapPath string) ([]string, error) { ... files, err := ioutil.ReadDir(mapPath) ... for _, file := range files { if file.Mode()&os.ModeDevice != os.ModeDevice { continue } filename := file.Name() refs = append(refs, filepath.Join(mapPath, filename)) } return refs, nil } // TearDownDevice removes traces of the SetUpDevice. func (m *csiBlockMapper) TearDownDevice(globalMapPath, devicePath string) error { ... // csi.NodeUnstageVolume // stagingPath:/var/lib/kubelet/plugins/kubernetes.io/csi/volumeDevices/staging/{specName} // 释放设备文件,删除stagingPath目录 m.unstageVolumeForBlock(ctx, csiClient, stagingPath) ... // os.Remove(publishDir) // os.Remove(stagingPath) // os.Remove(globalMapPath) m.cleanupOrphanDeviceFiles() ... 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
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
注意
GenerateUnmapDeviceFunc用于释放nodeState的设备文件及目录，回收publish/globalMap相关目录资源
# 3.4.orphanVolume
rc.cleanOrphanVolumes()会检测dsw未注册，asw重建失败的volume，根据存储类型进行相关volume回收，以避免存储泄漏。// cleanOrphanVolumes tries to clean up all volumes that failed reconstruction. func (rc *reconciler) cleanOrphanVolumes() { // 没有失败重建的 if len(rc.volumesFailedReconstruction) == 0 { return } for _, volume := range rc.volumesFailedReconstruction { // dsw注册过 if rc.dsw.VolumeExistsWithSpecName(volume.podName, volume.volumeSpecName) { // Some pod needs the volume, don't clean it up. continue } // 清理orphan volume rc.cleanupMounts(volume) } // Clean the cache, cleanup is one shot operation. rc.volumesFailedReconstruction = make([]podVolume, 0) } func (rc *reconciler) cleanupMounts(volume podVolume) { ... // TODO: will add to unmount both volume and device in the same routine. rc.operationExecutor.UnmountVolume(mountedVolume, rc.asw, rc.kubeletPodsDir) ... } func (oe *operationExecutor) UnmountVolume(...) error { // Filesystem volume if fsVolume { // Unmount a volume if a volume is mounted // subpath.CleanSubPaths // csi.NodeUnpublishVolume generatedOperations = oe.operationGenerator.GenerateUnmountVolumeFunc(volumeToUnmount, asw, podsDir) // Block volume } else { // Unmap a volume if a volume is mapped // release device file // unmap pod device // unmap global device generatedOperations = oe.operationGenerator.GenerateUnmapVolumeFunc(volumeToUnmount, asw) } ... // All volume plugins can execute unmount/unmap for multiple pods referencing the same volume in parallel podName := volumetypes.UniquePodName(volumeToUnmount.PodUID) return oe.pendingOperations.Run(volumeToUnmount.VolumeName, podName, "" /* nodeName */, generatedOperations) }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
注意
cleanOrphanVolumes检查重建失败的volume，触发相关存储资源回收