Merge pull request #59303 from dhirajh/localblock

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Block Volume Support: Local Volume Plugin update

**What this PR does / why we need it**:

Introduce block volume support to local volumes plugin.

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes #59500

**Special notes for your reviewer**:
@msau42 @mtanino @ianchakeres 

Adding support for block volumes as per https://github.com/kubernetes/features/issues/121

Other related PRs:
(#50457) API Change
(#53385) VolumeMode PV-PVC Binding change
(#51494) Container runtime interface change, volumemanager changes, operationexecutor changes

**Release note**:
```
Added support for Block Volume type to local-volume plugin.
```

pkg/volume/local/local.go

@@ -19,6 +19,7 @@ package local
 import (
 	"fmt"
 	"os"
+	"path"
 
 	"github.com/golang/glog"
 
@@ -49,6 +50,7 @@ type localVolumePlugin struct {
 
 var _ volume.VolumePlugin = &localVolumePlugin{}
 var _ volume.PersistentVolumePlugin = &localVolumePlugin{}
+var _ volume.BlockVolumePlugin = &localVolumePlugin{}
 
 const (
 	localVolumePluginName = "kubernetes.io/local-volume"
@@ -137,6 +139,36 @@ func (plugin *localVolumePlugin) NewUnmounter(volName string, podUID types.UID)
 	}, nil
 }
 
+func (plugin *localVolumePlugin) NewBlockVolumeMapper(spec *volume.Spec, pod *v1.Pod,
+	_ volume.VolumeOptions) (volume.BlockVolumeMapper, error) {
+	volumeSource, readOnly, err := getVolumeSource(spec)
+	if err != nil {
+		return nil, err
+	}
+
+	return &localVolumeMapper{
+		localVolume: &localVolume{
+			podUID:     pod.UID,
+			volName:    spec.Name(),
+			globalPath: volumeSource.Path,
+			plugin:     plugin,
+		},
+		readOnly: readOnly,
+	}, nil
+
+}
+
+func (plugin *localVolumePlugin) NewBlockVolumeUnmapper(volName string,
+	podUID types.UID) (volume.BlockVolumeUnmapper, error) {
+	return &localVolumeUnmapper{
+		localVolume: &localVolume{
+			podUID:  podUID,
+			volName: volName,
+			plugin:  plugin,
+		},
+	}, nil
+}
+
 // TODO: check if no path and no topology constraints are ok
 func (plugin *localVolumePlugin) ConstructVolumeSpec(volumeName, mountPath string) (*volume.Spec, error) {
 	localVolume := &v1.PersistentVolume{
@@ -154,6 +186,27 @@ func (plugin *localVolumePlugin) ConstructVolumeSpec(volumeName, mountPath strin
 	return volume.NewSpecFromPersistentVolume(localVolume, false), nil
 }
 
+func (plugin *localVolumePlugin) ConstructBlockVolumeSpec(podUID types.UID, volumeName,
+	mapPath string) (*volume.Spec, error) {
+	block := v1.PersistentVolumeBlock
+
+	localVolume := &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: volumeName,
+		},
+		Spec: v1.PersistentVolumeSpec{
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				Local: &v1.LocalVolumeSource{
+					Path: "",
+				},
+			},
+			VolumeMode: &block,
+		},
+	}
+
+	return volume.NewSpecFromPersistentVolume(localVolume, false), nil
+}
+
 // Local volumes represent a local directory on a node.
 // The directory at the globalPath will be bind-mounted to the pod's directory
 type localVolume struct {
@@ -307,3 +360,45 @@ func (u *localVolumeUnmounter) TearDownAt(dir string) error {
 	glog.V(4).Infof("Unmounting volume %q at path %q\n", u.volName, dir)
 	return util.UnmountMountPoint(dir, u.mounter, true) /* extensiveMountPointCheck = true */
 }
+
+// localVolumeMapper implements the BlockVolumeMapper interface for local volumes.
+type localVolumeMapper struct {
+	*localVolume
+	readOnly bool
+}
+
+var _ volume.BlockVolumeMapper = &localVolumeMapper{}
+
+// SetUpDevice provides physical device path for the local PV.
+func (m *localVolumeMapper) SetUpDevice() (string, error) {
+	glog.V(4).Infof("SetupDevice returning path %s", m.globalPath)
+	return m.globalPath, nil
+}
+
+// localVolumeUnmapper implements the BlockVolumeUnmapper interface for local volumes.
+type localVolumeUnmapper struct {
+	*localVolume
+}
+
+var _ volume.BlockVolumeUnmapper = &localVolumeUnmapper{}
+
+// TearDownDevice will undo SetUpDevice procedure. In local PV, all of this already handled by operation_generator.
+func (u *localVolumeUnmapper) TearDownDevice(mapPath, devicePath string) error {
+	glog.V(4).Infof("local: TearDownDevice completed for: %s", mapPath)
+	return nil
+}
+
+// GetGlobalMapPath returns global map path and error.
+// path: plugins/kubernetes.io/kubernetes.io/local-volume/volumeDevices/{volumeName}
+func (lv *localVolume) GetGlobalMapPath(spec *volume.Spec) (string, error) {
+	return path.Join(lv.plugin.host.GetVolumeDevicePluginDir(strings.EscapeQualifiedNameForDisk(localVolumePluginName)),
+		lv.volName), nil
+}
+
+// GetPodDeviceMapPath returns pod device map path and volume name.
+// path: pods/{podUid}/volumeDevices/kubernetes.io~local-volume
+// volName: local-pv-ff0d6d4
+func (lv *localVolume) GetPodDeviceMapPath() (string, string) {
+	return lv.plugin.host.GetPodVolumeDeviceDir(lv.podUID,
+		strings.EscapeQualifiedNameForDisk(localVolumePluginName)), lv.volName
+}

pkg/volume/local/local_test.go

@@ -32,9 +32,11 @@ import (
 )
 
 const (
-	testPVName    = "pvA"
+	testPVName     = "pvA"
-	testMountPath = "pods/poduid/volumes/kubernetes.io~local-volume/pvA"
+	testMountPath  = "pods/poduid/volumes/kubernetes.io~local-volume/pvA"
-	testNodeName  = "fakeNodeName"
+	testGlobalPath = "plugins/kubernetes.io~local-volume/volumeDevices/pvA"
+	testPodPath    = "pods/poduid/volumeDevices/kubernetes.io~local-volume"
+	testNodeName   = "fakeNodeName"
 )
 
 func getPlugin(t *testing.T) (string, volume.VolumePlugin) {
@@ -57,6 +59,25 @@ func getPlugin(t *testing.T) (string, volume.VolumePlugin) {
 	return tmpDir, plug
 }
 
+func getBlockPlugin(t *testing.T) (string, volume.BlockVolumePlugin) {
+	tmpDir, err := utiltesting.MkTmpdir("localVolumeTest")
+	if err != nil {
+		t.Fatalf("can't make a temp dir: %v", err)
+	}
+
+	plugMgr := volume.VolumePluginMgr{}
+	plugMgr.InitPlugins(ProbeVolumePlugins(), nil /* prober */, volumetest.NewFakeVolumeHost(tmpDir, nil, nil))
+	plug, err := plugMgr.FindMapperPluginByName(localVolumePluginName)
+	if err != nil {
+		os.RemoveAll(tmpDir)
+		t.Fatalf("Can't find the plugin by name: %q", localVolumePluginName)
+	}
+	if plug.GetPluginName() != localVolumePluginName {
+		t.Errorf("Wrong name: %s", plug.GetPluginName())
+	}
+	return tmpDir, plug
+}
+
 func getPersistentPlugin(t *testing.T) (string, volume.PersistentVolumePlugin) {
 	tmpDir, err := utiltesting.MkTmpdir("localVolumeTest")
 	if err != nil {
@@ -77,7 +98,7 @@ func getPersistentPlugin(t *testing.T) (string, volume.PersistentVolumePlugin) {
 	return tmpDir, plug
 }
 
-func getTestVolume(readOnly bool, path string) *volume.Spec {
+func getTestVolume(readOnly bool, path string, isBlock bool) *volume.Spec {
 	pv := &v1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: testPVName,
@@ -90,6 +111,11 @@ func getTestVolume(readOnly bool, path string) *volume.Spec {
 			},
 		},
 	}
+
+	if isBlock {
+		blockMode := v1.PersistentVolumeBlock
+		pv.Spec.VolumeMode = &blockMode
+	}
 	return volume.NewSpecFromPersistentVolume(pv, readOnly)
 }
 
@@ -97,7 +123,7 @@ func TestCanSupport(t *testing.T) {
 	tmpDir, plug := getPlugin(t)
 	defer os.RemoveAll(tmpDir)
 
-	if !plug.CanSupport(getTestVolume(false, tmpDir)) {
+	if !plug.CanSupport(getTestVolume(false, tmpDir, false)) {
 		t.Errorf("Expected true")
 	}
 }
@@ -123,7 +149,7 @@ func TestGetVolumeName(t *testing.T) {
 	tmpDir, plug := getPersistentPlugin(t)
 	defer os.RemoveAll(tmpDir)
 
-	volName, err := plug.GetVolumeName(getTestVolume(false, tmpDir))
+	volName, err := plug.GetVolumeName(getTestVolume(false, tmpDir, false))
 	if err != nil {
 		t.Errorf("Failed to get volume name: %v", err)
 	}
@@ -137,7 +163,7 @@ func TestInvalidLocalPath(t *testing.T) {
 	defer os.RemoveAll(tmpDir)
 
 	pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: types.UID("poduid")}}
-	mounter, err := plug.NewMounter(getTestVolume(false, "/no/backsteps/allowed/.."), pod, volume.VolumeOptions{})
+	mounter, err := plug.NewMounter(getTestVolume(false, "/no/backsteps/allowed/..", false), pod, volume.VolumeOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -154,7 +180,7 @@ func TestMountUnmount(t *testing.T) {
 	defer os.RemoveAll(tmpDir)
 
 	pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: types.UID("poduid")}}
-	mounter, err := plug.NewMounter(getTestVolume(false, tmpDir), pod, volume.VolumeOptions{})
+	mounter, err := plug.NewMounter(getTestVolume(false, tmpDir, false), pod, volume.VolumeOptions{})
 	if err != nil {
 		t.Errorf("Failed to make a new Mounter: %v", err)
 	}
@@ -197,8 +223,64 @@ func TestMountUnmount(t *testing.T) {
 	}
 }
 
+// TestMapUnmap tests block map and unmap interfaces.
+func TestMapUnmap(t *testing.T) {
+	tmpDir, plug := getBlockPlugin(t)
+	defer os.RemoveAll(tmpDir)
+
+	pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: types.UID("poduid")}}
+	volSpec := getTestVolume(false, tmpDir, true /*isBlock*/)
+	mapper, err := plug.NewBlockVolumeMapper(volSpec, pod, volume.VolumeOptions{})
+	if err != nil {
+		t.Errorf("Failed to make a new Mounter: %v", err)
+	}
+	if mapper == nil {
+		t.Fatalf("Got a nil Mounter")
+	}
+
+	expectedGlobalPath := path.Join(tmpDir, testGlobalPath)
+	globalPath, err := mapper.GetGlobalMapPath(volSpec)
+	if err != nil {
+		t.Errorf("Failed to get global path: %v", err)
+	}
+	if globalPath != expectedGlobalPath {
+		t.Errorf("Got unexpected path: %s, expected %s", globalPath, expectedGlobalPath)
+	}
+	expectedPodPath := path.Join(tmpDir, testPodPath)
+	podPath, volName := mapper.GetPodDeviceMapPath()
+	if podPath != expectedPodPath {
+		t.Errorf("Got unexpected pod path: %s, expected %s", podPath, expectedPodPath)
+	}
+	if volName != testPVName {
+		t.Errorf("Got unexpected volNamne: %s, expected %s", volName, testPVName)
+	}
+	devPath, err := mapper.SetUpDevice()
+	if err != nil {
+		t.Errorf("Failed to SetUpDevice, err: %v", err)
+	}
+	if _, err := os.Stat(devPath); err != nil {
+		if os.IsNotExist(err) {
+			t.Errorf("SetUpDevice() failed, volume path not created: %s", devPath)
+		} else {
+			t.Errorf("SetUpDevice() failed: %v", err)
+		}
+	}
+
+	unmapper, err := plug.NewBlockVolumeUnmapper(testPVName, pod.UID)
+	if err != nil {
+		t.Fatalf("Failed to make a new Unmapper: %v", err)
+	}
+	if unmapper == nil {
+		t.Fatalf("Got a nil Unmapper")
+	}
+
+	if err := unmapper.TearDownDevice(globalPath, devPath); err != nil {
+		t.Errorf("TearDownDevice failed, err: %v", err)
+	}
+}
+
 func testFSGroupMount(plug volume.VolumePlugin, pod *v1.Pod, tmpDir string, fsGroup int64) error {
-	mounter, err := plug.NewMounter(getTestVolume(false, tmpDir), pod, volume.VolumeOptions{})
+	mounter, err := plug.NewMounter(getTestVolume(false, tmpDir, false), pod, volume.VolumeOptions{})
 	if err != nil {
 		return err
 	}
@@ -290,13 +372,54 @@ func TestConstructVolumeSpec(t *testing.T) {
 	}
 }
 
+func TestConstructBlockVolumeSpec(t *testing.T) {
+	tmpDir, plug := getBlockPlugin(t)
+	defer os.RemoveAll(tmpDir)
+
+	podPath := path.Join(tmpDir, testPodPath)
+	spec, err := plug.ConstructBlockVolumeSpec(types.UID("poduid"), testPVName, podPath)
+	if err != nil {
+		t.Errorf("ConstructBlockVolumeSpec() failed: %v", err)
+	}
+	if spec == nil {
+		t.Fatalf("ConstructBlockVolumeSpec() returned nil")
+	}
+
+	volName := spec.Name()
+	if volName != testPVName {
+		t.Errorf("Expected volume name %q, got %q", testPVName, volName)
+	}
+
+	if spec.Volume != nil {
+		t.Errorf("Volume object returned, expected nil")
+	}
+
+	pv := spec.PersistentVolume
+	if pv == nil {
+		t.Fatalf("PersistentVolume object nil")
+	}
+
+	if spec.PersistentVolume.Spec.VolumeMode == nil {
+		t.Fatalf("Volume mode has not been set.")
+	}
+
+	if *spec.PersistentVolume.Spec.VolumeMode != v1.PersistentVolumeBlock {
+		t.Errorf("Unexpected volume mode %q", *spec.PersistentVolume.Spec.VolumeMode)
+	}
+
+	ls := pv.Spec.PersistentVolumeSource.Local
+	if ls == nil {
+		t.Fatalf("LocalVolumeSource object nil")
+	}
+}
+
 func TestPersistentClaimReadOnlyFlag(t *testing.T) {
 	tmpDir, plug := getPlugin(t)
 	defer os.RemoveAll(tmpDir)
 
 	// Read only == true
 	pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: types.UID("poduid")}}
-	mounter, err := plug.NewMounter(getTestVolume(true, tmpDir), pod, volume.VolumeOptions{})
+	mounter, err := plug.NewMounter(getTestVolume(true, tmpDir, false), pod, volume.VolumeOptions{})
 	if err != nil {
 		t.Errorf("Failed to make a new Mounter: %v", err)
 	}
@@ -308,7 +431,7 @@ func TestPersistentClaimReadOnlyFlag(t *testing.T) {
 	}
 
 	// Read only == false
-	mounter, err = plug.NewMounter(getTestVolume(false, tmpDir), pod, volume.VolumeOptions{})
+	mounter, err = plug.NewMounter(getTestVolume(false, tmpDir, false), pod, volume.VolumeOptions{})
 	if err != nil {
 		t.Errorf("Failed to make a new Mounter: %v", err)
 	}
@@ -329,7 +452,7 @@ func TestUnsupportedPlugins(t *testing.T) {
 
 	plugMgr := volume.VolumePluginMgr{}
 	plugMgr.InitPlugins(ProbeVolumePlugins(), nil /* prober */, volumetest.NewFakeVolumeHost(tmpDir, nil, nil))
-	spec := getTestVolume(false, tmpDir)
+	spec := getTestVolume(false, tmpDir, false)
 
 	recyclePlug, err := plugMgr.FindRecyclablePluginBySpec(spec)
 	if err == nil && recyclePlug != nil {

pkg/volume/testing/testing.go

@@ -89,8 +89,8 @@ func (f *fakeVolumeHost) GetPluginDir(podUID string) string {
 	return path.Join(f.rootDir, "plugins", podUID)
 }
 
-func (f *fakeVolumeHost) GetVolumeDevicePluginDir(podUID string) string {
+func (f *fakeVolumeHost) GetVolumeDevicePluginDir(pluginName string) string {
-	return path.Join(f.rootDir, "plugins", podUID)
+	return path.Join(f.rootDir, "plugins", pluginName, "volumeDevices")
 }
 
 func (f *fakeVolumeHost) GetPodVolumeDir(podUID types.UID, pluginName, volumeName string) string {