BlockVolumesSupport: CRI, VolumeManager and OperationExecutor changes

pkg/controller/volume/attachdetach/attach_detach_controller.go

@@ -45,6 +45,7 @@ import (
 	"k8s.io/kubernetes/pkg/util/io"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
+	volumeutil "k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
 	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
 )
@@ -136,6 +137,7 @@ func NewAttachDetachController(
 	eventBroadcaster.StartLogging(glog.Infof)
 	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: v1core.New(kubeClient.CoreV1().RESTClient()).Events("")})
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "attachdetach-controller"})
+	blkutil := volumeutil.NewBlockVolumePathHandler()
 
 	adc.desiredStateOfWorld = cache.NewDesiredStateOfWorld(&adc.volumePluginMgr)
 	adc.actualStateOfWorld = cache.NewActualStateOfWorld(&adc.volumePluginMgr)
@@ -144,7 +146,8 @@ func NewAttachDetachController(
 			kubeClient,
 			&adc.volumePluginMgr,
 			recorder,
-			false)) // flag for experimental binary check for volume mount
+			false, // flag for experimental binary check for volume mount
+			blkutil))
 	adc.nodeStatusUpdater = statusupdater.NewNodeStatusUpdater(
 		kubeClient, nodeInformer.Lister(), adc.actualStateOfWorld)
 
@@ -515,6 +518,10 @@ func (adc *attachDetachController) GetPluginDir(podUID string) string {
 	return ""
 }
 
+func (adc *attachDetachController) GetVolumeDevicePluginDir(podUID string) string {
+	return ""
+}
+
 func (adc *attachDetachController) GetPodVolumeDir(podUID types.UID, pluginName, volumeName string) string {
 	return ""
 }
@@ -523,6 +530,10 @@ func (adc *attachDetachController) GetPodPluginDir(podUID types.UID, pluginName
 	return ""
 }
 
+func (adc *attachDetachController) GetPodVolumeDeviceDir(podUID types.UID, pluginName string) string {
+	return ""
+}
+
 func (adc *attachDetachController) GetKubeClient() clientset.Interface {
 	return adc.kubeClient
 }

pkg/controller/volume/attachdetach/reconciler/reconciler_test.go

@@ -50,7 +50,13 @@ func Test_Run_Positive_DoNothing(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
+	fakeHandler := volumetesting.NewBlockVolumePathHandler()
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
+		fakeKubeClient,
+		volumePluginMgr,
+		fakeRecorder,
+		false, /* checkNodeCapabilitiesBeforeMount */
+		fakeHandler))
 	informerFactory := informers.NewSharedInformerFactory(fakeKubeClient, controller.NoResyncPeriodFunc())
 	nsu := statusupdater.NewNodeStatusUpdater(
 		fakeKubeClient, informerFactory.Core().V1().Nodes().Lister(), asw)
@@ -80,7 +86,13 @@ func Test_Run_Positive_OneDesiredVolumeAttach(t *testing.T) {
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
+	fakeHandler := volumetesting.NewBlockVolumePathHandler()
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
+		fakeKubeClient,
+		volumePluginMgr,
+		fakeRecorder,
+		false, /* checkNodeCapabilitiesBeforeMount */
+		fakeHandler))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, false, dsw, asw, ad, nsu, fakeRecorder)
@@ -126,7 +138,13 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithUnmountedVolume(t *te
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
+	fakeHandler := volumetesting.NewBlockVolumePathHandler()
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
+		fakeKubeClient,
+		volumePluginMgr,
+		fakeRecorder,
+		false, /* checkNodeCapabilitiesBeforeMount */
+		fakeHandler))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, false, dsw, asw, ad, nsu, fakeRecorder)
@@ -193,7 +211,13 @@ func Test_Run_Positive_OneDesiredVolumeAttachThenDetachWithMountedVolume(t *test
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
+	fakeHandler := volumetesting.NewBlockVolumePathHandler()
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
+		fakeKubeClient,
+		volumePluginMgr,
+		fakeRecorder,
+		false, /* checkNodeCapabilitiesBeforeMount */
+		fakeHandler))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, false, dsw, asw, ad, nsu, fakeRecorder)
@@ -260,7 +284,13 @@ func Test_Run_Negative_OneDesiredVolumeAttachThenDetachWithUnmountedVolumeUpdate
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
+	fakeHandler := volumetesting.NewBlockVolumePathHandler()
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
+		fakeKubeClient,
+		volumePluginMgr,
+		fakeRecorder,
+		false, /* checkNodeCapabilitiesBeforeMount */
+		fakeHandler))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(true /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, false, dsw, asw, ad, nsu, fakeRecorder)
@@ -330,7 +360,13 @@ func Test_Run_OneVolumeAttachAndDetachMultipleNodesWithReadWriteMany(t *testing.
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
+	fakeHandler := volumetesting.NewBlockVolumePathHandler()
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
+		fakeKubeClient,
+		volumePluginMgr,
+		fakeRecorder,
+		false, /* checkNodeCapabilitiesBeforeMount */
+		fakeHandler))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, false, dsw, asw, ad, nsu, fakeRecorder)
@@ -416,7 +452,13 @@ func Test_Run_OneVolumeAttachAndDetachMultipleNodesWithReadWriteOnce(t *testing.
 	asw := cache.NewActualStateOfWorld(volumePluginMgr)
 	fakeKubeClient := controllervolumetesting.CreateTestClient()
 	fakeRecorder := &record.FakeRecorder{}
-	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(fakeKubeClient, volumePluginMgr, fakeRecorder, false /* checkNodeCapabilitiesBeforeMount */))
+	fakeHandler := volumetesting.NewBlockVolumePathHandler()
+	ad := operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
+		fakeKubeClient,
+		volumePluginMgr,
+		fakeRecorder,
+		false, /* checkNodeCapabilitiesBeforeMount */
+		fakeHandler))
 	nsu := statusupdater.NewFakeNodeStatusUpdater(false /* returnError */)
 	reconciler := NewReconciler(
 		reconcilerLoopPeriod, maxWaitForUnmountDuration, syncLoopPeriod, false, dsw, asw, ad, nsu, fakeRecorder)

pkg/controller/volume/expand/expand_controller.go

@@ -42,6 +42,7 @@ import (
 	"k8s.io/kubernetes/pkg/util/io"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
 )
 
@@ -117,12 +118,14 @@ func NewExpandController(
 	eventBroadcaster.StartLogging(glog.Infof)
 	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: v1core.New(kubeClient.CoreV1().RESTClient()).Events("")})
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "volume_expand"})
+	blkutil := util.NewBlockVolumePathHandler()
 
 	expc.opExecutor = operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
 		kubeClient,
 		&expc.volumePluginMgr,
 		recorder,
-		false))
+		false,
+		blkutil))
 
 	expc.resizeMap = cache.NewVolumeResizeMap(expc.kubeClient)
 
@@ -203,10 +206,18 @@ func (expc *expandController) GetPluginDir(pluginName string) string {
 	return ""
 }
 
+func (expc *expandController) GetVolumeDevicePluginDir(pluginName string) string {
+	return ""
+}
+
 func (expc *expandController) GetPodVolumeDir(podUID types.UID, pluginName string, volumeName string) string {
 	return ""
 }
 
+func (expc *expandController) GetPodVolumeDeviceDir(podUID types.UID, pluginName string) string {
+	return ""
+}
+
 func (expc *expandController) GetPodPluginDir(podUID types.UID, pluginName string) string {
 	return ""
 }

pkg/controller/volume/persistentvolume/volume_host.go

@@ -37,6 +37,10 @@ func (ctrl *PersistentVolumeController) GetPluginDir(pluginName string) string {
 	return ""
 }
 
+func (ctrl *PersistentVolumeController) GetVolumeDevicePluginDir(pluginName string) string {
+	return ""
+}
+
 func (ctrl *PersistentVolumeController) GetPodVolumeDir(podUID types.UID, pluginName string, volumeName string) string {
 	return ""
 }
@@ -45,6 +49,10 @@ func (ctrl *PersistentVolumeController) GetPodPluginDir(podUID types.UID, plugin
 	return ""
 }
 
+func (ctrl *PersistentVolumeController) GetPodVolumeDeviceDir(ppodUID types.UID, pluginName string) string {
+	return ""
+}
+
 func (ctrl *PersistentVolumeController) GetKubeClient() clientset.Interface {
 	return ctrl.kubeClient
 }

pkg/kubelet/config/defaults.go

@@ -19,6 +19,7 @@ package config
 const (
 	DefaultKubeletPodsDirName             = "pods"
 	DefaultKubeletVolumesDirName          = "volumes"
+	DefaultKubeletVolumeDevicesDirName    = "volumeDevices"
 	DefaultKubeletPluginsDirName          = "plugins"
 	DefaultKubeletContainersDirName       = "containers"
 	DefaultKubeletPluginContainersDirName = "plugin-containers"

pkg/kubelet/container/runtime.go

@@ -446,9 +446,14 @@ type RunContainerOptions struct {
 type VolumeInfo struct {
 	// Mounter is the volume's mounter
 	Mounter volume.Mounter
+	// BlockVolumeMapper is the Block volume's mapper
+	BlockVolumeMapper volume.BlockVolumeMapper
 	// SELinuxLabeled indicates whether this volume has had the
 	// pod's SELinux label applied to it or not
 	SELinuxLabeled bool
+	// Whether the volume permission is set to read-only or not
+	// This value is passed from volume.spec
+	ReadOnly bool
 }
 
 type VolumeMap map[string]VolumeInfo

pkg/kubelet/events/event.go

@@ -53,6 +53,8 @@ const (
 	FailedMountVolume                    = "FailedMount"
 	VolumeResizeFailed                   = "VolumeResizeFailed"
 	FailedUnMountVolume                  = "FailedUnMount"
+	FailedMapVolume                      = "FailedMapVolume"
+	FailedUnmapDevice                    = "FailedUnmapDevice"
 	WarnAlreadyMountedVolume             = "AlreadyMountedVolume"
 	SuccessfulDetachVolume               = "SuccessfulDetachVolume"
 	SuccessfulMountVolume                = "SuccessfulMountVolume"

pkg/kubelet/kubelet_getters.go

@@ -64,6 +64,21 @@ func (kl *Kubelet) getPluginDir(pluginName string) string {
 	return filepath.Join(kl.getPluginsDir(), pluginName)
 }
 
+// getVolumeDevicePluginsDir returns the full path to the directory under which plugin
+// directories are created.  Plugins can use these directories for data that
+// they need to persist.  Plugins should create subdirectories under this named
+// after their own names.
+func (kl *Kubelet) getVolumeDevicePluginsDir() string {
+	return filepath.Join(kl.getRootDir(), config.DefaultKubeletPluginsDirName)
+}
+
+// getVolumeDevicePluginDir returns a data directory name for a given plugin name.
+// Plugins can use these directories to store data that they need to persist.
+// For per-pod plugin data, see getVolumeDevicePluginsDir.
+func (kl *Kubelet) getVolumeDevicePluginDir(pluginName string) string {
+	return filepath.Join(kl.getVolumeDevicePluginsDir(), pluginName, config.DefaultKubeletVolumeDevicesDirName)
+}
+
 // GetPodDir returns the full path to the per-pod data directory for the
 // specified pod. This directory may not exist if the pod does not exist.
 func (kl *Kubelet) GetPodDir(podUID types.UID) string {
@@ -90,6 +105,19 @@ func (kl *Kubelet) getPodVolumeDir(podUID types.UID, pluginName string, volumeNa
 	return filepath.Join(kl.getPodVolumesDir(podUID), pluginName, volumeName)
 }
 
+// getPodVolumeDevicesDir returns the full path to the per-pod data directory under
+// which volumes are created for the specified pod. This directory may not
+// exist if the pod does not exist.
+func (kl *Kubelet) getPodVolumeDevicesDir(podUID types.UID) string {
+	return filepath.Join(kl.getPodDir(podUID), config.DefaultKubeletVolumeDevicesDirName)
+}
+
+// getPodVolumeDeviceDir returns the full path to the directory which represents the
+// named plugin for specified pod. This directory may not exist if the pod does not exist.
+func (kl *Kubelet) getPodVolumeDeviceDir(podUID types.UID, pluginName string) string {
+	return filepath.Join(kl.getPodVolumeDevicesDir(podUID), pluginName)
+}
+
 // getPodPluginsDir returns the full path to the per-pod data directory under
 // which plugins may store data for the specified pod.  This directory may not
 // exist if the pod does not exist.

pkg/kubelet/kubelet_pods.go

@@ -64,6 +64,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/util/format"
 	utilfile "k8s.io/kubernetes/pkg/util/file"
 	"k8s.io/kubernetes/pkg/volume"
+	volumeutil "k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
 	volumevalidation "k8s.io/kubernetes/pkg/volume/validation"
 	"k8s.io/kubernetes/third_party/forked/golang/expansion"
@@ -91,9 +92,9 @@ func (kl *Kubelet) GetActivePods() []*v1.Pod {
 	return activePods
 }
 
-// makeDevices determines the devices for the given container.
+// makeGPUDevices determines the devices for the given container.
 // Experimental.
-func (kl *Kubelet) makeDevices(pod *v1.Pod, container *v1.Container) ([]kubecontainer.DeviceInfo, error) {
+func (kl *Kubelet) makeGPUDevices(pod *v1.Pod, container *v1.Container) ([]kubecontainer.DeviceInfo, error) {
 	if container.Resources.Limits.NvidiaGPU().IsZero() {
 		return nil, nil
 	}
@@ -128,6 +129,39 @@ func makeAbsolutePath(goos, path string) string {
 	return "c:\\" + path
 }
 
+// makeBlockVolumes maps the raw block devices specified in the path of the container
+// Experimental
+func (kl *Kubelet) makeBlockVolumes(pod *v1.Pod, container *v1.Container, podVolumes kubecontainer.VolumeMap, blkutil volumeutil.BlockVolumePathHandler) ([]kubecontainer.DeviceInfo, error) {
+	var devices []kubecontainer.DeviceInfo
+	for _, device := range container.VolumeDevices {
+		// check path is absolute
+		if !filepath.IsAbs(device.DevicePath) {
+			return nil, fmt.Errorf("error DevicePath `%s` must be an absolute path", device.DevicePath)
+		}
+		vol, ok := podVolumes[device.Name]
+		if !ok || vol.BlockVolumeMapper == nil {
+			glog.Errorf("Block volume cannot be satisfied for container %q, because the volume is missing or the volume mapper is nil: %+v", container.Name, device)
+			return nil, fmt.Errorf("cannot find volume %q to pass into container %q", device.Name, container.Name)
+		}
+		// Get a symbolic link associated to a block device under pod device path
+		dirPath, volName := vol.BlockVolumeMapper.GetPodDeviceMapPath()
+		symlinkPath := path.Join(dirPath, volName)
+		if islinkExist, checkErr := blkutil.IsSymlinkExist(symlinkPath); checkErr != nil {
+			return nil, checkErr
+		} else if islinkExist {
+			// Check readOnly in PVCVolumeSource and set read only permission if it's true.
+			permission := "mrw"
+			if vol.ReadOnly {
+				permission = "r"
+			}
+			glog.V(4).Infof("Device will be attached to container %q. Path on host: %v", container.Name, symlinkPath)
+			devices = append(devices, kubecontainer.DeviceInfo{PathOnHost: symlinkPath, PathInContainer: device.DevicePath, Permissions: permission})
+		}
+	}
+
+	return devices, nil
+}
+
 // makeMounts determines the mount points for the given container.
 func makeMounts(pod *v1.Pod, podDir string, container *v1.Container, hostName, hostDomain, podIP string, podVolumes kubecontainer.VolumeMap) ([]kubecontainer.Mount, error) {
 	// Kubernetes only mounts on /etc/hosts if:
@@ -416,12 +450,22 @@ func (kl *Kubelet) GenerateRunContainerOptions(pod *v1.Pod, container *v1.Contai
 
 	opts.PortMappings = kubecontainer.MakePortMappings(container)
 	// TODO(random-liu): Move following convert functions into pkg/kubelet/container
-	devices, err := kl.makeDevices(pod, container)
+	devices, err := kl.makeGPUDevices(pod, container)
 	if err != nil {
 		return nil, err
 	}
 	opts.Devices = append(opts.Devices, devices...)
 
+	// TODO: remove feature gate check after no longer needed
+	if utilfeature.DefaultFeatureGate.Enabled(features.BlockVolume) {
+		blkutil := volumeutil.NewBlockVolumePathHandler()
+		blkVolumes, err := kl.makeBlockVolumes(pod, container, volumes, blkutil)
+		if err != nil {
+			return nil, err
+		}
+		opts.Devices = append(opts.Devices, blkVolumes...)
+	}
+
 	mounts, err := makeMounts(pod, kl.getPodDir(pod.UID), container, hostname, hostDomainName, podIP, volumes)
 	if err != nil {
 		return nil, err

pkg/kubelet/kubelet_pods_test.go

@@ -47,6 +47,7 @@ import (
 	containertest "k8s.io/kubernetes/pkg/kubelet/container/testing"
 	"k8s.io/kubernetes/pkg/kubelet/server/portforward"
 	"k8s.io/kubernetes/pkg/kubelet/server/remotecommand"
+	volumetest "k8s.io/kubernetes/pkg/volume/testing"
 )
 
 func TestMakeAbsolutePath(t *testing.T) {
@@ -352,6 +353,139 @@ func TestMakeMounts(t *testing.T) {
 	}
 }
 
+func TestMakeBlockVolumes(t *testing.T) {
+	testKubelet := newTestKubelet(t, false /* controllerAttachDetachEnabled */)
+	defer testKubelet.Cleanup()
+	kubelet := testKubelet.kubelet
+	testCases := map[string]struct {
+		container       v1.Container
+		podVolumes      kubecontainer.VolumeMap
+		expectErr       bool
+		expectedErrMsg  string
+		expectedDevices []kubecontainer.DeviceInfo
+	}{
+		"valid volumeDevices in container": {
+			podVolumes: kubecontainer.VolumeMap{
+				"disk1": kubecontainer.VolumeInfo{BlockVolumeMapper: &stubBlockVolume{dirPath: "/dev/", volName: "sda"}},
+				"disk2": kubecontainer.VolumeInfo{BlockVolumeMapper: &stubBlockVolume{dirPath: "/dev/disk/by-path/", volName: "diskPath"}, ReadOnly: true},
+				"disk3": kubecontainer.VolumeInfo{BlockVolumeMapper: &stubBlockVolume{dirPath: "/dev/disk/by-id/", volName: "diskUuid"}},
+				"disk4": kubecontainer.VolumeInfo{BlockVolumeMapper: &stubBlockVolume{dirPath: "/var/lib/", volName: "rawdisk"}, ReadOnly: true},
+			},
+			container: v1.Container{
+				Name: "container1",
+				VolumeDevices: []v1.VolumeDevice{
+					{
+						DevicePath: "/dev/sda",
+						Name:       "disk1",
+					},
+					{
+						DevicePath: "/dev/xvda",
+						Name:       "disk2",
+					},
+					{
+						DevicePath: "/dev/xvdb",
+						Name:       "disk3",
+					},
+					{
+						DevicePath: "/mnt/rawdisk",
+						Name:       "disk4",
+					},
+				},
+			},
+			expectedDevices: []kubecontainer.DeviceInfo{
+				{
+					PathInContainer: "/dev/sda",
+					PathOnHost:      "/dev/sda",
+					Permissions:     "mrw",
+				},
+				{
+					PathInContainer: "/dev/xvda",
+					PathOnHost:      "/dev/disk/by-path/diskPath",
+					Permissions:     "r",
+				},
+				{
+					PathInContainer: "/dev/xvdb",
+					PathOnHost:      "/dev/disk/by-id/diskUuid",
+					Permissions:     "mrw",
+				},
+				{
+					PathInContainer: "/mnt/rawdisk",
+					PathOnHost:      "/var/lib/rawdisk",
+					Permissions:     "r",
+				},
+			},
+			expectErr: false,
+		},
+		"invalid absolute Path": {
+			podVolumes: kubecontainer.VolumeMap{
+				"disk": kubecontainer.VolumeInfo{BlockVolumeMapper: &stubBlockVolume{dirPath: "/dev/", volName: "sda"}},
+			},
+			container: v1.Container{
+				VolumeDevices: []v1.VolumeDevice{
+					{
+						DevicePath: "must/be/absolute",
+						Name:       "disk",
+					},
+				},
+			},
+			expectErr:      true,
+			expectedErrMsg: "error DevicePath `must/be/absolute` must be an absolute path",
+		},
+		"volume doesn't exist": {
+			podVolumes: kubecontainer.VolumeMap{},
+			container: v1.Container{
+				VolumeDevices: []v1.VolumeDevice{
+					{
+						DevicePath: "/dev/sdaa",
+						Name:       "disk",
+					},
+				},
+			},
+			expectErr:      true,
+			expectedErrMsg: "cannot find volume \"disk\" to pass into container \"\"",
+		},
+		"volume BlockVolumeMapper is nil": {
+			podVolumes: kubecontainer.VolumeMap{
+				"disk": kubecontainer.VolumeInfo{},
+			},
+			container: v1.Container{
+				VolumeDevices: []v1.VolumeDevice{
+					{
+						DevicePath: "/dev/sdzz",
+						Name:       "disk",
+					},
+				},
+			},
+			expectErr:      true,
+			expectedErrMsg: "cannot find volume \"disk\" to pass into container \"\"",
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			pod := v1.Pod{
+				Spec: v1.PodSpec{
+					HostNetwork: true,
+				},
+			}
+			blkutil := volumetest.NewBlockVolumePathHandler()
+			blkVolumes, err := kubelet.makeBlockVolumes(&pod, &tc.container, tc.podVolumes, blkutil)
+			// validate only the error if we expect an error
+			if tc.expectErr {
+				if err == nil || err.Error() != tc.expectedErrMsg {
+					t.Fatalf("expected error message `%s` but got `%v`", tc.expectedErrMsg, err)
+				}
+				return
+			}
+			// otherwise validate the devices
+			if err != nil {
+				t.Fatal(err)
+			}
+			assert.Equal(t, tc.expectedDevices, blkVolumes, "devices of container %+v", tc.container)
+		})
+	}
+}
+
 func TestNodeHostsFileContent(t *testing.T) {
 	testCases := []struct {
 		hostsFileName            string

pkg/kubelet/kubelet_volumes_test.go

@@ -448,3 +448,23 @@ func (f *stubVolume) SetUp(fsGroup *int64) error {
 func (f *stubVolume) SetUpAt(dir string, fsGroup *int64) error {
 	return nil
 }
+
+type stubBlockVolume struct {
+	dirPath string
+	volName string
+}
+
+func (f *stubBlockVolume) GetGlobalMapPath(spec *volume.Spec) (string, error) {
+	return "", nil
+}
+
+func (f *stubBlockVolume) GetPodDeviceMapPath() (string, string) {
+	return f.dirPath, f.volName
+}
+
+func (f *stubBlockVolume) SetUpDevice() (string, error) {
+	return "", nil
+}
+func (f *stubBlockVolume) TearDownDevice(mapPath string, devicePath string) error {
+	return nil
+}

pkg/kubelet/volume_host.go

@@ -86,10 +86,18 @@ type kubeletVolumeHost struct {
 	mountPodManager  mountpod.Manager
 }
 
+func (kvh *kubeletVolumeHost) GetVolumeDevicePluginDir(pluginName string) string {
+	return kvh.kubelet.getVolumeDevicePluginDir(pluginName)
+}
+
 func (kvh *kubeletVolumeHost) GetPodVolumeDir(podUID types.UID, pluginName string, volumeName string) string {
 	return kvh.kubelet.getPodVolumeDir(podUID, pluginName, volumeName)
 }
 
+func (kvh *kubeletVolumeHost) GetPodVolumeDeviceDir(podUID types.UID, pluginName string) string {
+	return kvh.kubelet.getPodVolumeDeviceDir(podUID, pluginName)
+}
+
 func (kvh *kubeletVolumeHost) GetPodPluginDir(podUID types.UID, pluginName string) string {
 	return kvh.kubelet.getPodPluginDir(podUID, pluginName)
 }

pkg/kubelet/volumemanager/cache/actual_state_of_world.go

@@ -58,7 +58,7 @@ type ActualStateOfWorld interface {
 	// volume, reset the pod's remountRequired value.
 	// If a volume with the name volumeName does not exist in the list of
 	// attached volumes, an error is returned.
-	AddPodToVolume(podName volumetypes.UniquePodName, podUID types.UID, volumeName v1.UniqueVolumeName, mounter volume.Mounter, outerVolumeSpecName string, volumeGidValue string) error
+	AddPodToVolume(podName volumetypes.UniquePodName, podUID types.UID, volumeName v1.UniqueVolumeName, mounter volume.Mounter, blockVolumeMapper volume.BlockVolumeMapper, outerVolumeSpecName string, volumeGidValue string) error
 
 	// MarkRemountRequired marks each volume that is successfully attached and
 	// mounted for the specified pod as requiring remount (if the plugin for the
@@ -254,6 +254,9 @@ type mountedPod struct {
 	// mounter used to mount
 	mounter volume.Mounter
 
+	// mapper used to block volumes support
+	blockVolumeMapper volume.BlockVolumeMapper
+
 	// outerVolumeSpecName is the volume.Spec.Name() of the volume as referenced
 	// directly in the pod. If the volume was referenced through a persistent
 	// volume claim, this contains the volume.Spec.Name() of the persistent
@@ -287,6 +290,7 @@ func (asw *actualStateOfWorld) MarkVolumeAsMounted(
 	podUID types.UID,
 	volumeName v1.UniqueVolumeName,
 	mounter volume.Mounter,
+	blockVolumeMapper volume.BlockVolumeMapper,
 	outerVolumeSpecName string,
 	volumeGidValue string) error {
 	return asw.AddPodToVolume(
@@ -294,6 +298,7 @@ func (asw *actualStateOfWorld) MarkVolumeAsMounted(
 		podUID,
 		volumeName,
 		mounter,
+		blockVolumeMapper,
 		outerVolumeSpecName,
 		volumeGidValue)
 }
@@ -385,6 +390,7 @@ func (asw *actualStateOfWorld) AddPodToVolume(
 	podUID types.UID,
 	volumeName v1.UniqueVolumeName,
 	mounter volume.Mounter,
+	blockVolumeMapper volume.BlockVolumeMapper,
 	outerVolumeSpecName string,
 	volumeGidValue string) error {
 	asw.Lock()
@@ -403,6 +409,7 @@ func (asw *actualStateOfWorld) AddPodToVolume(
 			podName:             podName,
 			podUID:              podUID,
 			mounter:             mounter,
+			blockVolumeMapper:   blockVolumeMapper,
 			outerVolumeSpecName: outerVolumeSpecName,
 			volumeGidValue:      volumeGidValue,
 		}
@@ -682,5 +689,7 @@ func getMountedVolume(
 			PluginName:          attachedVolume.pluginName,
 			PodUID:              mountedPod.podUID,
 			Mounter:             mountedPod.mounter,
-			VolumeGidValue:      mountedPod.volumeGidValue}}
+			BlockVolumeMapper:   mountedPod.blockVolumeMapper,
+			VolumeGidValue:      mountedPod.volumeGidValue,
+			VolumeSpec:          attachedVolume.spec}}
 }

pkg/kubelet/volumemanager/cache/actual_state_of_world_test.go

@@ -204,9 +204,14 @@ func Test_AddPodToVolume_Positive_ExistingVolumeNewNode(t *testing.T) {
 		t.Fatalf("NewMounter failed. Expected: <no error> Actual: <%v>", err)
 	}
 
+	mapper, err := plugin.NewBlockVolumeMapper(volumeSpec, pod, volume.VolumeOptions{})
+	if err != nil {
+		t.Fatalf("NewBlockVolumeMapper failed. Expected: <no error> Actual: <%v>", err)
+	}
+
 	// Act
 	err = asw.AddPodToVolume(
-		podName, pod.UID, generatedVolumeName, mounter, volumeSpec.Name(), "" /* volumeGidValue */)
+		podName, pod.UID, generatedVolumeName, mounter, mapper, volumeSpec.Name(), "" /* volumeGidValue */)
 
 	// Assert
 	if err != nil {
@@ -263,15 +268,20 @@ func Test_AddPodToVolume_Positive_ExistingVolumeExistingNode(t *testing.T) {
 		t.Fatalf("NewMounter failed. Expected: <no error> Actual: <%v>", err)
 	}
 
+	mapper, err := plugin.NewBlockVolumeMapper(volumeSpec, pod, volume.VolumeOptions{})
+	if err != nil {
+		t.Fatalf("NewBlockVolumeMapper failed. Expected: <no error> Actual: <%v>", err)
+	}
+
 	err = asw.AddPodToVolume(
-		podName, pod.UID, generatedVolumeName, mounter, volumeSpec.Name(), "" /* volumeGidValue */)
+		podName, pod.UID, generatedVolumeName, mounter, mapper, volumeSpec.Name(), "" /* volumeGidValue */)
 	if err != nil {
 		t.Fatalf("AddPodToVolume failed. Expected: <no error> Actual: <%v>", err)
 	}
 
 	// Act
 	err = asw.AddPodToVolume(
-		podName, pod.UID, generatedVolumeName, mounter, volumeSpec.Name(), "" /* volumeGidValue */)
+		podName, pod.UID, generatedVolumeName, mounter, mapper, volumeSpec.Name(), "" /* volumeGidValue */)
 
 	// Assert
 	if err != nil {
@@ -318,6 +328,15 @@ func Test_AddPodToVolume_Negative_VolumeDoesntExist(t *testing.T) {
 			volumeSpec,
 			err)
 	}
+
+	blockplugin, err := volumePluginMgr.FindMapperPluginBySpec(volumeSpec)
+	if err != nil {
+		t.Fatalf(
+			"volumePluginMgr.FindMapperPluginBySpec failed to find volume plugin for %#v with: %v",
+			volumeSpec,
+			err)
+	}
+
 	volumeName, err := volumehelper.GetUniqueVolumeNameFromSpec(
 		plugin, volumeSpec)
 
@@ -328,9 +347,14 @@ func Test_AddPodToVolume_Negative_VolumeDoesntExist(t *testing.T) {
 		t.Fatalf("NewMounter failed. Expected: <no error> Actual: <%v>", err)
 	}
 
+	mapper, err := blockplugin.NewBlockVolumeMapper(volumeSpec, pod, volume.VolumeOptions{})
+	if err != nil {
+		t.Fatalf("NewBlockVolumeMapper failed. Expected: <no error> Actual: <%v>", err)
+	}
+
 	// Act
 	err = asw.AddPodToVolume(
-		podName, pod.UID, volumeName, mounter, volumeSpec.Name(), "" /* volumeGidValue */)
+		podName, pod.UID, volumeName, mounter, mapper, volumeSpec.Name(), "" /* volumeGidValue */)
 
 	// Assert
 	if err == nil {

pkg/kubelet/volumemanager/populator/desired_state_of_world_populator.go

@@ -31,7 +31,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/kubelet/config"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/pod"
@@ -260,11 +262,12 @@ func (dswp *desiredStateOfWorldPopulator) processPodVolumes(pod *v1.Pod) {
 	}
 
 	allVolumesAdded := true
+	mountsMap, devicesMap := dswp.makeVolumeMap(pod.Spec.Containers)
 
 	// Process volume spec for each volume defined in pod
 	for _, podVolume := range pod.Spec.Volumes {
 		volumeSpec, volumeGidValue, err :=
-			dswp.createVolumeSpec(podVolume, pod.Namespace)
+			dswp.createVolumeSpec(podVolume, pod.Name, pod.Namespace, mountsMap, devicesMap)
 		if err != nil {
 			glog.Errorf(
 				"Error processing volume %q for pod %q: %v",
@@ -336,7 +339,7 @@ func (dswp *desiredStateOfWorldPopulator) deleteProcessedPod(
 // specified volume. It dereference any PVC to get PV objects, if needed.
 // Returns an error if unable to obtain the volume at this time.
 func (dswp *desiredStateOfWorldPopulator) createVolumeSpec(
-	podVolume v1.Volume, podNamespace string) (*volume.Spec, string, error) {
+	podVolume v1.Volume, podName string, podNamespace string, mountsMap map[string]bool, devicesMap map[string]bool) (*volume.Spec, string, error) {
 	if pvcSource :=
 		podVolume.VolumeSource.PersistentVolumeClaim; pvcSource != nil {
 		glog.V(10).Infof(
@@ -381,6 +384,31 @@ func (dswp *desiredStateOfWorldPopulator) createVolumeSpec(
 			pvcSource.ClaimName,
 			pvcUID)
 
+		// TODO: remove feature gate check after no longer needed
+		if utilfeature.DefaultFeatureGate.Enabled(features.BlockVolume) {
+			volumeMode, err := volumehelper.GetVolumeMode(volumeSpec)
+			if err != nil {
+				return nil, "", err
+			}
+			// Error if a container has volumeMounts but the volumeMode of PVC isn't Filesystem
+			if mountsMap[podVolume.Name] && volumeMode != v1.PersistentVolumeFilesystem {
+				return nil, "", fmt.Errorf(
+					"Volume %q has volumeMode %q, but is specified in volumeMounts for pod %q/%q",
+					podVolume.Name,
+					volumeMode,
+					podNamespace,
+					podName)
+			}
+			// Error if a container has volumeDevices but the volumeMode of PVC isn't Block
+			if devicesMap[podVolume.Name] && volumeMode != v1.PersistentVolumeBlock {
+				return nil, "", fmt.Errorf(
+					"Volume %q has volumeMode %q, but is specified in volumeDevices for pod %q/%q",
+					podVolume.Name,
+					volumeMode,
+					podNamespace,
+					podName)
+			}
+		}
 		return volumeSpec, volumeGidValue, nil
 	}
 
@@ -449,6 +477,28 @@ func (dswp *desiredStateOfWorldPopulator) getPVSpec(
 	return volume.NewSpecFromPersistentVolume(pv, pvcReadOnly), volumeGidValue, nil
 }
 
+func (dswp *desiredStateOfWorldPopulator) makeVolumeMap(containers []v1.Container) (map[string]bool, map[string]bool) {
+	volumeDevicesMap := make(map[string]bool)
+	volumeMountsMap := make(map[string]bool)
+
+	for _, container := range containers {
+		if container.VolumeMounts != nil {
+			for _, mount := range container.VolumeMounts {
+				volumeMountsMap[mount.Name] = true
+			}
+		}
+		// TODO: remove feature gate check after no longer needed
+		if utilfeature.DefaultFeatureGate.Enabled(features.BlockVolume) &&
+			container.VolumeDevices != nil {
+			for _, device := range container.VolumeDevices {
+				volumeDevicesMap[device.Name] = true
+			}
+		}
+	}
+
+	return volumeMountsMap, volumeDevicesMap
+}
+
 func getPVVolumeGidAnnotationValue(pv *v1.PersistentVolume) string {
 	if volumeGid, ok := pv.Annotations[volumehelper.VolumeGidAnnotationKey]; ok {
 		return volumeGid

pkg/kubelet/volumemanager/volume_manager.go

@@ -41,6 +41,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/volumemanager/reconciler"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
 	"k8s.io/kubernetes/pkg/volume/util/types"
 	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
@@ -168,8 +169,8 @@ func NewVolumeManager(
 			kubeClient,
 			volumePluginMgr,
 			recorder,
-			checkNodeCapabilitiesBeforeMount),
-		),
+			checkNodeCapabilitiesBeforeMount,
+			util.NewBlockVolumePathHandler())),
 	}
 
 	vm.desiredStateOfWorldPopulator = populator.NewDesiredStateOfWorldPopulator(
@@ -253,7 +254,11 @@ func (vm *volumeManager) Run(sourcesReady config.SourcesReady, stopCh <-chan str
 func (vm *volumeManager) GetMountedVolumesForPod(podName types.UniquePodName) container.VolumeMap {
 	podVolumes := make(container.VolumeMap)
 	for _, mountedVolume := range vm.actualStateOfWorld.GetMountedVolumesForPod(podName) {
-		podVolumes[mountedVolume.OuterVolumeSpecName] = container.VolumeInfo{Mounter: mountedVolume.Mounter}
+		podVolumes[mountedVolume.OuterVolumeSpecName] = container.VolumeInfo{
+			Mounter:           mountedVolume.Mounter,
+			BlockVolumeMapper: mountedVolume.BlockVolumeMapper,
+			ReadOnly:          mountedVolume.VolumeSpec.ReadOnly,
+		}
 	}
 	return podVolumes
 }

pkg/volume/plugins.go

@@ -208,6 +208,26 @@ type ExpandableVolumePlugin interface {
 	RequiresFSResize() bool
 }
 
+// BlockVolumePlugin is an extend interface of VolumePlugin and is used for block volumes support.
+type BlockVolumePlugin interface {
+	VolumePlugin
+	// NewBlockVolumeMapper creates a new volume.BlockVolumeMapper from an API specification.
+	// Ownership of the spec pointer in *not* transferred.
+	// - spec: The v1.Volume spec
+	// - pod: The enclosing pod
+	NewBlockVolumeMapper(spec *Spec, podRef *v1.Pod, opts VolumeOptions) (BlockVolumeMapper, error)
+	// NewBlockVolumeUnmapper creates a new volume.BlockVolumeUnmapper from recoverable state.
+	// - name: The volume name, as per the v1.Volume spec.
+	// - podUID: The UID of the enclosing pod
+	NewBlockVolumeUnmapper(name string, podUID types.UID) (BlockVolumeUnmapper, error)
+	// ConstructBlockVolumeSpec constructs a volume spec based on the given
+	// podUID, volume name and a pod device map path.
+	// The spec may have incomplete information due to limited information
+	// from input. This function is used by volume manager to reconstruct
+	// volume spec by reading the volume directories from disk.
+	ConstructBlockVolumeSpec(podUID types.UID, volumeName, mountPath string) (*Spec, error)
+}
+
 // VolumeHost is an interface that plugins can use to access the kubelet.
 type VolumeHost interface {
 	// GetPluginDir returns the absolute path to a directory under which
@@ -216,6 +236,11 @@ type VolumeHost interface {
 	// GetPodPluginDir().
 	GetPluginDir(pluginName string) string
 
+	// GetVolumeDevicePluginDir returns the absolute path to a directory
+	// under which a given plugin may store data.
+	// ex. plugins/kubernetes.io/{PluginName}/{DefaultKubeletVolumeDevicesDirName}/{volumePluginDependentPath}/
+	GetVolumeDevicePluginDir(pluginName string) string
+
 	// GetPodVolumeDir returns the absolute path a directory which
 	// represents the named volume under the named plugin for the given
 	// pod.  If the specified pod does not exist, the result of this call
@@ -228,6 +253,13 @@ type VolumeHost interface {
 	// directory might not actually exist on disk yet.
 	GetPodPluginDir(podUID types.UID, pluginName string) string
 
+	// GetPodVolumeDeviceDir returns the absolute path a directory which
+	// represents the named plugin for the given pod.
+	// If the specified pod does not exist, the result of this call
+	// might not exist.
+	// ex. pods/{podUid}/{DefaultKubeletVolumeDevicesDirName}/{escapeQualifiedPluginName}/
+	GetPodVolumeDeviceDir(podUID types.UID, pluginName string) string
+
 	// GetKubeClient returns a client interface
 	GetKubeClient() clientset.Interface
 
@@ -675,6 +707,32 @@ func (pm *VolumePluginMgr) FindExpandablePluginByName(name string) (ExpandableVo
 	return nil, nil
 }
 
+// FindMapperPluginBySpec fetches a block volume plugin by spec.
+func (pm *VolumePluginMgr) FindMapperPluginBySpec(spec *Spec) (BlockVolumePlugin, error) {
+	volumePlugin, err := pm.FindPluginBySpec(spec)
+	if err != nil {
+		return nil, err
+	}
+
+	if blockVolumePlugin, ok := volumePlugin.(BlockVolumePlugin); ok {
+		return blockVolumePlugin, nil
+	}
+	return nil, nil
+}
+
+// FindMapperPluginByName fetches a block volume plugin by name.
+func (pm *VolumePluginMgr) FindMapperPluginByName(name string) (BlockVolumePlugin, error) {
+	volumePlugin, err := pm.FindPluginByName(name)
+	if err != nil {
+		return nil, err
+	}
+
+	if blockVolumePlugin, ok := volumePlugin.(BlockVolumePlugin); ok {
+		return blockVolumePlugin, nil
+	}
+	return nil, nil
+}
+
 // NewPersistentVolumeRecyclerPodTemplate creates a template for a recycler
 // pod.  By default, a recycler pod simply runs "rm -rf" on a volume and tests
 // for emptiness.  Most attributes of the template will be correct for most

pkg/volume/util/operationexecutor/operation_executor_test.go

@@ -40,6 +40,10 @@ const (
 	numVolumesToDetach                   = 2
 	numVolumesToVerifyAttached           = 2
 	numVolumesToVerifyControllerAttached = 2
+	numVolumesToMap                      = 2
+	numAttachableVolumesToUnmap          = 2
+	numNonAttachableVolumesToUnmap       = 2
+	numDevicesToUnmap                    = 2
 )
 
 var _ OperationGenerator = &fakeOperationGenerator{}
@@ -228,6 +232,113 @@ func TestOperationExecutor_VerifyControllerAttachedVolumeConcurrently(t *testing
 	}
 }
 
+func TestOperationExecutor_MapVolume_ConcurrentMapForNonAttachablePlugins(t *testing.T) {
+	// Arrange
+	ch, quit, oe := setup()
+	volumesToMount := make([]VolumeToMount, numVolumesToMap)
+	secretName := "secret-volume"
+	volumeName := v1.UniqueVolumeName(secretName)
+
+	// Act
+	for i := range volumesToMount {
+		podName := "pod-" + strconv.Itoa((i + 1))
+		pod := getTestPodWithSecret(podName, secretName)
+		volumesToMount[i] = VolumeToMount{
+			Pod:                pod,
+			VolumeName:         volumeName,
+			PluginIsAttachable: false, // this field determines whether the plugin is attachable
+			ReportedInUse:      true,
+		}
+		oe.MapVolume(0 /* waitForAttachTimeOut */, volumesToMount[i], nil /* actualStateOfWorldMounterUpdater */)
+	}
+
+	// Assert
+	if !isOperationRunConcurrently(ch, quit, numVolumesToMap) {
+		t.Fatalf("Unable to start map operations in Concurrent for non-attachable volumes")
+	}
+}
+
+func TestOperationExecutor_MapVolume_ConcurrentMapForAttachablePlugins(t *testing.T) {
+	// Arrange
+	ch, quit, oe := setup()
+	volumesToMount := make([]VolumeToMount, numVolumesToAttach)
+	pdName := "pd-volume"
+	volumeName := v1.UniqueVolumeName(pdName)
+
+	// Act
+	for i := range volumesToMount {
+		podName := "pod-" + strconv.Itoa((i + 1))
+		pod := getTestPodWithGCEPD(podName, pdName)
+		volumesToMount[i] = VolumeToMount{
+			Pod:                pod,
+			VolumeName:         volumeName,
+			PluginIsAttachable: true, // this field determines whether the plugin is attachable
+			ReportedInUse:      true,
+		}
+		oe.MapVolume(0 /* waitForAttachTimeout */, volumesToMount[i], nil /* actualStateOfWorldMounterUpdater */)
+	}
+
+	// Assert
+	if !isOperationRunSerially(ch, quit) {
+		t.Fatalf("Map operations should not start concurrently for attachable volumes")
+	}
+}
+
+func TestOperationExecutor_UnmapVolume_ConcurrentUnmapForAllPlugins(t *testing.T) {
+	// Arrange
+	ch, quit, oe := setup()
+	volumesToUnmount := make([]MountedVolume, numAttachableVolumesToUnmap+numNonAttachableVolumesToUnmap)
+	pdName := "pd-volume"
+	secretName := "secret-volume"
+
+	// Act
+	for i := 0; i < numNonAttachableVolumesToUnmap+numAttachableVolumesToUnmap; i++ {
+		podName := "pod-" + strconv.Itoa(i+1)
+		if i < numNonAttachableVolumesToUnmap {
+			pod := getTestPodWithSecret(podName, secretName)
+			volumesToUnmount[i] = MountedVolume{
+				PodName:    volumetypes.UniquePodName(podName),
+				VolumeName: v1.UniqueVolumeName(secretName),
+				PodUID:     pod.UID,
+			}
+		} else {
+			pod := getTestPodWithGCEPD(podName, pdName)
+			volumesToUnmount[i] = MountedVolume{
+				PodName:    volumetypes.UniquePodName(podName),
+				VolumeName: v1.UniqueVolumeName(pdName),
+				PodUID:     pod.UID,
+			}
+		}
+		oe.UnmapVolume(volumesToUnmount[i], nil /* actualStateOfWorldMounterUpdater */)
+	}
+
+	// Assert
+	if !isOperationRunConcurrently(ch, quit, numNonAttachableVolumesToUnmap+numAttachableVolumesToUnmap) {
+		t.Fatalf("Unable to start unmap operations concurrently for volume plugins")
+	}
+}
+
+func TestOperationExecutor_UnmapDeviceConcurrently(t *testing.T) {
+	// Arrange
+	ch, quit, oe := setup()
+	attachedVolumes := make([]AttachedVolume, numDevicesToUnmap)
+	pdName := "pd-volume"
+
+	// Act
+	for i := range attachedVolumes {
+		attachedVolumes[i] = AttachedVolume{
+			VolumeName: v1.UniqueVolumeName(pdName),
+			NodeName:   "node-name",
+		}
+		oe.UnmapDevice(attachedVolumes[i], nil /* actualStateOfWorldMounterUpdater */, nil /* mount.Interface */)
+	}
+
+	// Assert
+	if !isOperationRunSerially(ch, quit) {
+		t.Fatalf("Unmap device operations should not start concurrently")
+	}
+}
+
 type fakeOperationGenerator struct {
 	ch   chan interface{}
 	quit chan interface{}
@@ -302,6 +413,27 @@ func (fopg *fakeOperationGenerator) GenerateBulkVolumeVerifyFunc(
 	}, nil
 }
 
+func (fopg *fakeOperationGenerator) GenerateMapVolumeFunc(waitForAttachTimeout time.Duration, volumeToMount VolumeToMount, actualStateOfWorldMounterUpdater ActualStateOfWorldMounterUpdater) (func() error, string, error) {
+	return func() error {
+		startOperationAndBlock(fopg.ch, fopg.quit)
+		return nil
+	}, "", nil
+}
+
+func (fopg *fakeOperationGenerator) GenerateUnmapVolumeFunc(volumeToUnmount MountedVolume, actualStateOfWorld ActualStateOfWorldMounterUpdater) (func() error, string, error) {
+	return func() error {
+		startOperationAndBlock(fopg.ch, fopg.quit)
+		return nil
+	}, "", nil
+}
+
+func (fopg *fakeOperationGenerator) GenerateUnmapDeviceFunc(deviceToDetach AttachedVolume, actualStateOfWorld ActualStateOfWorldMounterUpdater, mounter mount.Interface) (func() error, string, error) {
+	return func() error {
+		startOperationAndBlock(fopg.ch, fopg.quit)
+		return nil
+	}, "", nil
+}
+
 func (fopg *fakeOperationGenerator) GetVolumePluginMgr() *volume.VolumePluginMgr {
 	return nil
 }

pkg/volume/util/util.go

@@ -21,7 +21,7 @@ import (
 	"io/ioutil"
 	"os"
 	"path"
-
+	"path/filepath"
 	"strings"
 
 	"github.com/golang/glog"
@@ -30,6 +30,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
@@ -38,7 +39,14 @@ import (
 	"k8s.io/kubernetes/pkg/util/mount"
 )
 
-const readyFileName = "ready"
+const (
+	readyFileName = "ready"
+	losetupPath   = "losetup"
+
+	ErrDeviceNotFound     = "device not found"
+	ErrDeviceNotSupported = "device not supported"
+	ErrNotAvailable       = "not available"
+)
 
 // IsReady checks for the existence of a regular file
 // called 'ready' in the given directory and returns
@@ -270,3 +278,201 @@ func stringToSet(str, delimiter string) (sets.String, error) {
 	}
 	return zonesSet, nil
 }
+
+// BlockVolumePathHandler defines a set of operations for handling block volume-related operations
+type BlockVolumePathHandler interface {
+	// MapDevice creates a symbolic link to block device under specified map path
+	MapDevice(devicePath string, mapPath string, linkName string) error
+	// UnmapDevice removes a symbolic link to block device under specified map path
+	UnmapDevice(mapPath string, linkName string) error
+	// RemovePath removes a file or directory on specified map path
+	RemoveMapPath(mapPath string) error
+	// IsSymlinkExist retruns true if specified symbolic link exists
+	IsSymlinkExist(mapPath string) (bool, error)
+	// GetDeviceSymlinkRefs searches symbolic links under global map path
+	GetDeviceSymlinkRefs(devPath string, mapPath string) ([]string, error)
+	// FindGlobalMapPathUUIDFromPod finds {pod uuid} symbolic link under globalMapPath
+	// corresponding to map path symlink, and then return global map path with pod uuid.
+	FindGlobalMapPathUUIDFromPod(pluginDir, mapPath string, podUID types.UID) (string, error)
+	// AttachFileDevice takes a path to a regular file and makes it available as an
+	// attached block device.
+	AttachFileDevice(path string) (string, error)
+	// GetLoopDevice returns the full path to the loop device associated with the given path.
+	GetLoopDevice(path string) (string, error)
+	// RemoveLoopDevice removes specified loopback device
+	RemoveLoopDevice(device string) error
+}
+
+// NewBlockVolumePathHandler returns a new instance of BlockVolumeHandler.
+func NewBlockVolumePathHandler() BlockVolumePathHandler {
+	var volumePathHandler VolumePathHandler
+	return volumePathHandler
+}
+
+// VolumePathHandler is path related operation handlers for block volume
+type VolumePathHandler struct {
+}
+
+// MapDevice creates a symbolic link to block device under specified map path
+func (v VolumePathHandler) MapDevice(devicePath string, mapPath string, linkName string) error {
+	// Example of global map path:
+	//   globalMapPath/linkName: plugins/kubernetes.io/{PluginName}/{DefaultKubeletVolumeDevicesDirName}/{volumePluginDependentPath}/{podUid}
+	//   linkName: {podUid}
+	//
+	// Example of pod device map path:
+	//   podDeviceMapPath/linkName: pods/{podUid}/{DefaultKubeletVolumeDevicesDirName}/{escapeQualifiedPluginName}/{volumeName}
+	//   linkName: {volumeName}
+	if len(devicePath) == 0 {
+		return fmt.Errorf("Failed to map device to map path. devicePath is empty")
+	}
+	if len(mapPath) == 0 {
+		return fmt.Errorf("Failed to map device to map path. mapPath is empty")
+	}
+	if !filepath.IsAbs(mapPath) {
+		return fmt.Errorf("The map path should be absolute: map path: %s", mapPath)
+	}
+	glog.V(5).Infof("MapDevice: devicePath %s", devicePath)
+	glog.V(5).Infof("MapDevice: mapPath %s", mapPath)
+	glog.V(5).Infof("MapDevice: linkName %s", linkName)
+
+	// Check and create mapPath
+	_, err := os.Stat(mapPath)
+	if err != nil && !os.IsNotExist(err) {
+		glog.Errorf("cannot validate map path: %s", mapPath)
+		return err
+	}
+	if err = os.MkdirAll(mapPath, 0750); err != nil {
+		return fmt.Errorf("Failed to mkdir %s, error %v", mapPath, err)
+	}
+	// Remove old symbolic link(or file) then create new one.
+	// This should be done because current symbolic link is
+	// stale accross node reboot.
+	linkPath := path.Join(mapPath, string(linkName))
+	if err = os.Remove(linkPath); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	err = os.Symlink(devicePath, linkPath)
+	return err
+}
+
+// UnmapDevice removes a symbolic link associated to block device under specified map path
+func (v VolumePathHandler) UnmapDevice(mapPath string, linkName string) error {
+	if len(mapPath) == 0 {
+		return fmt.Errorf("Failed to unmap device from map path. mapPath is empty")
+	}
+	glog.V(5).Infof("UnmapDevice: mapPath %s", mapPath)
+	glog.V(5).Infof("UnmapDevice: linkName %s", linkName)
+
+	// Check symbolic link exists
+	linkPath := path.Join(mapPath, string(linkName))
+	if islinkExist, checkErr := v.IsSymlinkExist(linkPath); checkErr != nil {
+		return checkErr
+	} else if !islinkExist {
+		glog.Warningf("Warning: Unmap skipped because symlink does not exist on the path: %v", linkPath)
+		return nil
+	}
+	err := os.Remove(linkPath)
+	return err
+}
+
+// RemoveMapPath removes a file or directory on specified map path
+func (v VolumePathHandler) RemoveMapPath(mapPath string) error {
+	if len(mapPath) == 0 {
+		return fmt.Errorf("Failed to remove map path. mapPath is empty")
+	}
+	glog.V(5).Infof("RemoveMapPath: mapPath %s", mapPath)
+	err := os.RemoveAll(mapPath)
+	if err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	return nil
+}
+
+// IsSymlinkExist returns true if specified file exists and the type is symbolik link.
+// If file doesn't exist, or file exists but not symbolick link, return false with no error.
+// On other cases, return false with error from Lstat().
+func (v VolumePathHandler) IsSymlinkExist(mapPath string) (bool, error) {
+	fi, err := os.Lstat(mapPath)
+	if err == nil {
+		// If file exits and it's symbolick link, return true and no error
+		if fi.Mode()&os.ModeSymlink == os.ModeSymlink {
+			return true, nil
+		}
+		// If file exits but it's not symbolick link, return fale and no error
+		return false, nil
+	}
+	// If file doesn't exist, return false and no error
+	if os.IsNotExist(err) {
+		return false, nil
+	}
+	// Return error from Lstat()
+	return false, err
+}
+
+// GetDeviceSymlinkRefs searches symbolic links under global map path
+func (v VolumePathHandler) GetDeviceSymlinkRefs(devPath string, mapPath string) ([]string, error) {
+	var refs []string
+	files, err := ioutil.ReadDir(mapPath)
+	if err != nil {
+		return nil, fmt.Errorf("Directory cannot read %v", err)
+	}
+	for _, file := range files {
+		if file.Mode()&os.ModeSymlink != os.ModeSymlink {
+			continue
+		}
+		filename := file.Name()
+		filepath, err := os.Readlink(path.Join(mapPath, filename))
+		if err != nil {
+			return nil, fmt.Errorf("Symbolic link cannot be retrieved %v", err)
+		}
+		glog.V(5).Infof("GetDeviceSymlinkRefs: filepath: %v, devPath: %v", filepath, devPath)
+		if filepath == devPath {
+			refs = append(refs, path.Join(mapPath, filename))
+		}
+	}
+	glog.V(5).Infof("GetDeviceSymlinkRefs: refs %v", refs)
+	return refs, nil
+}
+
+// FindGlobalMapPathUUIDFromPod finds {pod uuid} symbolic link under globalMapPath
+// corresponding to map path symlink, and then return global map path with pod uuid.
+// ex. mapPath symlink: pods/{podUid}}/{DefaultKubeletVolumeDevicesDirName}/{escapeQualifiedPluginName}/{volumeName} -> /dev/sdX
+//     globalMapPath/{pod uuid}: plugins/kubernetes.io/{PluginName}/{DefaultKubeletVolumeDevicesDirName}/{volumePluginDependentPath}/{pod uuid} -> /dev/sdX
+func (v VolumePathHandler) FindGlobalMapPathUUIDFromPod(pluginDir, mapPath string, podUID types.UID) (string, error) {
+	var globalMapPathUUID string
+	// Find symbolic link named pod uuid under plugin dir
+	err := filepath.Walk(pluginDir, func(path string, fi os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if (fi.Mode()&os.ModeSymlink == os.ModeSymlink) && (fi.Name() == string(podUID)) {
+			glog.V(5).Infof("FindGlobalMapPathFromPod: path %s, mapPath %s", path, mapPath)
+			if res, err := compareSymlinks(path, mapPath); err == nil && res {
+				globalMapPathUUID = path
+			}
+		}
+		return nil
+	})
+	if err != nil {
+		return "", err
+	}
+	glog.V(5).Infof("FindGlobalMapPathFromPod: globalMapPathUUID %s", globalMapPathUUID)
+	// Return path contains global map path + {pod uuid}
+	return globalMapPathUUID, nil
+}
+
+func compareSymlinks(global, pod string) (bool, error) {
+	devGlobal, err := os.Readlink(global)
+	if err != nil {
+		return false, err
+	}
+	devPod, err := os.Readlink(pod)
+	if err != nil {
+		return false, err
+	}
+	glog.V(5).Infof("CompareSymlinks: devGloBal %s, devPod %s", devGlobal, devPod)
+	if devGlobal == devPod {
+		return true, nil
+	}
+	return false, nil
+}

pkg/volume/util/util_linux.go

+// +build linux
+
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
+
+	"github.com/golang/glog"
+)
+
+// AttachFileDevice takes a path to a regular file and makes it available as an
+// attached block device.
+func (v VolumePathHandler) AttachFileDevice(path string) (string, error) {
+	blockDevicePath, err := v.GetLoopDevice(path)
+	if err != nil && err.Error() != ErrDeviceNotFound {
+		return "", err
+	}
+
+	// If no existing loop device for the path, create one
+	if blockDevicePath == "" {
+		glog.V(4).Infof("Creating device for path: %s", path)
+		blockDevicePath, err = makeLoopDevice(path)
+		if err != nil {
+			return "", err
+		}
+	}
+	return blockDevicePath, nil
+}
+
+// GetLoopDevice returns the full path to the loop device associated with the given path.
+func (v VolumePathHandler) GetLoopDevice(path string) (string, error) {
+	_, err := os.Stat(path)
+	if os.IsNotExist(err) {
+		return "", errors.New(ErrNotAvailable)
+	}
+	if err != nil {
+		return "", fmt.Errorf("not attachable: %v", err)
+	}
+
+	args := []string{"-j", path}
+	cmd := exec.Command(losetupPath, args...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		glog.V(2).Infof("Failed device discover command for path %s: %v", path, err)
+		return "", err
+	}
+	return parseLosetupOutputForDevice(out)
+}
+
+func makeLoopDevice(path string) (string, error) {
+	args := []string{"-f", "--show", path}
+	cmd := exec.Command(losetupPath, args...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		glog.V(2).Infof("Failed device create command for path %s: %v", path, err)
+		return "", err
+	}
+	return parseLosetupOutputForDevice(out)
+}
+
+// RemoveLoopDevice removes specified loopback device
+func (v VolumePathHandler) RemoveLoopDevice(device string) error {
+	args := []string{"-d", device}
+	cmd := exec.Command(losetupPath, args...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		if !strings.Contains(string(out), "No such device or address") {
+			return err
+		}
+	}
+	return nil
+}
+
+func parseLosetupOutputForDevice(output []byte) (string, error) {
+	if len(output) == 0 {
+		return "", errors.New(ErrDeviceNotFound)
+	}
+
+	// losetup returns device in the format:
+	// /dev/loop1: [0073]:148662 (/dev/sda)
+	device := strings.TrimSpace(strings.SplitN(string(output), ":", 2)[0])
+	if len(device) == 0 {
+		return "", errors.New(ErrDeviceNotFound)
+	}
+	return device, nil
+}

pkg/volume/util/util_unsupported.go

+// +build !linux
+
+/*
+Copyright 2017 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package util
+
+import (
+	"fmt"
+)
+
+// AttachFileDevice takes a path to a regular file and makes it available as an
+// attached block device.
+func (v VolumePathHandler) AttachFileDevice(path string) (string, error) {
+	return "", fmt.Errorf("AttachFileDevice not supported for this build.")
+}
+
+// GetLoopDevice returns the full path to the loop device associated with the given path.
+func (v VolumePathHandler) GetLoopDevice(path string) (string, error) {
+	return "", fmt.Errorf("GetLoopDevice not supported for this build.")
+}
+
+// RemoveLoopDevice removes specified loopback device
+func (v VolumePathHandler) RemoveLoopDevice(device string) error {
+	return fmt.Errorf("RemoveLoopDevice not supported for this build.")
+}

pkg/volume/util/volumehelper/volumehelper.go

@@ -136,3 +136,24 @@ func NewSafeFormatAndMountFromHost(pluginName string, host volume.VolumeHost) *m
 	exec := host.GetExec(pluginName)
 	return &mount.SafeFormatAndMount{Interface: mounter, Exec: exec}
 }
+
+// GetVolumeMode retrieves VolumeMode from pv.
+// If the volume doesn't have PersistentVolume, it's an inline volume,
+// should return volumeMode as filesystem to keep existing behavior.
+func GetVolumeMode(volumeSpec *volume.Spec) (v1.PersistentVolumeMode, error) {
+	if volumeSpec == nil || volumeSpec.PersistentVolume == nil {
+		return v1.PersistentVolumeFilesystem, nil
+	}
+	if volumeSpec.PersistentVolume.Spec.VolumeMode != nil {
+		return *volumeSpec.PersistentVolume.Spec.VolumeMode, nil
+	}
+	return "", fmt.Errorf("cannot get volumeMode for volume: %v", volumeSpec.Name())
+}
+
+// GetPersistentVolumeClaimVolumeMode retrieves VolumeMode from pvc.
+func GetPersistentVolumeClaimVolumeMode(claim *v1.PersistentVolumeClaim) (v1.PersistentVolumeMode, error) {
+	if claim.Spec.VolumeMode != nil {
+		return *claim.Spec.VolumeMode, nil
+	}
+	return "", fmt.Errorf("cannot get volumeMode from pvc: %v", claim.Name)
+}

pkg/volume/volume.go

@@ -37,6 +37,19 @@ type Volume interface {
 	MetricsProvider
 }
 
+// BlockVolume interface provides methods to generate global map path
+// and pod device map path.
+type BlockVolume interface {
+	// GetGlobalMapPath returns a global map path which contains
+	// symbolic links associated to a block device.
+	// ex. plugins/kubernetes.io/{PluginName}/{DefaultKubeletVolumeDevicesDirName}/{volumePluginDependentPath}/{pod uuid}
+	GetGlobalMapPath(spec *Spec) (string, error)
+	// GetPodDeviceMapPath returns a pod device map path
+	// and name of a symbolic link associated to a block device.
+	// ex. pods/{podUid}}/{DefaultKubeletVolumeDevicesDirName}/{escapeQualifiedPluginName}/{volumeName}
+	GetPodDeviceMapPath() (string, string)
+}
+
 // MetricsProvider exposes metrics (e.g. used,available space) related to a
 // Volume.
 type MetricsProvider interface {
@@ -132,6 +145,34 @@ type Unmounter interface {
 	TearDownAt(dir string) error
 }
 
+// BlockVolumeMapper interface provides methods to set up/map the volume.
+type BlockVolumeMapper interface {
+	BlockVolume
+	// SetUpDevice prepares the volume to a self-determined directory path,
+	// which may or may not exist yet and returns combination of physical
+	// device path of a block volume and error.
+	// If the plugin is non-attachable, it should prepare the device
+	// in /dev/ (or where appropriate) and return unique device path.
+	// Unique device path across kubelet node reboot is required to avoid
+	// unexpected block volume destruction.
+	// If the plugin is attachable, it should not do anything here,
+	// just return empty string for device path.
+	// Instead, attachable plugin have to return unique device path
+	// at attacher.Attach() and attacher.WaitForAttach().
+	// This may be called more than once, so implementations must be idempotent.
+	SetUpDevice() (string, error)
+}
+
+// BlockVolumeUnmapper interface provides methods to cleanup/unmap the volumes.
+type BlockVolumeUnmapper interface {
+	BlockVolume
+	// TearDownDevice removes traces of the SetUpDevice procedure under
+	// a self-determined directory.
+	// If the plugin is non-attachable, this method detaches the volume
+	// from a node.
+	TearDownDevice(mapPath string, devicePath string) error
+}
+
 // Provisioner is an interface that creates templates for PersistentVolumes
 // and can create the volume as a new resource in the infrastructure provider.
 type Provisioner interface {