深入理解kubelet - VolumeManager源码解析

 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

// volumeManager implements the VolumeManager interface
type volumeManager struct {
    // DesiredStateOfWorldPopulator 用来与 API 服务器通信以获取 PV 和 PVC 对象的 API 客户端
	kubeClient clientset.Interface
    
    // VolumePluginMgr 是用于访问 VolumePlugin 插件的 VolumePlugin 管理器。它必须预初始化。
	volumePluginMgr *volume.VolumePluginMgr

    // desiredStateOfWorld 是一个数据结构，包含根据 VM 所需的状态：即应附加哪些卷以及 "哪些pod” 正在引用这些卷。
    // 使用 kubelet pod manager 根据 world populator 的所需状态填充数据结构。
	desiredStateOfWorld cache.DesiredStateOfWorld

	// 与 desiredStateOfWorld 相似，是实际状态：即哪些卷被 attacted 到该 Node 以及 volume 被 mounted 到哪些 pod。
    // 成功完成 reconciler attach,detach, mount, 和 unmount 操作后，将填充数据结构。
	actualStateOfWorld cache.ActualStateOfWorld

	// operationExecutor 用于启动异步 attach,detach, mount, 和 unmount 操作。
	operationExecutor operationexecutor.OperationExecutor

	// reconciler reconciler 运行异步周期性循环，通过使用操作执行器触发 attach,detach, mount, 和 unmount操作
    // 来协调 desiredStateOfWorld 与 actualStateOfWorld。
	reconciler reconciler.Reconciler

    // desiredStateOfWorldPopulator 运行异步周期性循环以使用 kubelet Pod Manager 填充desiredStateOfWorld。
	desiredStateOfWorldPopulator populator.DesiredStateOfWorldPopulator

	// csiMigratedPluginManager keeps track of CSI migration status of plugins
	csiMigratedPluginManager csimigration.PluginManager

	// intreeToCSITranslator translates in-tree volume specs to CSI
	intreeToCSITranslator csimigration.InTreeToCSITranslator
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

vm := &volumeManager{
    kubeClient:          kubeClient,
    volumePluginMgr:     volumePluginMgr,
    desiredStateOfWorld: cache.NewDesiredStateOfWorld(volumePluginMgr),
    actualStateOfWorld:  cache.NewActualStateOfWorld(nodeName, volumePluginMgr),
    operationExecutor: operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
        kubeClient,
        volumePluginMgr,
        recorder,
        checkNodeCapabilitiesBeforeMount,
        blockVolumePathHandler)),
}

 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

vm.intreeToCSITranslator = intreeToCSITranslator
vm.csiMigratedPluginManager = csiMigratedPluginManager
vm.desiredStateOfWorldPopulator = populator.NewDesiredStateOfWorldPopulator(
    kubeClient,
    desiredStateOfWorldPopulatorLoopSleepPeriod,
    desiredStateOfWorldPopulatorGetPodStatusRetryDuration,
    podManager,
    podStatusProvider,
    vm.desiredStateOfWorld,
    vm.actualStateOfWorld,
    kubeContainerRuntime,
    keepTerminatedPodVolumes,
    csiMigratedPluginManager,
    intreeToCSITranslator)
vm.reconciler = reconciler.NewReconciler(
    kubeClient,
    controllerAttachDetachEnabled,
    reconcilerLoopSleepPeriod,
    waitForAttachTimeout,
    nodeName,
    vm.desiredStateOfWorld,
    vm.actualStateOfWorld,
    vm.desiredStateOfWorldPopulator.HasAddedPods,
    vm.operationExecutor,
    mounter,
    hostutil,
    volumePluginMgr,
    kubeletPodsDir)

1
2

// Start volume manager
go kl.volumeManager.Run(kl.sourcesReady, wait.NeverStop)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

func (vm *volumeManager) Run(sourcesReady config.SourcesReady, stopCh <-chan struct{}) {
	defer runtime.HandleCrash()

	go vm.desiredStateOfWorldPopulator.Run(sourcesReady, stopCh)
	klog.V(2).Infof("The desired_state_of_world populator starts")

	klog.Infof("Starting Kubelet Volume Manager")
	go vm.reconciler.Run(stopCh)

	metrics.Register(vm.actualStateOfWorld, vm.desiredStateOfWorld, vm.volumePluginMgr)

	if vm.kubeClient != nil {
		// start informer for CSIDriver
		vm.volumePluginMgr.Run(stopCh)
	}

	<-stopCh
	klog.Infof("Shutting down Kubelet Volume Manager")
}

 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

func (rc *reconciler) Run(stopCh <-chan struct{}) {
	wait.Until(rc.reconciliationLoopFunc(), rc.loopSleepDuration, stopCh)
}

func (rc *reconciler) reconciliationLoopFunc() func() {
	return func() {
		rc.reconcile()

		// Sync the state with the reality once after all existing pods are added to the desired state from all sources.
		// Otherwise, the reconstruct process may clean up pods' volumes that are still in use because
		// desired state of world does not contain a complete list of pods.
		if rc.populatorHasAddedPods() && !rc.StatesHasBeenSynced() {
			klog.Infof("Reconciler: start to sync state")
			rc.sync()
		}
	}
}

func (rc *reconciler) reconcile() {
   // Unmounts are triggered before mounts so that a volume that was
   // referenced by a pod that was deleted and is now referenced by another
   // pod is unmounted from the first pod before being mounted to the new
   // pod.
   // 卸载会在挂载之前触发，以便已删除的 Pod 引用的卷现在被另一个 Pod 引用，
   // 然后再挂载到新 Pod 之前从第一个 Pod 中卸载。
   rc.unmountVolumes()

   // Next we mount required volumes. This function could also trigger
   // attach if kubelet is responsible for attaching volumes.
   // If underlying PVC was resized while in-use then this function also handles volume
   // resizing.
   // 接下来我们安装所需的卷。如果 kubelet 负责附加卷，
   // 则此函数还可以触发附加。如果底层 PVC 在使用时调整了大小，则此函数还可以处理卷大小调整。
   rc.mountAttachVolumes()

   // Ensure devices that should be detached/unmounted are detached/unmounted.
   // 确保应 detached/unmounted 的设备已完成 detached/unmounted。
   rc.unmountDetachDevices()
}