support local volume with block source reconstruction

pkg/kubelet/volumemanager/reconciler/reconciler.go

@@ -353,10 +353,10 @@ type reconstructedVolume struct {
 	volumeSpec          *volumepkg.Spec
 	outerVolumeSpecName string
 	pod                 *v1.Pod
-	attachablePlugin    volumepkg.AttachableVolumePlugin
 	volumeGidValue      string
 	devicePath          string
 	mounter             volumepkg.Mounter
+	deviceMounter       volumepkg.DeviceMounter
 	blockVolumeMapper   volumepkg.BlockVolumeMapper
 }
 
@@ -500,6 +500,7 @@ func (rc *reconciler) reconstructVolume(volume podVolume) (*reconstructedVolume,
 
 	var volumeMapper volumepkg.BlockVolumeMapper
 	var volumeMounter volumepkg.Mounter
+	var deviceMounter volumepkg.DeviceMounter
 	// Path to the mount or block device to check
 	var checkPath string
 
@@ -537,6 +538,17 @@ func (rc *reconciler) reconstructVolume(volume podVolume) (*reconstructedVolume,
 				err)
 		}
 		checkPath = volumeMounter.GetPath()
+		if deviceMountablePlugin != nil {
+			deviceMounter, err = deviceMountablePlugin.NewDeviceMounter()
+			if err != nil {
+				return nil, fmt.Errorf("reconstructVolume.NewDeviceMounter failed for volume %q (spec.Name: %q) pod %q (UID: %q) with: %v",
+					uniqueVolumeName,
+					volumeSpec.Name(),
+					volume.podName,
+					pod.UID,
+					err)
+			}
+		}
 	}
 
 	// Check existence of mount point for filesystem volume or symbolic link for block volume
@@ -558,7 +570,7 @@ func (rc *reconciler) reconstructVolume(volume podVolume) (*reconstructedVolume,
 		// TODO: in case pod is added back before reconciler starts to unmount, we can update this field from desired state information
 		outerVolumeSpecName: volume.volumeSpecName,
 		pod:                 pod,
-		attachablePlugin:    attachablePlugin,
+		deviceMounter:       deviceMounter,
 		volumeGidValue:      "",
 		// devicePath is updated during updateStates() by checking node status's VolumesAttached data.
 		// TODO: get device path directly from the volume mount path.
@@ -585,19 +597,18 @@ func (rc *reconciler) updateDevicePath(volumesNeedUpdate map[v1.UniqueVolumeName
 	}
 }
 
+// getDeviceMountPath returns device mount path for block volume which
+// implements BlockVolumeMapper or filesystem volume which implements
+// DeviceMounter
 func getDeviceMountPath(volume *reconstructedVolume) (string, error) {
 	if volume.blockVolumeMapper != nil {
 		// for block volume, we return its global map path
 		return volume.blockVolumeMapper.GetGlobalMapPath(volume.volumeSpec)
-	} else if volume.attachablePlugin != nil {
-		// for filesystem volume, we return its device mount path if the plugin implements AttachableVolumePlugin
-		volumeAttacher, err := volume.attachablePlugin.NewAttacher()
-		if volumeAttacher == nil || err != nil {
-			return "", err
-		}
-		return volumeAttacher.GetDeviceMountPath(volume.volumeSpec)
+	} else if volume.deviceMounter != nil {
+		// for filesystem volume, we return its device mount path if the plugin implements DeviceMounter
+		return volume.deviceMounter.GetDeviceMountPath(volume.volumeSpec)
 	} else {
-		return "", fmt.Errorf("blockVolumeMapper or attachablePlugin required")
+		return "", fmt.Errorf("blockVolumeMapper or deviceMounter required")
 	}
 }
 
@@ -628,7 +639,7 @@ func (rc *reconciler) updateStates(volumesNeedUpdate map[v1.UniqueVolumeName]*re
 		}
 		klog.V(4).Infof("Volume: %s (pod UID %s) is marked as mounted and added into the actual state", volume.volumeName, volume.podName)
 		// If the volume has device to mount, we mark its device as mounted.
-		if volume.attachablePlugin != nil || volume.blockVolumeMapper != nil {
+		if volume.deviceMounter != nil || volume.blockVolumeMapper != nil {
 			deviceMountPath, err := getDeviceMountPath(volume)
 			if err != nil {
 				klog.Errorf("Could not find device mount path for volume %s", volume.volumeName)

pkg/util/mount/mount.go

@@ -48,7 +48,7 @@ type Interface interface {
 	// most notably linux bind mounts and symbolic link.
 	IsLikelyNotMountPoint(file string) (bool, error)
 	// GetMountRefs finds all mount references to the path, returns a
-	// list of paths. Path could be a mountpoint path, device or a normal
+	// list of paths. Path could be a mountpoint or a normal
 	// directory (for bind mount).
 	GetMountRefs(pathname string) ([]string, error)
 }

pkg/util/mount/mount_linux.go

@@ -238,7 +238,7 @@ func (mounter *Mounter) IsLikelyNotMountPoint(file string) (bool, error) {
 }
 
 // GetMountRefs finds all mount references to pathname, returns a
-// list of paths. Path could be a mountpoint path, device or a normal
+// list of paths. Path could be a mountpoint or a normal
 // directory (for bind mount).
 func (mounter *Mounter) GetMountRefs(pathname string) ([]string, error) {
 	pathExists, pathErr := PathExists(pathname)
@@ -441,7 +441,8 @@ func parseProcMounts(content []byte) ([]MountPoint, error) {
 
 // SearchMountPoints finds all mount references to the source, returns a list of
 // mountpoints.
-// This function assumes source cannot be device.
+// The source can be a mount point or a normal directory (bind mount). We
+// didn't support device because there is no use case by now.
 // Some filesystems may share a source name, e.g. tmpfs. And for bind mounting,
 // it's possible to mount a non-root path of a filesystem, so we need to use
 // root path and major:minor to represent mount source uniquely.

pkg/volume/local/local.go

@@ -191,6 +191,32 @@ func (plugin *localVolumePlugin) NewBlockVolumeUnmapper(volName string,
 // TODO: check if no path and no topology constraints are ok
 func (plugin *localVolumePlugin) ConstructVolumeSpec(volumeName, mountPath string) (*volume.Spec, error) {
 	fs := v1.PersistentVolumeFilesystem
+	// The main purpose of reconstructed volume is to clean unused mount points
+	// and directories.
+	// For filesystem volume with directory source, no global mount path is
+	// needed to clean. Empty path is ok.
+	// For filesystem volume with block source, we should resolve to its device
+	// path if global mount path exists.
+	var path string
+	mounter := plugin.host.GetMounter(plugin.GetPluginName())
+	refs, err := mounter.GetMountRefs(mountPath)
+	if err != nil {
+		return nil, err
+	}
+	baseMountPath := plugin.generateBlockDeviceBaseGlobalPath()
+	for _, ref := range refs {
+		if mount.PathWithinBase(ref, baseMountPath) {
+			// If the global mount for block device exists, the source is block
+			// device.
+			// The resolved device path may not be the exact same as path in
+			// local PV object if symbolic link is used. However, it's the true
+			// source and can be used in reconstructed volume.
+			path, _, err = mount.GetDeviceNameFromMount(mounter, ref)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
 	localVolume := &v1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: volumeName,
@@ -198,7 +224,7 @@ func (plugin *localVolumePlugin) ConstructVolumeSpec(volumeName, mountPath strin
 		Spec: v1.PersistentVolumeSpec{
 			PersistentVolumeSource: v1.PersistentVolumeSource{
 				Local: &v1.LocalVolumeSource{
-					Path: "",
+					Path: path,
 				},
 			},
 			VolumeMode: &fs,

pkg/volume/local/local_test.go

@@ -422,9 +422,55 @@ func testFSGroupMount(plug volume.VolumePlugin, pod *v1.Pod, tmpDir string, fsGr
 }
 
 func TestConstructVolumeSpec(t *testing.T) {
-	tmpDir, plug := getPlugin(t)
-	defer os.RemoveAll(tmpDir)
+	tests := []struct {
+		name         string
+		mountPoints  []mount.MountPoint
+		expectedPath string
+	}{
+		{
+			name: "filesystem volume with directory source",
+			mountPoints: []mount.MountPoint{
+				{
+					Device: "/mnt/disk/ssd0",
+					Path:   "pods/poduid/volumes/kubernetes.io~local-volume/pvA",
+				},
+			},
+			expectedPath: "",
+		},
+		{
+			name: "filesystem volume with block source",
+			mountPoints: []mount.MountPoint{
+				{
+					Device: "/dev/loop0",
+					Path:   testMountPath,
+				},
+				{
+					Device: "/dev/loop0",
+					Path:   testBlockFormattingToFSGlobalPath,
+				},
+			},
+			expectedPath: "/dev/loop0",
+		},
+	}
 
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tmpDir, err := utiltesting.MkTmpdir("localVolumeTest")
+			if err != nil {
+				t.Fatalf("can't make a temp dir: %v", err)
+			}
+			defer os.RemoveAll(tmpDir)
+			plug := &localVolumePlugin{
+				host: volumetest.NewFakeVolumeHost(tmpDir, nil, nil),
+			}
+			mounter := plug.host.GetMounter(plug.GetPluginName())
+			fakeMountPoints := []mount.MountPoint{}
+			for _, mp := range tt.mountPoints {
+				fakeMountPoint := mp
+				fakeMountPoint.Path = filepath.Join(tmpDir, mp.Path)
+				fakeMountPoints = append(fakeMountPoints, fakeMountPoint)
+			}
+			mounter.(*mount.FakeMounter).MountPoints = fakeMountPoints
 			volPath := filepath.Join(tmpDir, testMountPath)
 			spec, err := plug.ConstructVolumeSpec(testPVName, volPath)
 			if err != nil {
@@ -460,6 +506,13 @@ func TestConstructVolumeSpec(t *testing.T) {
 			if ls == nil {
 				t.Fatalf("LocalVolumeSource object nil")
 			}
+
+			if pv.Spec.PersistentVolumeSource.Local.Path != tt.expectedPath {
+				t.Fatalf("Unexpected path got %q, expected %q", pv.Spec.PersistentVolumeSource.Local.Path, tt.expectedPath)
+			}
+		})
+	}
+
 }
 
 func TestConstructBlockVolumeSpec(t *testing.T) {