volumeManager
南风未起 2026-03-21 19:39:22 csi
# 1.pluginMgr
# 1.1.registry
kubelet会执行UnsecuredDependencies()注册volumePlugin，加载in-tree、csi及flex插件及激活目录事件监听，实现动态加载。// returns a Dependencies suitable for being run, or an error if the server setup is not valid. func UnsecuredDependencies(...) (*kubelet.Dependencies, error) { ... // mount相关 mounter := mount.New(s.ExperimentalMounterPath) subpather := subpath.New(mounter) hu := hostutil.NewHostUtil() // flex plugin执行相关 pluginRunner := exec.New() // 加载in-tree和csi插件 plugins, err := ProbeVolumePlugins(featureGate) ... return &kubelet.Dependencies{ ... HostUtil: hu, Mounter: mounter, Subpather: subpather, OOMAdjuster: oom.NewOOMAdjuster(), OSInterface: kubecontainer.RealOS{}, VolumePlugins: plugins, DynamicPluginProber: GetDynamicPluginProber(s.VolumePluginDir, pluginRunner), // flex plugin ... } // ProbeVolumePlugins collects all volume plugins into an easy to use list. func ProbeVolumePlugins(featureGate featuregate.FeatureGate) ([]volume.VolumePlugin, error) { ... // 云厂商相关Plugin及迁移检查 allPlugins, err = appendLegacyProviderVolumes(allPlugins, featureGate) ... // in-tree plugin allPlugins = append(allPlugins, emptydir.ProbeVolumePlugins()...) allPlugins = append(allPlugins, git_repo.ProbeVolumePlugins()...) allPlugins = append(allPlugins, hostpath.ProbeVolumePlugins(volume.VolumeConfig{})...) allPlugins = append(allPlugins, nfs.ProbeVolumePlugins(volume.VolumeConfig{})...) allPlugins = append(allPlugins, secret.ProbeVolumePlugins()...) allPlugins = append(allPlugins, iscsi.ProbeVolumePlugins()...) allPlugins = append(allPlugins, cephfs.ProbeVolumePlugins()...) allPlugins = append(allPlugins, downwardapi.ProbeVolumePlugins()...) allPlugins = append(allPlugins, fc.ProbeVolumePlugins()...) allPlugins = append(allPlugins, configmap.ProbeVolumePlugins()...) allPlugins = append(allPlugins, projected.ProbeVolumePlugins()...) allPlugins = append(allPlugins, local.ProbeVolumePlugins()...) // csi plugin allPlugins = append(allPlugins, csi.ProbeVolumePlugins()...) return allPlugins, nil } // gets the probers of dynamically discoverable plugins for kubelet. func GetDynamicPluginProber(pluginDir string, runner exec.Interface) volume.DynamicPluginProber { return flexvolume.GetDynamicPluginProber(pluginDir, runner) } // GetDynamicPluginProber creates dynamic plugin prober func GetDynamicPluginProber(pluginDir string, runner exec.Interface) volume.DynamicPluginProber { return &flexVolumeProber{ pluginDir: pluginDir, // /usr/libexec/kubernetes/kubelet-plugins/volume/exec/ watcher: utilfs.NewFsnotifyWatcher(), // 文件目录事件监听 factory: pluginFactory{}, // 加载的flex plugin会放到这个工厂 runner: runner, // 执行模块 fs: &utilfs.DefaultFs{}, // 文件操作模块 } }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
注意
flex plugin基于watcher监听目录感知，csi加载则基于pluginMgr监听后注册过来，后面会介绍
# 1.2.initialize
NewMainKubelet()会执行NewInitializedVolumePluginMgr()加载pluginMgr，初始化plugin及激活flex prober监听动态插件。// returns volume.VolumePluginMgr initialized with kubelets implementation of the volume.VolumeHost interface. func NewInitializedVolumePluginMgr(...) (*volume.VolumePluginMgr, error) { ... if kubelet.kubeClient != nil { ... // csiDriver缓存 csiDriverInformer := informerFactory.Storage().V1().CSIDrivers() csiDriverLister = csiDriverInformer.Lister() ... } // 初始化volumeHost kvh := &kubeletVolumeHost{ kubelet: kubelet, volumePluginMgr: volume.VolumePluginMgr{}, secretManager: secretManager, configMapManager: configMapManager, tokenManager: tokenManager, informerFactory: informerFactory, csiDriverLister: csiDriverLister, ... exec: utilexec.New(), } // 初始化pluginMgr的插件 kvh.volumePluginMgr.InitPlugins(plugins, prober, kvh) ... return &kvh.volumePluginMgr, nil } // initializes each plugin. All plugins must have unique names. func (pm *VolumePluginMgr) InitPlugins(...) error { ... pm.Host = host // flex prober if prober == nil { pm.prober = &dummyPluginProber{} } else { pm.prober = prober } // 激活prober探测 if err := pm.prober.Init(); err != nil { // 重置为空实现 pm.prober = &dummyPluginProber{} } ... for _, plugin := range plugins { name := plugin.GetPluginName() // 名称非法 if !validation.IsQualifiedName(name) { continue } // plugin加载过 if _, found := pm.plugins[name]; found { continue } // 初始化plugin plugin.Init(host) ... pm.plugins[name] = plugin } return utilerrors.NewAggregate(allErrs) } // in-tree plugin only set volumeHost func (plugin *cephfsPlugin) Init(host volume.VolumeHost) error { plugin.host = host 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
注意
plugin.Init()会初始化in-tree/csi plugin，csi会进一步加载csiClient、informer lister及创建csiNode
# 1.3.prober
prober本质是flex plugin，prober.Init()会初始化watcher监听器，动态感知及加载flex目录下的可执行插件，后续基于exec执行。func (prober *flexVolumeProber) Init() error { // 标记全量加载 prober.testAndSetProbeAllNeeded(true) prober.eventsMap = map[string]volume.ProbeOperation{} // flex plugin目录(/usr/libexec/kubernetes/kubelet-plugins/volume/exec) prober.createPluginDir() ... // 激活fswatcher prober.initWatcher() ... return nil } // Creates a new filesystem watcher and adds watches for the plugin directory and all of its subdirectories. func (prober *flexVolumeProber) initWatcher() error { // 注册回调 prober.watcher.Init(func(event fsnotify.Event) { prober.handleWatchEvent(event) ... }, func(err error) { klog.Errorf("Received an error from watcher: %s", err) }) ... // 递归注册目录watcher prober.addWatchRecursive(prober.pluginDir) ... // 启动watcher prober.watcher.Run() return nil } /usr/libexec/kubernetes/kubelet-plugins/volume/exec/ └── example.com~fastdisk/ └── fastdisk (可执行文件) // 假设flex plugin权限调整 func (prober *flexVolumeProber) handleWatchEvent(event fsnotify.Event) error { // Ignore files beginning with '.' if filepath.Base(event.Name)[0] == '.' { return nil } // /usr/libexec/kubernetes/kubelet-plugins/volume/exec/example.com~fastdisk/fastdisk eventPathAbs, err := filepath.Abs(event.Name) ... // /usr/libexec/kubernetes/kubelet-plugins/volume/exec/example.com~fastdisk parentPathAbs := filepath.Dir(eventPathAbs) // /usr/libexec/kubernetes/kubelet-plugins/volume/exec pluginDirAbs, err := filepath.Abs(prober.pluginDir) ... // plugin目录 if eventPathAbs == pluginDirAbs { // flexvolume plugin directory is removed if event.Has(fsnotify.Remove) { // 重建 prober.createPluginDir() ... // 重新注册 prober.addWatchRecursive(pluginDirAbs) ... } return nil } // watch newly added subdirectories inside a driver directory if event.Has(fsnotify.Create) { // 注册新目录watcher prober.addWatchRecursive(eventPathAbs) ... } // example.com~fastdisk/fastdisk eventRelPath := filepath.Rel(pluginDirAbs, eventPathAbs) ... // event inside specific driver dir if len(eventRelPath) > 0 { // example.com~fastdisk driverDirName := strings.Split(eventRelPath, string(os.PathSeparator))[0] // /usr/libexec/kubernetes/kubelet-plugins/volume/exec/example.com~fastdisk driverDirAbs := filepath.Join(pluginDirAbs, driverDirName) // plugin删除或plugin目录删除 if event.Remove && (eventRelPath == executablePathRel(driverDirName) || parentPathAbs == pluginDirAbs) { // 标记driver移除 prober.updateEventsMap(driverDirAbs, volume.ProbeRemove) } else { // 标记driver注册 prober.updateEventsMap(driverDirAbs, volume.ProbeAddOrUpdate) } } return nil } func (w *fsnotifyWatcher) Run() { go func() { defer w.watcher.Close() for { select { // watch到事件 case event := <-w.watcher.Events: if w.eventHandler != nil { // 触发handleWatchEvent w.eventHandler(event) } ... } } }() }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
注意
flex prober会持续监听/usr/libexec/kubernetes/kubelet-plugins/volume/exec，插件的注册或移除均会感知及缓存
# 1.4.csiplugin
csiPlugin作为所有csi插件的交互入口，csidriver基于sidecar注册到kubelet会加载到csiPlugin供volumeMgr调用。func (p *csiPlugin) Init(host volume.VolumeHost) error { p.host = host csiClient := host.GetKubeClient() if csiClient != nil { ... // kubelet host实现 kletHost, ok := host.(volume.KubeletVolumeHost) if ok { // 加载csidriver Lister p.csiDriverLister = kletHost.CSIDriverLister() ... // sa token获取 p.serviceAccountTokenGetter = host.GetServiceAccountTokenFunc() ... // We don't run the volumeAttachmentLister in the kubelet context p.volumeAttachmentLister = nil } } ... // Initializing the label management channels nim = nodeinfomanager.NewNodeInfoManager(host.GetNodeName(), host, migratedPlugins) // This function prevents Kubelet from posting Ready status until CSINode is both installed and initialized initializeCSINode(host) ... return nil } func initializeCSINode(host volume.VolumeHost) error { kvh, ok := host.(volume.KubeletVolumeHost) ... kubeClient := host.GetKubeClient() ... go func() { ... // 获取nodeName nodeName := host.GetNodeName() // 持续获取csiNode直至成功 waitForAPIServerForever(kubeClient, nodeName) ... // 尝试6次{90ms,540ms,3.2s,19s,114s} wait.ExponentialBackoff(initBackoff, func() (bool, error) { // csiNode创建 err := nim.InitializeCSINodeWithAnnotation() ... // Successfully initialized drivers, allow Kubelet to post Ready kvh.SetKubeletError(nil) return true, nil }) ... }() return nil } func (nim *nodeInfoManager) InitializeCSINodeWithAnnotation() error { csiKubeClient := nim.volumeHost.GetKubeClient() ... // 尝试4次{10ms,50ms,250ms,1.25s} err := wait.ExponentialBackoff(updateBackoff, func() (bool, error) { nim.tryInitializeCSINodeWithAnnotation(csiKubeClient) ... return true, nil }) ... return nil } func (nim *nodeInfoManager) tryInitializeCSINodeWithAnnotation(csiKubeClient clientset.Interface) error { // 获取csiNode nodeInfo, err := csiKubeClient.StorageV1().CSINodes().Get(context.TODO(), nim.nodeName, metav1.GetOptions{}) // 未找到 if nodeInfo == nil || errors.IsNotFound(err) { // createCSINode will set the annotation return nim.CreateCSINode() } ... // 更新csiNode注解(注入迁移plugin) annotationModified := setMigrationAnnotation(nim.migratedPlugins, nodeInfo) if annotationModified { return csiKubeClient.StorageV1().CSINodes().Update(context.TODO(), nodeInfo, metav1.UpdateOptions{}) } 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
90
91
92
93
94
95
96
97
98
99
100
101
102
注意
csiPlugin加载会初始化csiNode，后续注册过来的csi plugin会更新到csiNode
# 2.volumeMgr
# 2.1.initialize
NewMainKubelet()调用NewVolumeManager()进行初始化，核心组件是dswp和reconciler，负责dsw缓存生成及预期卷挂载卸载。// returns a new concrete instance implementing the VolumeManager interface. func NewVolumeManager(...) VolumeManager { vm := &volumeManager{ kubeClient: kubeClient, volumePluginMgr: volumePluginMgr, desiredStateOfWorld: cache.NewDesiredStateOfWorld(volumePluginMgr, util.NewSELinuxLabelTranslator()), actualStateOfWorld: cache.NewActualStateOfWorld(nodeName, volumePluginMgr), operationExecutor: operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator( kubeClient, volumePluginMgr, recorder, blockVolumePathHandler)), } // plugin迁移检查 vm.intreeToCSITranslator := csitrans.New() vm.csiMigratedPluginManager := csimigration.NewPluginManager(intreeToCSITranslator, DefaultFeatureGate) // dswp同步模块 vm.desiredStateOfWorldPopulator = populator.NewDesiredStateOfWorldPopulator( kubeClient, desiredStateOfWorldPopulatorLoopSleepPeriod, podManager, podStateProvider, vm.desiredStateOfWorld, vm.actualStateOfWorld, kubeContainerRuntime, keepTerminatedPodVolumes, csiMigratedPluginManager, intreeToCSITranslator, volumePluginMgr) // 协调模块 vm.reconciler = reconciler.NewReconciler( kubeClient, controllerAttachDetachEnabled, reconcilerLoopSleepPeriod, waitForAttachTimeout, nodeName, vm.desiredStateOfWorld, vm.actualStateOfWorld, vm.desiredStateOfWorldPopulator.HasAddedPods, vm.operationExecutor, mounter, hostutil, volumePluginMgr, kubeletPodsDir) return vm }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
注意
dswp主要用于同步scheduled pod使用的volume，更新dsw维护的预期状态
# 2.2.runworker
kubelet会执行volumeMgr.Run()唤醒同步协程，调用dswp.Run()生成及填充dsw状态，调用reconciler.Run()协调volume。func (vm *volumeManager) Run(sourcesReady config.SourcesReady, stopCh <-chan struct{}) { // start informer for CSIDriver go vm.volumePluginMgr.Run(stopCh) // dswp同步 go vm.desiredStateOfWorldPopulator.Run(sourcesReady, stopCh) // 基于dsw协调 go vm.reconciler.Run(stopCh) ... <-stopCh } // dswp-->dsw func (dswp *desiredStateOfWorldPopulator) Run(sourcesReady config.SourcesReady, stopCh <-chan struct{}) { // 间隔100ms触发一次直到sourceReady wait.PollUntil(dswp.loopSleepDuration, func() (bool, error) { // 同步state至dsw dswp.populatorLoop() return sourcesReady.AllReady(), nil }, stopCh) ... // 标记初始化完成 if !dswp.hasAddedPods { dswp.hasAddedPods = true } ... // 间隔100ms触发一次 wait.Until(dswp.populatorLoop, dswp.loopSleepDuration, stopCh) } func (dswp *desiredStateOfWorldPopulator) populatorLoop() { // 补充新的pod volume dswp.findAndAddNewPods() // 删除旧的pod volume dswp.findAndRemoveDeletedPods() } // dsw-->asw func (rc *reconciler) Run(stopCh <-chan struct{}) { rc.reconstructVolumes() // 间隔100ms触发一次 wait.Until(rc.reconcileNew, rc.loopSleepDuration, stopCh) }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
注意
reconcile后续会分析reconcileNew实现，reconcileOld后续已废弃，这里不再赘述
# 2.3.findAddPod
dswp.findAndAddNewPods()会加载asw mounted volume，基于podManager获取的Pod将volume注册到dsw维护的预期缓存。// Iterate through all pods and add to desired state of world if they don't exist but should func (dswp *desiredStateOfWorldPopulator) findAndAddNewPods() { ... // asw mounted volume for _, mountedVolume := range dswp.actualStateOfWorld.GetMountedVolumes() { // 还未处理 mountedVolumes, exist := mountedVolumesForPod[mountedVolume.PodName] if !exist { ... // 缓存 outerName-->podName-->volume mountedVolumesForPod[mountedVolume.PodName] = mountedVolumes } // 注册到mountedVolumes mountedVolumes[mountedVolume.OuterVolumeSpecName] = mountedVolume } // 加载podManager缓存的pod for _, pod := range dswp.podManager.GetPods() { // 正常运行&pod正在终止(不会再运行container) if dswp.hasAddedPods && dswp.podStateProvider.ShouldPodContainersBeTerminating(pod.UID) { continue } // 重启&pod确定删除(没有运行的container) if !dswp.hasAddedPods && dswp.podStateProvider.ShouldPodRuntimeBeRemoved(pod.UID) { continue } // 更新dsw dswp.processPodVolumes(pod, mountedVolumesForPod) } } // processes the volumes in the given pod and adds them to the dsw. func (dswp *desiredStateOfWorldPopulator) processPodVolumes(...) { if pod == nil { return } // pod volume处理过 if dswp.podPreviouslyProcessed(util.GetUniquePodName(pod)) { return } allVolumesAdded := true // 获取pod container volume(volumeMount及volumeDevice) mounts, devices, seLinuxContainerContexts := util.GetPodVolumeNames(pod) // Process volume spec for each volume defined in pod for _, podVolume := range pod.Spec.Volumes { // container未使用的volume/device if !mounts.Has(podVolume.Name) && !devices.Has(podVolume.Name) { // Volume is not used in the pod, ignore it. continue } // pod关联的pvc,pv及guid pvc, volumeSpec, volumeGidValue, err := dswp.createVolumeSpec(podVolume, pod, mounts, devices) if err != nil { // 记录pod err(最大10个err) dswp.desiredStateOfWorld.AddErrorToPod(uniquePodName, err.Error()) // 标记volume未注册 allVolumesAdded = false continue } // pod volume注册到dsw uniqueVolumeName, err := dswp.dsw.AddPodToVolume(uniquePodName, pod, volumeSpec, podVolume.Name, volumeGidValue, seLinuxContainerContexts[podVolume.Name]) if err != nil { // 记录pod err(最大10个err) dswp.dsw.AddErrorToPod(uniquePodName, err.Error()) // 标记volume未注册 allVolumesAdded = false } ... // dsw/asw volumeSize修正 dswp.checkVolumeFSResize(pod, podVolume, pvc, volumeSpec, uniquePodName, mountedVolumesForPod) } // pod volume注册完成 if allVolumesAdded { // 标记pod已处理 dswp.markPodProcessed(uniquePodName) // 标记asw volume重挂载 dswp.asw.MarkRemountRequired(uniquePodName) // 清理dsw错误 dswp.dsw.PopPodErrors(uniquePodName) // pod volume之前成功过 } else if dswp.podHasBeenSeenOnce(uniquePodName) { // 标记pod已处理 dswp.markPodProcessed(uniquePodName) } }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
注意
findAndAddNewPods会将未处理的pod volume注册到dsw，调整dsw缓存的volume size
# 2.4.volumeSpec
dswp.createVolumeSpec()会解析pod volume格式，基于关联的PV/PVC将volume转为volumeMgr内部可识别的表示，用于后续协调处理。// creates and returns a mutable volume.Spec object for the specified volume. func (dswp *desiredStateOfWorldPopulator) createVolumeSpec(...) (pvc, *volume.Spec, string, error) { pvcSource := podVolume.VolumeSource.PersistentVolumeClaim // ephemeral volume isEphemeral := pvcSource == nil && podVolume.VolumeSource.Ephemeral != nil if isEphemeral { // pvc由ephemeral controller创建 pvcSource = &v1.PersistentVolumeClaimVolumeSource{ // podName+volumeName ClaimName: ephemeral.VolumeClaimName(pod, &podVolume), } } // pod关联PVC if pvcSource != nil { // 获取PVC(!deletion+bound) pvc, err := dswp.getPVCExtractPV(pod.Namespace, pvcSource.ClaimName) ... // ephemeral volume if isEphemeral { // volume ownedPod ephemeral.VolumeIsForPod(pod, pvc) ... } pvName, pvcUID := pvc.Spec.VolumeName, pvc.UID // 获取bound PV,生成volumeSpec volumeSpec, volumeGidValue, err := dswp.getPVSpec(pvName, pvcSource.ReadOnly, pvcUID) ... // 匹配in-tree plugin,检查migratable migratable := dswp.csiMigratedPluginManager.IsMigratable(volumeSpec) ... // in-tree迁移 if migratable { // 转为csi volumeSpec volumeSpec = csimigration.TranslateInTreeSpecToCSI(volumeSpec, pod.Namespace, intreeToCSITranslator) ... } volumeMode := *volumeSpec.PersistentVolume.Spec.VolumeMode(v1.PersistentVolumeFilesystem) // container has volumeMounts but the volumeMode of PVC isn't Filesystem. if mounts.Has(podVolume.Name) && volumeMode != v1.PersistentVolumeFilesystem { return nil, nil, "", fmt.Errorf(...) } // container has volumeDevices but the volumeMode of PVC isn't Block if devices.Has(podVolume.Name) && volumeMode != v1.PersistentVolumeBlock { return nil, nil, "", fmt.Errorf(...) } return pvc, volumeSpec, volumeGidValue, nil } ... // Do not return the original volume object, since the source could mutate it spec := volume.NewSpecFromVolume(podVolume.DeepCopy()) // 匹配in-tree plugin,检查migratable if dswp.csiMigratedPluginManager.IsMigratable(spec) { // 转为csi volumeSpec spec, err = csimigration.TranslateInTreeSpecToCSI(spec, pod.Namespace, dswp.intreeToCSITranslator) ... } return nil, spec, "", 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
注意
createVolumeSpec会获取相关PVC/PV，进而生成volumeSpec
# 2.5.addVolume
dswp.dsw.AddPodToVolume()负责将pod volume注册至dsw，登记待mount的pod volume，以对比asw mounted volume执行协调。func (dsw *desiredStateOfWorld) AddPodToVolume(...) (v1.UniqueVolumeName, error) { ... // 匹配plugin volumePlugin, err := dsw.volumePluginMgr.FindPluginBySpec(volumeSpec) ... // attachable attachable := util.IsAttachableVolume(volumeSpec, dsw.volumePluginMgr) // device-mountable deviceMountable := util.IsDeviceMountableVolume(volumeSpec, dsw.volumePluginMgr) // attachable或device-mountable if attachable || deviceMountable { // volumePlugin.GetVolumeName volumeName, err = util.GetUniqueVolumeNameFromSpec(volumePlugin, volumeSpec) ... } else { // pluginName/podName-volumeSpecName volumeName = util.GetUniqueVolumeNameFromSpecWithPod(podName, volumePlugin, volumeSpec) } // plugin支持SELinuxMount,获取seLinuxFileLabel seLinuxFileLabel, supportsSELinuxContextMount := dsw.getSELinuxLabel(volumeSpec, seLinuxContainerContexts) ... // dsw未注册过pod volume if vol, volumeExists := dsw.volumesToMount[volumeName]; !volumeExists { ... // volume来源 if volumeSpec.Volume != nil { // emptyDir+mediumStorage||configmap if util.IsLocalEphemeralVolume(*volumeSpec.Volume) { // 统计container+overhead资源 podLimits := resourcehelper.PodLimits(pod, resourcehelper.PodResourcesOptions{}) // ephemeral-storage limit ephemeralStorageLimit := podLimits[v1.ResourceEphemeralStorage] sizeLimit = resource.NewQuantity(ephemeralStorageLimit.Value(), resource.BinarySI) // ephemeral-storage limit修正 if volumeSpec.Volume.EmptyDir != nil && volumeSpec.Volume.EmptyDir.SizeLimit != nil && volumeSpec.Volume.EmptyDir.SizeLimit.Value() > 0 && (sizeLimit.Value() == 0 || volumeSpec.Volume.EmptyDir.SizeLimit.Value() < sizeLimit.Value()){ sizeLimit = resource.NewQuantity(volumeSpec.Volume.EmptyDir.SizeLimit.Value(), resource.BinarySI) } } } effectiveSELinuxMountLabel := seLinuxFileLabel if !util.VolumeSupportsSELinuxMount(volumeSpec) { // Clear SELinux label for the volume with unsupported access modes. effectiveSELinuxMountLabel = "" } ... // 初始化vmt对象 vmt := volumeToMount{ volumeName: volumeName, podsToMount: make(map[types.UniquePodName]podToMount), pluginIsAttachable: attachable, pluginIsDeviceMountable: deviceMountable, volumeGidValue: volumeGidValue, reportedInUse: false, desiredSizeLimit: sizeLimit, effectiveSELinuxMountFileLabel: effectiveSELinuxMountLabel, originalSELinuxLabel: seLinuxFileLabel, } // pv来源 if volumeSpec.PersistentVolume != nil { // 设置容量 pvCap := volumeSpec.PersistentVolume.Spec.Capacity.Storage() if pvCap != nil { vmt.persistentVolumeSize = &pvCap.DeepCopy() } } // 注册至dsw dsw.volumesToMount[volumeName] = vmt // dsw已注册volume } else { // 支持seLinuxMount if pluginSupportsSELinuxContextMount { // seLinuxFileLabel调整,更新metric if seLinuxFileLabel != vol.originalSELinuxLabel { supported := util.VolumeSupportsSELinuxMount(volumeSpec) handleSELinuxMetricError(fullErr, supported, misMatchWarnings, mismatchErrors) ... } } } // dsw volume oldPodMount, ok := dsw.volumesToMount[volumeName].podsToMount[podName] mountRequestTime := time.Now() // plugin不支持remount,不更新mountRequestTime if ok && !volumePlugin.RequiresRemount(volumeSpec) { mountRequestTime = oldPodMount.mountRequestTime } // 更新dsw volume dsw.volumesToMount[volumeName].podsToMount[podName] = podToMount{ podName: podName, pod: pod, volumeSpec: volumeSpec, outerVolumeSpecName: outerVolumeSpecName, mountRequestTime: mountRequestTime, } return volumeName, 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
注意
AddPodToVolume基于PVC/Pod/VolumeSpec计算属性相关信息，实例化volumeToMount注册到dsw
# 2.6.findDelPod
dswp.findAndRemoveDeletedPods()会遍历dsw维护的预期pod volume，由dsw缓存中清理podManager未找到的pod volume。// Iterate through all pods in dsw, and remove if they no longer exist. func (dswp *desiredStateOfWorldPopulator) findAndRemoveDeletedPods() { for _, volumeToMount := range dswp.desiredStateOfWorld.GetVolumesToMount() { // 匹配podManager的Pod pod, podExists := dswp.podManager.GetPodByUID(volumeToMount.Pod.UID) if podExists { // attachable if volumeToMount.PluginIsAttachable { attachablePlugin := dswp.volumePluginMgr.FindAttachablePluginBySpec(volumeToMount.VolumeSpec) // non-attachable if attachablePlugin == nil { // 更新dsw volume non-attachable dswp.desiredStateOfWorld.MarkVolumeAttachability(volumeToMount.VolumeName, false) continue } } // Exclude known pods that we expect to be running if !dswp.podStateProvider.ShouldPodRuntimeBeRemoved(pod.UID) { continue } // 保留terminated pod volume if dswp.keepTerminatedPodVolumes { continue } } // Exclude known pods that we expect to be running if !dswp.podStateProvider.ShouldPodRuntimeBeRemoved(volumeToMount.Pod.UID) { continue } ... // 检查asw volume相关挂载信息 removed := dswp.asw.PodRemovedFromVolume(volumeToMount.PodName, volumeToMount.VolumeName) // asw volume已清理(unmount完成)+podManager未清理Pod if removed && podExists { continue } // 清理dsw volume信息 dswp.desiredStateOfWorld.DeletePodFromVolume(volumeToMount.PodName, volumeToMount.VolumeName) // 重置处理标记 dswp.deleteProcessedPod(volumeToMount.PodName) } // 弹出pod相关err podsWithError := dswp.desiredStateOfWorld.GetPodsWithErrors() for _, podName := range podsWithError { if _, podExists := dswp.podManager.GetPodByUID(types.UID(podName)); !podExists { dswp.desiredStateOfWorld.PopPodErrors(podName) } } }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
补充
findAndRemoveDeletedPods基于dsw和podManager相关的pod清理dsw缓存
# 3.reconciler
# 3.1.reconstructs
rc.reconstructVolumes()负责加载磁盘已挂载的pod volume，基于asw状态触发volume重建及注册reconstructedVolume至asw。// tries to reconstruct the asw by scanning all pods' volume directories from the disk. func (rc *reconciler) reconstructVolumes() { // 加载/var/lib/kubelet/pods/<pod_uid>/<volumes/volumeDevices>/{escapeQualifiedPluginName}/{volumeName} podVolumes, err := getVolumesFromPodDir(rc.kubeletPodsDir) ... for _, volume := range podVolumes { // pod volume已注册至asw if rc.actualStateOfWorld.VolumeExistsWithSpecName(volume.podName, volume.volumeSpecName) { // There is nothing to reconstruct continue } // 重建volume reconstructedVolume, err := rc.reconstructVolume(volume) if err != nil { // Remember to check DSW after it's fully populated and force unmount the volume when it's orphaned. rc.volumesFailedReconstruction = append(rc.volumesFailedReconstruction, volume) continue } gvl := &globalVolumeInfo{ volumeName: reconstructedVolume.volumeName, volumeSpec: reconstructedVolume.volumeSpec, devicePath: reconstructedVolume.devicePath, deviceMounter: reconstructedVolume.deviceMounter, blockVolumeMapper: reconstructedVolume.blockVolumeMapper, mounter: reconstructedVolume.mounter, } // 初始化过gvl(确保globalVolumeInfo唯一) if cachedInfo, ok := reconstructedVolumes[reconstructedVolume.volumeName]; ok { gvl = cachedInfo } gvl.addPodVolume(reconstructedVolume) // 更新重建的volumeName reconstructedVolumeNames = append(reconstructedVolumeNames, reconstructedVolume.volumeName) // 缓存gvl reconstructedVolumes[reconstructedVolume.volumeName] = gvl } // 重建成功过 if len(reconstructedVolumes) > 0 { // Add the volumes to ASW rc.updateStatesNew(reconstructedVolumes) // Remember to update DSW with this information. rc.volumesNeedReportedInUse = reconstructedVolumeNames // Remember to update devicePath from node.status.volumesAttached rc.volumesNeedUpdateFromNodeStatus = reconstructedVolumeNames } } func (rc *reconciler) updateStatesNew(reconstructedVolumes map[v1.UniqueVolumeName]*globalVolumeInfo) { for _, gvl := range reconstructedVolumes { // volume注册到asw attachedVolume,状态为uncertain attachable rc.asw.AddAttachUncertainReconstructedVolume(gvl.volumeName,gvl.volumeSpec, rc.nodeName, gvl.devicePath) ... for _, volume := range gvl.podVolumes { markVolumeOpts := operationexecutor.MarkVolumeOpts{ PodName: volume.podName, PodUID: types.UID(volume.podName), VolumeName: volume.volumeName, Mounter: volume.mounter, BlockVolumeMapper: volume.blockVolumeMapper, OuterVolumeSpecName: volume.outerVolumeSpecName, VolumeGidVolume: volume.volumeGidValue, VolumeSpec: volume.volumeSpec, VolumeMountState: operationexecutor.VolumeMountUncertain, SELinuxMountContext: volume.seLinuxMountContext, } // volume注册到asw mountedVolume,状态为uncertain mount rc.asw.CheckAndMarkVolumeAsUncertainViaReconstruction(markVolumeOpts) ... seLinuxMountContext = volume.seLinuxMountContext } // If the volume has device to mount, we mark its device as uncertain. if gvl.deviceMounter != nil || gvl.blockVolumeMapper != nil { // deviceMount:/var/lib/kubelet/plugins/kubernetes.io/<plugin-name>/<volume-id>/globalmount // deviceMap:/var/lib/kubelet/plugins/kubernetes.io/<plugin-name>/<volume-id>/dev // attached device,若以filesystem暴露给pod,走deviceMounter-->mounter // attached device,若以device暴露给pod,走deviceMapper-->mounter deviceMountPath := getDeviceMountPath(gvl) ... // device volume注册至asw attachedVolumes,状态为uncertain DeviceMount rc.asw.MarkDeviceAsUncertain(gvl.volumeName, gvl.devicePath, deviceMountPath, seLinuxMountContext) ... } } }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
注意
rc.reconstructVolumes用于加载磁盘上的pod volume，触发volume重建及注册到asw
# 3.2.reconstruct
rc.reconstructVolume()执行真正的volume重建，基于pod volume匹配plugin，调用plugin.ConstructVolume生成volumeSpec。// ReconstructVolumeOperation return a func to create volumeSpec from mount path func (oe *operationExecutor) ReconstructVolumeOperation(...) (volume.ReconstructedVolume, error) { // filesystem Volume if volumeMode == v1.PersistentVolumeFilesystem { // Create volumeSpec from mount path reconstructed, err := plugin.ConstructVolumeSpec(volumeSpecName, volumePath) ... return reconstructed, nil } // block Volume volumeSpec, err := mapperPlugin.ConstructBlockVolumeSpec(uid, volumeSpecName, volumePath) ... return volume.ReconstructedVolume{Spec: volumeSpec}, nil } // Reconstruct volume data structure by reading the pod's volume directories func (rc *reconciler) reconstructVolume(volume podVolume) (rvolume *reconstructedVolume, rerr error) { ... // 匹配plugin plugin, err := rc.volumePluginMgr.FindPluginByName(volume.pluginName) ... // Create pod object pod := &v1.Pod{ ObjectMeta: metav1.ObjectMeta{ UID: types.UID(volume.podName), }, } // 匹配blockMapper plugin mapperPlugin, err := rc.volumePluginMgr.FindMapperPluginByName(volume.pluginName) ... // blockVolume&!mapperPlugin if volume.volumeMode == v1.PersistentVolumeBlock && mapperPlugin == nil { return nil, fmt.Errorf("could not find block volume plugin %q (spec.Name: %q) pod %q (UID: %q)") } // volume重建 reconstructed, err := rc.operationExecutor.ReconstructVolumeOperation( volume.volumeMode, plugin, mapperPlugin, pod.UID, volume.podName, volume.volumeSpecName, volume.volumePath, volume.pluginName) ... // 匹配deviceMountable plugin deviceMountablePlugin, err := rc.volumePluginMgr.FindDeviceMountablePluginBySpec(volumeSpec) ... // The unique volume name used depends on whether the volume is attachable/device-mountable needsNameFromSpec := deviceMountablePlugin != nil if !needsNameFromSpec { attachablePlugin, err := rc.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec) ... needsNameFromSpec = attachablePlugin != nil } ... // deviceMount/attachable if needsNameFromSpec { // 基于plugin+volumeSpec生成唯一名称,挂载至globalPath,pod内仅引用 uniqueVolumeName, err = util.GetUniqueVolumeNameFromSpec(plugin, volumeSpec) ... // 无需挂载globalPath的volume } else { // 混入podName生成唯一名称,挂载至pod/volume目录 uniqueVolumeName = util.GetUniqueVolumeNameFromSpecWithPod(volume.podName, plugin, volumeSpec) } ... // block volume if volume.volumeMode == v1.PersistentVolumeBlock { volumeMapper = mapperPlugin.NewBlockVolumeMapper(volumeSpec, pod, volumepkg.VolumeOptions{}) ... } else { volumeMounter = plugin.NewMounter(volumeSpec, pod, volumepkg.VolumeOptions{}) ... if deviceMountablePlugin != nil { deviceMounter = deviceMountablePlugin.NewDeviceMounter() ... } } reconstructedVolume := &reconstructedVolume{ volumeName: uniqueVolumeName, podName: volume.podName, volumeSpec: volumeSpec, outerVolumeSpecName: volume.volumeSpecName, pod: pod, deviceMounter: deviceMounter, volumeGidValue: "", devicePath: "", mounter: volumeMounter, blockVolumeMapper: volumeMapper, seLinuxMountContext: reconstructed.SELinuxMountContext, } return reconstructedVolume, 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
注意
rc.reconstructVolume基于dis volume匹配plugin，基于plugin.ConstructVolume重建vplumeSpec
# 3.3.reconcile
rc.reconcileNew()作为总调度循环，基于重建的dsw/asw数据执行清理、补齐及收尾的工作，将volume由实际状态驱动推向期望状态。func (rc *reconciler) reconcileNew() { // dsw同步完成&asw重建完成 if rc.readyToUnmount() { // Unmounts are triggered before mounts rc.unmountVolumes() } // mount required volumes. rc.mountOrAttachVolumes() // prevent unmounting volume that is still needed, but it did not reach DSW yet. if readyToUnmount { // Ensure devices that should be detached/unmounted are detached/unmounted. rc.unmountDetachDevices() // Clean up any orphan volumes that failed reconstruction. rc.cleanOrphanVolumes() } // asw volume attachable更新 if len(rc.volumesNeedUpdateFromNodeStatus) != 0 { rc.updateReconstructedFromNodeStatus() } .. // 标记dsw volume占用状态 if len(rc.volumesNeedReportedInUse) != 0 && rc.populatorHasAddedPods() { // 更新dsw volumeInIUse状态(statusManager也会更新) rc.dsw.MarkVolumesReportedInUse(rc.volumesNeedReportedInUse) rc.volumesNeedReportedInUse = nil } } // tries to file devicePaths of reconstructed volumes from node.Status.VolumesAttached. func (rc *reconciler) updateReconstructedFromNodeStatus() { ... node := rc.kubeClient.CoreV1().Nodes().Get(context.TODO(), string(rc.nodeName), metav1.GetOptions{}) ... for _, volumeID := range rc.volumesNeedUpdateFromNodeStatus { attachable := false for _, attachedVolume := range node.Status.VolumesAttached { if volumeID != attachedVolume.Name { continue } // 更新asw volume devicePath rc.actualStateOfWorld.UpdateReconstructedDevicePath(volumeID, attachedVolume.DevicePath) attachable = true } // 更新asw volume attachable状态 rc.actualStateOfWorld.UpdateReconstructedVolumeAttachability(volumeID, attachable) } rc.volumesNeedUpdateFromNodeStatus = 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
注意
reconcileNew依次触发pod volume unmount-->pod volume mount-->device unmount-->device detach