Kubelet VolumeManager源码分析: 卷的卸载

func (rc *reconciler) reconcile() {
    rc.unmountVolumes()
    rc.mountAttachVolumes()
    rc.unmountDetachDevices()
}

func (rc *reconciler) unmountVolumes() {
    for _, mountedVolume := range rc.actualStateOfWorld.GetAllMountedVolumes() {
        if !rc.desiredStateOfWorld.PodExistsInVolume(mountedVolume.PodName, mountedVolume.VolumeName) {
            ...
            err := rc.operationExecutor.UnmountVolume(
                mountedVolume.MountedVolume, rc.actualStateOfWorld, rc.kubeletPodsDir)
            ...
            }
        }
    }
}

func (oe *operationExecutor) UnmountVolume(
    volumeToUnmount MountedVolume,
    actualStateOfWorld ActualStateOfWorldMounterUpdater,
    podsDir string) error {
    fsVolume, err := util.CheckVolumeModeFilesystem(volumeToUnmount.VolumeSpec)
    if err != nil {
        return err
    }
    var generatedOperations volumetypes.GeneratedOperations
    if fsVolume {
        // Filesystem volume case
        // Unmount a volume if a volume is mounted
        generatedOperations, err = oe.operationGenerator.GenerateUnmountVolumeFunc(
            volumeToUnmount, actualStateOfWorld, podsDir)
    } else {
        // Block volume case
        // Unmap a volume if a volume is mapped
        generatedOperations, err = oe.operationGenerator.GenerateUnmapVolumeFunc(
            volumeToUnmount, actualStateOfWorld)
    }
    if err != nil {
        return err
    }
    // 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)
}

func (o *GeneratedOperations) Run() (eventErr, detailedErr error) {
    var context OperationContext
    if o.CompleteFunc != nil {
        c := CompleteFuncParam{
            Err:      &context.DetailedErr,
            Migrated: &context.Migrated,
        }
        defer o.CompleteFunc(c)
    }
    if o.EventRecorderFunc != nil {
        defer o.EventRecorderFunc(&eventErr)
    }
    // Handle panic, if any, from operationFunc()
    defer runtime.RecoverFromPanic(&detailedErr)

    context = o.OperationFunc()
    return context.EventErr, context.DetailedErr
}

func (og *operationGenerator) GenerateUnmountVolumeFunc(
    volumeToUnmount MountedVolume,
    actualStateOfWorld ActualStateOfWorldMounterUpdater,
    podsDir string) (volumetypes.GeneratedOperations, error) {
    ...
    unmountVolumeFunc := func() volumetypes.OperationContext {
        subpather := og.volumePluginMgr.Host.GetSubpather()

        migrated := getMigratedStatusBySpec(volumeToUnmount.VolumeSpec)

        // Remove all bind-mounts for subPaths
        podDir := filepath.Join(podsDir, string(volumeToUnmount.PodUID))
        if err := subpather.CleanSubPaths(podDir, volumeToUnmount.InnerVolumeSpecName); err != nil {
            eventErr, detailedErr := volumeToUnmount.GenerateError("error cleaning subPath mounts", err)
            return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
        }

        // Execute unmount
        unmountErr := volumeUnmounter.TearDown()
        if unmountErr != nil {
            ...
            return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
        }
        ...
        return volumetypes.NewOperationContext(nil, nil, migrated)
    }

    return volumetypes.GeneratedOperations{
        OperationName:     "volume_unmount",
        OperationFunc:     unmountVolumeFunc,
        CompleteFunc:      util.OperationCompleteHook(util.GetFullQualifiedPluginNameForVolume(volumePlugin.GetPluginName(), volumeToUnmount.VolumeSpec), "volume_unmount"),
        EventRecorderFunc: nil, // nil because we do not want to generate event on error
    }, nil
}

func (sp *subpath) CleanSubPaths(podDir string, volumeName string) error {
    return doCleanSubPaths(sp.mounter, podDir, volumeName)
}

func doCleanSubPaths(mounter mount.Interface, podDir string, volumeName string) error {
    // scan /var/lib/kubelet/pods/<uid>/volume-subpaths/<volume>/*
    subPathDir := filepath.Join(podDir, containerSubPathDirectoryName, volumeName)
    ...
    containerDirs, err := ioutil.ReadDir(subPathDir)
    if err != nil {
        if os.IsNotExist(err) {
            return nil
        }
        return fmt.Errorf("error reading %s: %s", subPathDir, err)
    }

    for _, containerDir := range containerDirs {
        if !containerDir.IsDir() {
            ...
            continue
        }
        ...
        // scan /var/lib/kubelet/pods/<uid>/volume-subpaths/<volume>/<container name>/*
        fullContainerDirPath := filepath.Join(subPathDir, containerDir.Name())
        err = filepath.Walk(fullContainerDirPath, func(path string, info os.FileInfo, _ error) error {
            if path == fullContainerDirPath {
                // Skip top level directory
                return nil
            }

            // pass through errors and let doCleanSubPath handle them
            if err = doCleanSubPath(mounter, fullContainerDirPath, filepath.Base(path)); err != nil {
                return err
            }

            // We need to check that info is not nil. This may happen when the incoming err is not nil due to stale mounts or permission errors.
            if info != nil && info.IsDir() {
                // skip subdirs of the volume: it only matters the first level to unmount, otherwise it would try to unmount subdir of the volume
                return filepath.SkipDir
            }

            return nil
        })
        ...
        if err := os.Remove(fullContainerDirPath); err != nil {
            return fmt.Errorf("error deleting %s: %s", fullContainerDirPath, err)
        }
        ...
    }
    if err := os.Remove(subPathDir); err != nil {
        return fmt.Errorf("error deleting %s: %s", subPathDir, err)
    }
    ...
    podSubPathDir := filepath.Join(podDir, containerSubPathDirectoryName)
    if err := os.Remove(podSubPathDir); err != nil && !os.IsExist(err) {
        return fmt.Errorf("error deleting %s: %s", podSubPathDir, err)
    }
    ...
}

func doCleanSubPath(mounter mount.Interface, fullContainerDirPath, subPathIndex string) error {
    // process /var/lib/kubelet/pods/<uid>/volume-subpaths/<volume>/<container name>/<subPathName>
    ...
    fullSubPath := filepath.Join(fullContainerDirPath, subPathIndex)

    if err := mount.CleanupMountPoint(fullSubPath, mounter, true); err != nil {
        return fmt.Errorf("error cleaning subpath mount %s: %s", fullSubPath, err)
    }
    ...
    return nil
}

func CleanupMountPoint(mountPath string, mounter Interface, extensiveMountPointCheck bool) error {
    ...
    return doCleanupMountPoint(mountPath, mounter, extensiveMountPointCheck, corruptedMnt)
}

func doCleanupMountPoint(mountPath string, mounter Interface, extensiveMountPointCheck bool, corruptedMnt bool) error {
    if !corruptedMnt {
        notMnt, err = removePathIfNotMountPoint(mountPath, mounter, extensiveMountPointCheck)
        ...
    }
    ...
    if err := mounter.Unmount(mountPath); err != nil {
        return err
    }
    notMnt, err = removePathIfNotMountPoint(mountPath, mounter, extensiveMountPointCheck)
    ...
}

func (mounter *Mounter) Unmount(target string) error {
    command := exec.Command("umount", target)
    output, err := command.CombinedOutput()
    ...
}

// Unmounter interface provides methods to cleanup/unmount the volumes.
type Unmounter interface {
	Volume
	// TearDown unmounts the volume from a self-determined directory and
	// removes traces of the SetUp procedure.
	TearDown() error
	// TearDown unmounts the volume from the specified directory and
	// removes traces of the SetUp procedure.
	TearDownAt(dir string) error
}

apiVersion: v1
kind: PersistentVolume
metadata:
  name: example-local
spec:
  capacity:
    storage: 1Gi
  volumeMode: Block
  accessModes:
  - ReadWriteOnce
  persistentVolumeReclaimPolicy: Delete
  storageClassName: local-storage
  local:
    path: /dev/vdb
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - wanglei-k8s-containerd

---
kind: StorageClass
apiVersion: storage.k8s.io/v1
metadata:
  name: local-storage
provisioner: kubernetes.io/no-provisioner
volumeBindingMode: WaitForFirstConsumer

---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: example-local-claim
spec:
  volumeMode: Block
  storageClassName: local-storage
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 1Gi

func (og *operationGenerator) GenerateUnmapVolumeFunc(
    ...
    unmapVolumeFunc := func() volumetypes.OperationContext {
        ...
        // Execute common unmap
        unmapErr := util.UnmapBlockVolume(og.blkUtil, globalUnmapPath, podDeviceUnmapPath, volName, volumeToUnmount.PodUID)
        if unmapErr != nil {
            // On failure, return error. Caller will log and retry.
            eventErr, detailedErr := volumeToUnmount.GenerateError("UnmapVolume.UnmapBlockVolume failed", unmapErr)
            return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
        }

        // Call UnmapPodDevice if blockVolumeUnmapper implements CustomBlockVolumeUnmapper
        if customBlockVolumeUnmapper, ok := blockVolumeUnmapper.(volume.CustomBlockVolumeUnmapper); ok {
            // Execute plugin specific unmap
            unmapErr = customBlockVolumeUnmapper.UnmapPodDevice()
            if unmapErr != nil {
                // On failure, return error. Caller will log and retry.
                eventErr, detailedErr := volumeToUnmount.GenerateError("UnmapVolume.UnmapPodDevice failed", unmapErr)
                return volumetypes.NewOperationContext(eventErr, detailedErr, migrated)
            }
        }
        ...
        return volumetypes.NewOperationContext(nil, nil, migrated)
    }

    return volumetypes.GeneratedOperations{
        ...
        OperationFunc:     unmapVolumeFunc,
        ...
    }, nil
}

func UnmapBlockVolume(...) error {
    // Release file descriptor lock.
    err := blkUtil.DetachFileDevice(filepath.Join(globalUnmapPath, string(podUID)))
    if err != nil {
        return fmt.Errorf("blkUtil.DetachFileDevice failed. globalUnmapPath:%s, podUID: %s: %v",
            globalUnmapPath, string(podUID), err)
    }

    // unmap devicePath from pod volume path
    unmapDeviceErr := blkUtil.UnmapDevice(podDeviceUnmapPath, volumeMapName, false /* bindMount */)
    if unmapDeviceErr != nil {
        return fmt.Errorf("blkUtil.DetachFileDevice failed. podDeviceUnmapPath:%s, volumeMapName: %s, bindMount: %v: %v",
            podDeviceUnmapPath, volumeMapName, false, unmapDeviceErr)
    }

    // unmap devicePath from global node path
    unmapDeviceErr = blkUtil.UnmapDevice(globalUnmapPath, string(podUID), true /* bindMount */)
    if unmapDeviceErr != nil {
        return fmt.Errorf("blkUtil.DetachFileDevice failed. globalUnmapPath:%s, podUID: %s, bindMount: %v: %v",
            globalUnmapPath, string(podUID), true, unmapDeviceErr)
    }
    return nil
}

func (v VolumePathHandler) DetachFileDevice(path string) error {
    loopPath, err := v.GetLoopDevice(path)
    ...
    err = removeLoopDevice(loopPath)
    ...
}

func (v VolumePathHandler) GetLoopDevice(path string) (string, error) {
    ...
    return getLoopDeviceFromSysfs(path)
}

func getLoopDeviceFromSysfs(path string) (string, error) {
    // If the file is a symlink.
    realPath, err := filepath.EvalSymlinks(path)
    if err != nil {
        return "", fmt.Errorf("failed to evaluate path %s: %s", path, err)
    }

    devices, err := filepath.Glob("/sys/block/loop*")
    if err != nil {
        return "", fmt.Errorf("failed to list loop devices in sysfs: %s", err)
    }

    for _, device := range devices {
        backingFile := fmt.Sprintf("%s/loop/backing_file", device)

        // The contents of this file is the absolute path of "path".
        data, err := ioutil.ReadFile(backingFile)
        if err != nil {
            continue
        }

        // Return the first match.
        backingFilePath := strings.TrimSpace(string(data))
        if backingFilePath == path || backingFilePath == realPath {
            return fmt.Sprintf("/dev/%s", filepath.Base(device)), nil
        }
    }

    return "", errors.New(ErrDeviceNotFound)
}

func removeLoopDevice(device string) error {
    args := []string{"-d", device}
    cmd := exec.Command(losetupPath, args...)
    out, err := cmd.CombinedOutput()
    ...
}

func UnmapBlockVolume(...) error {
    ...
    unmapDeviceErr := blkUtil.UnmapDevice(podDeviceUnmapPath, volumeMapName, false /* bindMount */)
    ...
}

func (v VolumePathHandler) UnmapDevice(mapPath string, linkName string, bindMount bool) error {
    ...
    return unmapSymlinkDevice(v, mapPath, linkName)
}

func unmapSymlinkDevice(v VolumePathHandler, mapPath string, linkName string) error {
	// Check symbolic link exists
	linkPath := filepath.Join(mapPath, string(linkName))
	...
	return os.Remove(linkPath)
}

func UnmapBlockVolume(...) error {
    ...
    unmapDeviceErr = blkUtil.UnmapDevice(globalUnmapPath, string(podUID), true /* bindMount */)
    ...
}

func (v VolumePathHandler) UnmapDevice(mapPath string, linkName string, bindMount bool) error {
    ...
    if bindMount {
        return unmapBindMountDevice(v, mapPath, linkName)
    }
    ...
}

func unmapBindMountDevice(v VolumePathHandler, mapPath string, linkName string) error {
    ...
    mounter := &mount.SafeFormatAndMount{Interface: mount.New(""), Exec: utilexec.New()}
    if err := mounter.Unmount(linkPath); err != nil {
        return fmt.Errorf("failed to unmount linkPath %s: %v", linkPath, err)
    }

    // Remove file
    if err := os.Remove(linkPath); err != nil && !os.IsNotExist(err) {
        return fmt.Errorf("failed to remove file %s: %v", linkPath, err)
    }

    return nil
}

func (mounter *Mounter) Unmount(target string) error {
    ...
    command := exec.Command("umount", target)
    output, err := command.CombinedOutput()
    ...
}