Volume Manager read link

Pre

kubernetes 通过独立于 Kubelet 运行的 Attach/Dettach Controller，分离了卷的 attach/dettach 操作，新版本仅留下 volume 挂载（mount）卸载（unmount）功能，以解决 Node （即 kubelet）不可用时导致的存储问题。

kubelet 启动的时候添加 flag 可以选择是否转移这部分功能，以兼容旧版本

Interface

  1. type VolumeManager interface {
  2.     Run(sourcesReady config.SourcesReady, stopCh <-chan struct{})
  3.     WaitForAttachAndMount(pod *v1.Pod) error
  4.     WaitForUnmount(pod *v1.Pod) error
  5.     GetMountedVolumesForPod(podName types.UniquePodName) container.VolumeMap
  6.     GetPossiblyMountedVolumesForPod(podName types.UniquePodName) container.VolumeMap
  7.     GetExtraSupplementalGroupsForPod(pod *v1.Pod) []int64
  8.     GetVolumesInUse() []v1.UniqueVolumeName
  9.     ReconcilerStatesHasBeenSynced() bool
  10.     VolumeIsAttached(volumeName v1.UniqueVolumeName) bool
  11.     MarkVolumesAsReportedInUse(volumesReportedAsInUse []v1.UniqueVolumeName)
  12. }

快速浏览一下 VolumeManager 的接口，可以看到除了 Run 接口外，其他都是工具函数，帮助获取关于 volume 相关的信息，如挂载等。

于是主要关注 Run 方法。

  1. // VolumeManager runs a set of asynchronous loops that figure out which volumes
  2. // need to be attached/mounted/unmounted/detached based on the pods scheduled on
  3. // this node and makes it so.
  4. type VolumeManager interface {
  5.     // Starts the volume manager and all the asynchronous loops that it controls
  6.     Run(sourcesReady config.SourcesReady, stopCh <-chan struct{})
  8.     // WaitForAttachAndMount processes the volumes referenced in the specified
  9.     // pod and blocks until they are all attached and mounted (reflected in
  10.     // actual state of the world).
  11.     // An error is returned if all volumes are not attached and mounted within
  12.     // the duration defined in podAttachAndMountTimeout.
  13.     WaitForAttachAndMount(pod *v1.Pod) error
  15.     // WaitForUnmount processes the volumes referenced in the specified
  16.     // pod and blocks until they are all unmounted (reflected in the actual
  17.     // state of the world).
  18.     // An error is returned if all volumes are not unmounted within
  19.     // the duration defined in podAttachAndMountTimeout.
  20.     WaitForUnmount(pod *v1.Pod) error
  22.     // GetMountedVolumesForPod returns a VolumeMap containing the volumes
  23.     // referenced by the specified pod that are successfully attached and
  24.     // mounted. The key in the map is the OuterVolumeSpecName (i.e.
  25.     // pod.Spec.Volumes[x].Name). It returns an empty VolumeMap if pod has no
  26.     // volumes.
  27.     GetMountedVolumesForPod(podName types.UniquePodName) container.VolumeMap
  29.     // GetPossiblyMountedVolumesForPod returns a VolumeMap containing the volumes
  30.     // referenced by the specified pod that are either successfully attached
  31.     // and mounted or are "uncertain", i.e. a volume plugin may be mounting
  32.     // them right now. The key in the map is the OuterVolumeSpecName (i.e.
  33.     // pod.Spec.Volumes[x].Name). It returns an empty VolumeMap if pod has no
  34.     // volumes.
  35.     GetPossiblyMountedVolumesForPod(podName types.UniquePodName) container.VolumeMap
  37.     // GetExtraSupplementalGroupsForPod returns a list of the extra
  38.     // supplemental groups for the Pod. These extra supplemental groups come
  39.     // from annotations on persistent volumes that the pod depends on.
  40.     GetExtraSupplementalGroupsForPod(pod *v1.Pod) []int64
  42.     // GetVolumesInUse returns a list of all volumes that implement the volume.Attacher
  43.     // interface and are currently in use according to the actual and desired
  44.     // state of the world caches. A volume is considered "in use" as soon as it
  45.     // is added to the desired state of world, indicating it *should* be
  46.     // attached to this node and remains "in use" until it is removed from both
  47.     // the desired state of the world and the actual state of the world, or it
  48.     // has been unmounted (as indicated in actual state of world).
  49.     GetVolumesInUse() []v1.UniqueVolumeName
  51.     // ReconcilerStatesHasBeenSynced returns true only after the actual states in reconciler
  52.     // has been synced at least once after kubelet starts so that it is safe to update mounted
  53.     // volume list retrieved from actual state.
  54.     ReconcilerStatesHasBeenSynced() bool
  56.     // VolumeIsAttached returns true if the given volume is attached to this
  57.     // node.
  58.     VolumeIsAttached(volumeName v1.UniqueVolumeName) bool
  60.     // Marks the specified volume as having successfully been reported as "in
  61.     // use" in the nodes's volume status.
  62.     MarkVolumesAsReportedInUse(volumesReportedAsInUse []v1.UniqueVolumeName)
  63. }

Structure

  1. // volumeManager implements the VolumeManager interface
  2. type volumeManager struct {
  3.     // kubeClient is the kube API client used by DesiredStateOfWorldPopulator to
  4.     // communicate with the API server to fetch PV and PVC objects
  5.     kubeClient clientset.Interface
  7.     // volumePluginMgr is the volume plugin manager used to access volume
  8.     // plugins. It must be pre-initialized.
  9.     volumePluginMgr *volume.VolumePluginMgr
  11.     // desiredStateOfWorld is a data structure containing the desired state of
  12.     // the world according to the volume manager: i.e. what volumes should be
  13.     // attached and which pods are referencing the volumes).
  14.     // The data structure is populated by the desired state of the world
  15.     // populator using the kubelet pod manager.
  16.     desiredStateOfWorld cache.DesiredStateOfWorld
  18.     // actualStateOfWorld is a data structure containing the actual state of
  19.     // the world according to the manager: i.e. which volumes are attached to
  20.     // this node and what pods the volumes are mounted to.
  21.     // The data structure is populated upon successful completion of attach,
  22.     // detach, mount, and unmount actions triggered by the reconciler.
  23.     actualStateOfWorld cache.ActualStateOfWorld
  25.     // operationExecutor is used to start asynchronous attach, detach, mount,
  26.     // and unmount operations.
  27.     operationExecutor operationexecutor.OperationExecutor
  29.     // reconciler runs an asynchronous periodic loop to reconcile the
  30.     // desiredStateOfWorld with the actualStateOfWorld by triggering attach,
  31.     // detach, mount, and unmount operations using the operationExecutor.
  32.     reconciler reconciler.Reconciler
  34.     // desiredStateOfWorldPopulator runs an asynchronous periodic loop to
  35.     // populate the desiredStateOfWorld using the kubelet PodManager.
  36.     desiredStateOfWorldPopulator populator.DesiredStateOfWorldPopulator
  38.     // csiMigratedPluginManager keeps track of CSI migration status of plugins
  39.     csiMigratedPluginManager csimigration.PluginManager
  41.     // intreeToCSITranslator translates in-tree volume specs to CSI
  42.     intreeToCSITranslator csimigration.InTreeToCSITranslator
  43. }

其中 csiMigratedPluginManager，volumePluginMgr，intreeToCSITranslator，operationExecutor，kub，eClient 和核心逻辑无关，是功能的执行者。重要的数据结构是 desiredStateOfWorld，actualStateOfWorld，reconciler，desiredStateOfWorldPopulator 的交互。

Explanation

  • desiredStateOfWorld: 期望的世界状态，简称 DSW，描述期望的 Pod 和 Volume 的挂载关系。
  • actualStateOfWorld: 实际世界状态，简称ASW，即此时 volume 实际状态。实际状态未必是和期望状态一样，他代表现在进行时。

    Overview

    /
    Volume Manager - 图1

事件来源：pod 的数据由 desiredStateOfWorldPopulator从 PodManager （PodManager 是 kubelet 中的 manager，它通过轮询维护着真实的 pod 数据及状态）中进行同步到 desiredStateOfWorld 中去。
事件处理：由 reconciler的 Run 方法持续 reconcile，保证记录在desiredStateOfWorld中 volume 卷与 pod 的关系处理同步到 actualStateOfWorld中。
重启与残留清理：reconciler除了负责触发挂载事件交由实际程序处理之外，还会进行卷的重建和清理工作（不支持重建的卷插件会清理他）

Run(入口）

  1. go vm.volumePluginMgr.Run(stopCh)为 CSIDriver 启动 informer
  2. go vm.desiredStateOfWorldPopulator.Run(sourcesReady, stopCh)：开启循环，不断从 PodManager 中拿到 Pod 信息并处理这些 Pod 里的 Volume 信息，将它们加入到 DSW 中。

  3. go vm.reconciler.Run(stopCh)：开始 VolumeManager 的主要调谐循环。

    1. func (vm *volumeManager) Run(sourcesReady config.SourcesReady, stopCh <-chan struct{}) {
    2.  defer runtime.HandleCrash()
    4.  if vm.kubeClient != nil {
    5.      // start informer for CSIDriver
    6.      go vm.volumePluginMgr.Run(stopCh)
    7.  }
    9.  go vm.desiredStateOfWorldPopulator.Run(sourcesReady, stopCh)
    10.  klog.V(2).InfoS("The desired_state_of_world populator starts")
    12.  klog.InfoS("Starting Kubelet Volume Manager")
    13.  go vm.reconciler.Run(stopCh)
    15.  metrics.Register(vm.actualStateOfWorld, vm.desiredStateOfWorld, vm.volumePluginMgr)
    17.  <-stopCh
    18.  klog.InfoS("Shutting down Kubelet Volume Manager")
    19. }

    Reconcile

    数据结构

    1. type podVolume struct {
    2.  podName        volumetypes.UniquePodName
    3.  volumeSpecName string
    4.  volumePath     string
    5.  pluginName     string
    6.  volumeMode     v1.PersistentVolumeMode
    7. }

    Pod 和 Volume 的关系，包含必要的处理信息。

    状态同步

    VolumeManager 什么时候同步状态呢？可以在 Run 函数中找到答案

    1. func (rc *reconciler) reconciliationLoopFunc() func() {
    2.  return func() {
    3.      rc.reconcile()
    5.      // Sync the state with the reality once after all existing pods are added to the desired state from all sources.
    6.      // Otherwise, the reconstruct process may clean up pods' volumes that are still in use because
    7.      // desired state of world does not contain a complete list of pods.
    8.      if rc.populatorHasAddedPods() && !rc.StatesHasBeenSynced() {
    9.          klog.InfoS("Reconciler: start to sync state")
    10.          rc.sync()
    11.      }
    12.  }
    13. }

    其中 rc.sync()是同步函数。两个条件分别是

  • populator 开始运行添加了 podManager 中相关的 Pod：查看实现可以知道，只要 Populator.Run 开始运行，这个函数就会返回 true ```go rc.populatorHasAddedPods = vm.desiredStateOfWorldPopulator.HasAddedPods

func (dswp *desiredStateOfWorldPopulator) HasAddedPods() bool { dswp.hasAddedPodsLock.RLock() defer dswp.hasAddedPodsLock.RUnlock() return dswp.hasAddedPods }

// 在 Run 函数中置为 true。 func (dswp *desiredStateOfWorldPopulator) Run(sourcesReady config.SourcesReady, stopCh <-chan struct{}) { // Wait for the completion of a loop that started after sources are all ready, then set hasAddedPods accordingly klog.InfoS(“Desired state populator starts to run”) wait.PollUntil(dswp.loopSleepDuration, func() (bool, error) { done := sourcesReady.AllReady() dswp.populatorLoop() return done, nil }, stopCh) dswp.hasAddedPodsLock.Lock() dswp.hasAddedPods = true dswp.hasAddedPodsLock.Unlock() wait.Until(dswp.populatorLoop, dswp.loopSleepDuration, stopCh) }

  2. - 还没有同步过状态。
  3. ```go
  4. func (rc *reconciler) StatesHasBeenSynced() bool {
  5.     return !rc.timeOfLastSync.IsZero()
  6. }

同步可以保证 desired state of world 保留全部 pods 的列表，其意义是避免 重建 volume 时会清理正在使用的 pod 的 volume 卷。

sync() 详情

sync 会将 Pod 目录中所有的 volume 文件夹遍历一遍，不在 DSW 中的 volume 会被 重建 相关信息，并添加到 DSW 和 ASW 中。如果不支持 重建 的（一些 volume plugin 不支持）的卷，会在卷不在被使用之后清理他的挂载点

  1. func (rc *reconciler) sync() {
  2.     defer rc.updateLastSyncTime()
  3.     rc.syncStates()
  4. }
  6. func (rc *reconciler) syncStates() {
  7.     // Get volumes information by reading the pod's directory
  8.     podVolumes, err := getVolumesFromPodDir(rc.kubeletPodsDir)
  9.     if err != nil {
  10.         klog.ErrorS(err, "Cannot get volumes from disk, skip sync states for volume reconstruction")
  11.         return
  12.     }
  13.     volumesNeedUpdate := make(map[v1.UniqueVolumeName]*reconstructedVolume)
  14.     volumeNeedReport := []v1.UniqueVolumeName{}
  15.     for _, volume := range podVolumes {
  16.         if rc.actualStateOfWorld.VolumeExistsWithSpecName(volume.podName, volume.volumeSpecName) {
  17.             klog.V(4).InfoS("Volume exists in actual state, skip cleaning up mounts", "podName", volume.podName, "volumeSpecName", volume.volumeSpecName)
  18.             // There is nothing to reconstruct
  19.             continue
  20.         }
  21.         volumeInDSW := rc.desiredStateOfWorld.VolumeExistsWithSpecName(volume.podName, volume.volumeSpecName)
  23.         reconstructedVolume, err := rc.reconstructVolume(volume)
  24.         if err != nil {
  25.             if volumeInDSW {
  26.                 // Some pod needs the volume, don't clean it up and hope that
  27.                 // reconcile() calls SetUp and reconstructs the volume in ASW.
  28.                 klog.V(4).InfoS("Volume exists in desired state, skip cleaning up mounts", "podName", volume.podName, "volumeSpecName", volume.volumeSpecName)
  29.                 continue
  30.             }
  31.             // No pod needs the volume.
  32.             klog.InfoS("Could not construct volume information, cleaning up mounts", "podName", volume.podName, "volumeSpecName", volume.volumeSpecName, "err", err)
  33.             rc.cleanupMounts(volume)
  34.             continue
  35.         }
  36.         if volumeInDSW {
  37.             // Some pod needs the volume. And it exists on disk. Some previous
  38.             // kubelet must have created the directory, therefore it must have
  39.             // reported the volume as in use. Mark the volume as in use also in
  40.             // this new kubelet so reconcile() calls SetUp and re-mounts the
  41.             // volume if it's necessary.
  42.             volumeNeedReport = append(volumeNeedReport, reconstructedVolume.volumeName)
  43.             klog.V(4).InfoS("Volume exists in desired state, marking as InUse", "podName", volume.podName, "volumeSpecName", volume.volumeSpecName)
  44.             continue
  45.         }
  46.         // There is no pod that uses the volume.
  47.         if rc.operationExecutor.IsOperationPending(reconstructedVolume.volumeName, nestedpendingoperations.EmptyUniquePodName, nestedpendingoperations.EmptyNodeName) {
  48.             klog.InfoS("Volume is in pending operation, skip cleaning up mounts")
  49.         }
  50.         klog.V(2).InfoS("Reconciler sync states: could not find pod information in desired state, update it in actual state", "reconstructedVolume", reconstructedVolume)
  51.         volumesNeedUpdate[reconstructedVolume.volumeName] = reconstructedVolume
  52.     }
  54.     if len(volumesNeedUpdate) > 0 {
  55.         if err = rc.updateStates(volumesNeedUpdate); err != nil {
  56.             klog.ErrorS(err, "Error occurred during reconstruct volume from disk")
  57.         }
  58.     }
  59.     if len(volumeNeedReport) > 0 {
  60.         rc.desiredStateOfWorld.MarkVolumesReportedInUse(volumeNeedReport)
  61.     }
  62. }

Working

那么剩余的工作 reconcile() 则会负责

  1. func (rc *reconciler) reconcile() {
  2.     // Unmounts are triggered before mounts so that a volume that was
  3.     // referenced by a pod that was deleted and is now referenced by another
  4.     // pod is unmounted from the first pod before being mounted to the new
  5.     // pod.
  6.     rc.unmountVolumes()
  8.     // Next we mount required volumes. This function could also trigger
  9.     // attach if kubelet is responsible for attaching volumes.
  10.     // If underlying PVC was resized while in-use then this function also handles volume
  11.     // resizing.
  12.     rc.mountOrAttachVolumes()
  14.     // Ensure devices that should be detached/unmounted are detached/unmounted.
  15.     rc.unmountDetachDevices()
  16. }
  1. 取消卷挂载
  2. 挂载卷到节点或附加卷到 pod（attachvolume）

  3. 卸载卷或从 pod 中取消附加（detach）

    函数细节

    1. func (rc *reconciler) unmountVolumes() {
    2.  // Ensure volumes that should be unmounted are unmounted.
    3.  for _, mountedVolume := range rc.actualStateOfWorld.GetAllMountedVolumes() {
    4.      if !rc.desiredStateOfWorld.PodExistsInVolume(mountedVolume.PodName, mountedVolume.VolumeName) {
    5.          // Volume is mounted, unmount it
    6.          klog.V(5).InfoS(mountedVolume.GenerateMsgDetailed("Starting operationExecutor.UnmountVolume", ""))
    7.          err := rc.operationExecutor.UnmountVolume(
    8.              mountedVolume.MountedVolume, rc.actualStateOfWorld, rc.kubeletPodsDir)
    9.          if err != nil && !isExpectedError(err) {
    10.              klog.ErrorS(err, mountedVolume.GenerateErrorDetailed(fmt.Sprintf("operationExecutor.UnmountVolume failed (controllerAttachDetachEnabled %v)", rc.controllerAttachDetachEnabled), err).Error())
    11.          }
    12.          if err == nil {
    13.              klog.InfoS(mountedVolume.GenerateMsgDetailed("operationExecutor.UnmountVolume started", ""))
    14.          }
    15.      }
    16.  }
    17. }

    遍历 ASW 中所有已经挂载的卷，比较 DSW 是否含有该卷和对应 Pod 的关系，很明显，如果没有，就把他给取消挂载。

    1. func (rc *reconciler) mountOrAttachVolumes() {
    2.  // Ensure volumes that should be attached/mounted are attached/mounted.
    3.  for _, volumeToMount := range rc.desiredStateOfWorld.GetVolumesToMount() {
    4.      volMounted, devicePath, err := rc.actualStateOfWorld.PodExistsInVolume(volumeToMount.PodName, volumeToMount.VolumeName, volumeToMount.PersistentVolumeSize)
    5.      volumeToMount.DevicePath = devicePath
    6.      if cache.IsVolumeNotAttachedError(err) {
    7.          rc.waitForVolumeAttach(volumeToMount)
    8.      } else if !volMounted || cache.IsRemountRequiredError(err) {
    9.          rc.mountAttachedVolumes(volumeToMount, err)
    10.      } else if cache.IsFSResizeRequiredError(err) {
    11.          fsResizeRequiredErr, _ := err.(cache.FsResizeRequiredError)
    12.          rc.expandVolume(volumeToMount, fsResizeRequiredErr.CurrentSize)
    13.      }
    14.  }
    15. }

    遍历 DSW 中需要被挂载（Mount）的卷，判断该卷是否要进行挂载操作

  4. 判断是否有 Attach 错误，如果没有就等待其挂载 :::info 注意这里就和我们开头提到的 AD Controller 相关了。会判断是否开启 AD Controller，如果开启了就由其做 Attach 操作，这里只做 Attach 是否成功的检查。如果没有就是这里自己去 Attach :::

  5. 如果还没挂载（Mount）就 Mount 并 Attach 这个卷。

  6. 如果是要求 Resize，则调用 rc.expandVolume 进行扩容操作。
  1. func (rc *reconciler) unmountDetachDevices() {
  2.     for _, attachedVolume := range rc.actualStateOfWorld.GetUnmountedVolumes() {
  3.         // Check IsOperationPending to avoid marking a volume as detached if it's in the process of mounting.
  4.         if !rc.desiredStateOfWorld.VolumeExists(attachedVolume.VolumeName) &&
  5.             !rc.operationExecutor.IsOperationPending(attachedVolume.VolumeName, nestedpendingoperations.EmptyUniquePodName, nestedpendingoperations.EmptyNodeName) {
  6.             if attachedVolume.DeviceMayBeMounted() {
  7.                 // Volume is globally mounted to device, unmount it
  8.                 klog.V(5).InfoS(attachedVolume.GenerateMsgDetailed("Starting operationExecutor.UnmountDevice", ""))
  9.                 err := rc.operationExecutor.UnmountDevice(
  10.                     attachedVolume.AttachedVolume, rc.actualStateOfWorld, rc.hostutil)
  11.                 if err != nil && !isExpectedError(err) {
  12.                     klog.ErrorS(err, attachedVolume.GenerateErrorDetailed(fmt.Sprintf("operationExecutor.UnmountDevice failed (controllerAttachDetachEnabled %v)", rc.controllerAttachDetachEnabled), err).Error())
  13.                 }
  14.                 if err == nil {
  15.                     klog.InfoS(attachedVolume.GenerateMsgDetailed("operationExecutor.UnmountDevice started", ""))
  16.                 }
  17.             } else {
  18.                 // Volume is attached to node, detach it
  19.                 // Kubelet not responsible for detaching or this volume has a non-attachable volume plugin.
  20.                 if rc.controllerAttachDetachEnabled || !attachedVolume.PluginIsAttachable {
  21.                     rc.actualStateOfWorld.MarkVolumeAsDetached(attachedVolume.VolumeName, attachedVolume.NodeName)
  22.                     klog.InfoS(attachedVolume.GenerateMsgDetailed("Volume detached", fmt.Sprintf("DevicePath %q", attachedVolume.DevicePath)))
  23.                 } else {
  24.                     // Only detach if kubelet detach is enabled
  25.                     klog.V(5).InfoS(attachedVolume.GenerateMsgDetailed("Starting operationExecutor.DetachVolume", ""))
  26.                     err := rc.operationExecutor.DetachVolume(
  27.                         attachedVolume.AttachedVolume, false /* verifySafeToDetach */, rc.actualStateOfWorld)
  28.                     if err != nil && !isExpectedError(err) {
  29.                         klog.ErrorS(err, attachedVolume.GenerateErrorDetailed(fmt.Sprintf("operationExecutor.DetachVolume failed (controllerAttachDetachEnabled %v)", rc.controllerAttachDetachEnabled), err).Error())
  30.                     }
  31.                     if err == nil {
  32.                         klog.InfoS(attachedVolume.GenerateMsgDetailed("operationExecutor.DetachVolume started", ""))
  33.                     }
  34.                 }
  35.             }
  36.         }
  37.     }
  38. }
  1. 还是从 ASW 中遍历所有取消挂载的卷，过滤在 DSW 中声明应该有的卷，也过滤正在进行 Mount（挂载）操作的卷。
  2. 对过滤掉的卷，进行 Detach 操作，和前文一样，如果开启了 AD Controller。Detach 操作不会由 kubelet 直接执行。