Block volumes Support: FC plugin update

pkg/volume/fc/attacher.go

@@ -26,6 +26,8 @@ import (
 	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
 	volumeutil "k8s.io/kubernetes/pkg/volume/util"
@@ -176,9 +178,7 @@ func volumeSpecToMounter(spec *volume.Spec, host volume.VolumeHost) (*fcDiskMoun
 	} else {
 		return nil, fmt.Errorf("fc: no fc disk information found. failed to make a new mounter")
 	}
-
-	return &fcDiskMounter{
-		fcDisk: &fcDisk{
+	fcDisk := &fcDisk{
 		plugin: &fcPlugin{
 			host: host,
 		},
@@ -186,7 +186,24 @@ func volumeSpecToMounter(spec *volume.Spec, host volume.VolumeHost) (*fcDiskMoun
 		lun:   lun,
 		wwids: wwids,
 		io:    &osIOHandler{},
-		},
+	}
+	// TODO: remove feature gate check after no longer needed
+	if utilfeature.DefaultFeatureGate.Enabled(features.BlockVolume) {
+		volumeMode, err := volumehelper.GetVolumeMode(spec)
+		if err != nil {
+			return nil, err
+		}
+		glog.V(5).Infof("fc: volumeSpecToMounter volumeMode %s", volumeMode)
+		return &fcDiskMounter{
+			fcDisk:     fcDisk,
+			fsType:     fc.FSType,
+			volumeMode: volumeMode,
+			readOnly:   readOnly,
+			mounter:    volumehelper.NewSafeFormatAndMountFromHost(fcPluginName, host),
+		}, nil
+	}
+	return &fcDiskMounter{
+		fcDisk:   fcDisk,
 		fsType:   fc.FSType,
 		readOnly: readOnly,
 		mounter:  volumehelper.NewSafeFormatAndMountFromHost(fcPluginName, host),

pkg/volume/fc/disk_manager.go

@@ -27,6 +27,7 @@ import (
 // Abstract interface to disk operations.
 type diskManager interface {
 	MakeGlobalPDName(disk fcDisk) string
+	MakeGlobalVDPDName(disk fcDisk) string
 	// Attaches the disk to the kubelet's host machine.
 	AttachDisk(b fcDiskMounter) (string, error)
 	// Detaches the disk from the kubelet's host machine.

pkg/volume/fc/fc.go

@@ -23,11 +23,15 @@ import (
 
 	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/util/mount"
 	utilstrings "k8s.io/kubernetes/pkg/util/strings"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/util"
+	"k8s.io/kubernetes/pkg/volume/util/volumehelper"
 )
 
 // This is the primary entrypoint for volume plugins.
@@ -41,6 +45,7 @@ type fcPlugin struct {
 
 var _ volume.VolumePlugin = &fcPlugin{}
 var _ volume.PersistentVolumePlugin = &fcPlugin{}
+var _ volume.BlockVolumePlugin = &fcPlugin{}
 
 const (
 	fcPluginName = "kubernetes.io/fc"
@@ -115,29 +120,75 @@ func (plugin *fcPlugin) newMounterInternal(spec *volume.Spec, podUID types.UID,
 		return nil, err
 	}
 
-	var lun string
-	var wwids []string
-	if fc.Lun != nil && len(fc.TargetWWNs) != 0 {
-		lun = strconv.Itoa(int(*fc.Lun))
-	} else if len(fc.WWIDs) != 0 {
-		for _, wwid := range fc.WWIDs {
-			wwids = append(wwids, strings.Replace(wwid, " ", "_", -1))
-		}
-	} else {
+	wwns, lun, wwids, err := getWwnsLunWwids(fc)
+	if err != nil {
 		return nil, fmt.Errorf("fc: no fc disk information found. failed to make a new mounter")
 	}
-
+	fcDisk := &fcDisk{
+		podUID:  podUID,
+		volName: spec.Name(),
+		wwns:    wwns,
+		lun:     lun,
+		wwids:   wwids,
+		manager: manager,
+		io:      &osIOHandler{},
+		plugin:  plugin,
+	}
+	// TODO: remove feature gate check after no longer needed
+	if utilfeature.DefaultFeatureGate.Enabled(features.BlockVolume) {
+		volumeMode, err := volumehelper.GetVolumeMode(spec)
+		if err != nil {
+			return nil, err
+		}
+		glog.V(5).Infof("fc: newMounterInternal volumeMode %s", volumeMode)
 		return &fcDiskMounter{
+			fcDisk:     fcDisk,
+			fsType:     fc.FSType,
+			volumeMode: volumeMode,
+			readOnly:   readOnly,
+			mounter:    &mount.SafeFormatAndMount{Interface: mounter, Exec: exec},
+		}, nil
+	}
+	return &fcDiskMounter{
+		fcDisk:   fcDisk,
+		fsType:   fc.FSType,
+		readOnly: readOnly,
+		mounter:  &mount.SafeFormatAndMount{Interface: mounter, Exec: exec},
+	}, nil
+
+}
+
+func (plugin *fcPlugin) NewBlockVolumeMapper(spec *volume.Spec, pod *v1.Pod, _ volume.VolumeOptions) (volume.BlockVolumeMapper, error) {
+	// If this called via GenerateUnmapDeviceFunc(), pod is nil.
+	// Pass empty string as dummy uid since uid isn't used in the case.
+	var uid types.UID
+	if pod != nil {
+		uid = pod.UID
+	}
+	return plugin.newBlockVolumeMapperInternal(spec, uid, &FCUtil{}, plugin.host.GetMounter(plugin.GetPluginName()), plugin.host.GetExec(plugin.GetPluginName()))
+}
+
+func (plugin *fcPlugin) newBlockVolumeMapperInternal(spec *volume.Spec, podUID types.UID, manager diskManager, mounter mount.Interface, exec mount.Exec) (volume.BlockVolumeMapper, error) {
+	fc, readOnly, err := getVolumeSource(spec)
+	if err != nil {
+		return nil, err
+	}
+
+	wwns, lun, wwids, err := getWwnsLunWwids(fc)
+	if err != nil {
+		return nil, fmt.Errorf("fc: no fc disk information found. failed to make a new mapper")
+	}
+
+	return &fcDiskMapper{
 		fcDisk: &fcDisk{
 			podUID:  podUID,
 			volName: spec.Name(),
-			wwns:    fc.TargetWWNs,
+			wwns:    wwns,
 			lun:     lun,
 			wwids:   wwids,
 			manager: manager,
 			io:      &osIOHandler{},
 			plugin:  plugin},
-		fsType:   fc.FSType,
 		readOnly: readOnly,
 		mounter:  &mount.SafeFormatAndMount{Interface: mounter, Exec: exec},
 	}, nil
@@ -161,6 +212,22 @@ func (plugin *fcPlugin) newUnmounterInternal(volName string, podUID types.UID, m
 	}, nil
 }
 
+func (plugin *fcPlugin) NewBlockVolumeUnmapper(volName string, podUID types.UID) (volume.BlockVolumeUnmapper, error) {
+	return plugin.newUnmapperInternal(volName, podUID, &FCUtil{})
+}
+
+func (plugin *fcPlugin) newUnmapperInternal(volName string, podUID types.UID, manager diskManager) (volume.BlockVolumeUnmapper, error) {
+	return &fcDiskUnmapper{
+		fcDisk: &fcDisk{
+			podUID:  podUID,
+			volName: volName,
+			manager: manager,
+			plugin:  plugin,
+			io:      &osIOHandler{},
+		},
+	}, nil
+}
+
 func (plugin *fcPlugin) ConstructVolumeSpec(volumeName, mountPath string) (*volume.Spec, error) {
 	fcVolume := &v1.Volume{
 		Name: volumeName,
@@ -171,6 +238,55 @@ func (plugin *fcPlugin) ConstructVolumeSpec(volumeName, mountPath string) (*volu
 	return volume.NewSpecFromVolume(fcVolume), nil
 }
 
+// ConstructBlockVolumeSpec creates a new volume.Spec with following steps.
+//   - Searchs a file whose name is {pod uuid} under volume plugin directory.
+//   - If a file is found, then retreives volumePluginDependentPath from globalMapPathUUID.
+//   - Once volumePluginDependentPath is obtained, store volume information to VolumeSource
+// examples:
+//   mapPath: pods/{podUid}}/{DefaultKubeletVolumeDevicesDirName}/{escapeQualifiedPluginName}/{volumeName}
+//   globalMapPathUUID : plugins/kubernetes.io/{PluginName}/{DefaultKubeletVolumeDevicesDirName}/{volumePluginDependentPath}/{pod uuid}
+func (plugin *fcPlugin) ConstructBlockVolumeSpec(podUID types.UID, volumeName, mapPath string) (*volume.Spec, error) {
+	pluginDir := plugin.host.GetVolumeDevicePluginDir(fcPluginName)
+	blkutil := util.NewBlockVolumePathHandler()
+	globalMapPathUUID, err := blkutil.FindGlobalMapPathUUIDFromPod(pluginDir, mapPath, podUID)
+	if err != nil {
+		return nil, err
+	}
+	glog.V(5).Infof("globalMapPathUUID: %v, err: %v", globalMapPathUUID, err)
+
+	// Retreive volumePluginDependentPath from globalMapPathUUID
+	// globalMapPathUUID examples:
+	//   wwn+lun: plugins/kubernetes.io/fc/volumeDevices/50060e801049cfd1-lun-0/{pod uuid}
+	//   wwid: plugins/kubernetes.io/fc/volumeDevices/3600508b400105e210000900000490000/{pod uuid}
+	arr := strings.Split(globalMapPathUUID, "/")
+	if len(arr) < 2 {
+		return nil, fmt.Errorf("Fail to retreive volume plugin information from globalMapPathUUID: %v", globalMapPathUUID)
+	}
+	l := len(arr) - 2
+	volumeInfo := arr[l]
+
+	// Create volume from wwn+lun or wwid
+	var fcPV *v1.PersistentVolume
+	if strings.Contains(volumeInfo, "-lun-") {
+		wwnLun := strings.Split(volumeInfo, "-lun-")
+		lun, err := strconv.Atoi(wwnLun[1])
+		if err != nil {
+			return nil, err
+		}
+		lun32 := int32(lun)
+		fcPV = createPersistentVolumeFromFCVolumeSource(volumeName,
+			v1.FCVolumeSource{TargetWWNs: []string{wwnLun[0]}, Lun: &lun32})
+		glog.V(5).Infof("ConstructBlockVolumeSpec: TargetWWNs: %v, Lun: %v",
+			fcPV.Spec.PersistentVolumeSource.FC.TargetWWNs,
+			fcPV.Spec.PersistentVolumeSource.FC.Lun)
+	} else {
+		fcPV = createPersistentVolumeFromFCVolumeSource(volumeName,
+			v1.FCVolumeSource{WWIDs: []string{volumeInfo}})
+		glog.V(5).Infof("ConstructBlockVolumeSpec: WWIDs: %v", fcPV.Spec.PersistentVolumeSource.FC.WWIDs)
+	}
+	return volume.NewSpecFromPersistentVolume(fcPV, false), nil
+}
+
 type fcDisk struct {
 	volName string
 	podUID  types.UID
@@ -192,10 +308,25 @@ func (fc *fcDisk) GetPath() string {
 	return fc.plugin.host.GetPodVolumeDir(fc.podUID, utilstrings.EscapeQualifiedNameForDisk(name), fc.volName)
 }
 
+func (fc *fcDisk) fcGlobalMapPath(spec *volume.Spec) (string, error) {
+	mounter, err := volumeSpecToMounter(spec, fc.plugin.host)
+	if err != nil {
+		glog.Warningf("failed to get fc mounter: %v", err)
+		return "", err
+	}
+	return fc.manager.MakeGlobalVDPDName(*mounter.fcDisk), nil
+}
+
+func (fc *fcDisk) fcPodDeviceMapPath() (string, string) {
+	name := fcPluginName
+	return fc.plugin.host.GetPodVolumeDeviceDir(fc.podUID, utilstrings.EscapeQualifiedNameForDisk(name)), fc.volName
+}
+
 type fcDiskMounter struct {
 	*fcDisk
 	readOnly   bool
 	fsType     string
+	volumeMode v1.PersistentVolumeMode
 	mounter    *mount.SafeFormatAndMount
 }
 
@@ -246,7 +377,52 @@ func (c *fcDiskUnmounter) TearDownAt(dir string) error {
 	return util.UnmountPath(dir, c.mounter)
 }
 
+// Block Volumes Support
+type fcDiskMapper struct {
+	*fcDisk
+	readOnly bool
+	mounter  mount.Interface
+}
+
+var _ volume.BlockVolumeMapper = &fcDiskMapper{}
+
+func (b *fcDiskMapper) SetUpDevice() (string, error) {
+	return "", nil
+}
+
+type fcDiskUnmapper struct {
+	*fcDisk
+}
+
+var _ volume.BlockVolumeUnmapper = &fcDiskUnmapper{}
+
+func (c *fcDiskUnmapper) TearDownDevice(_, devicePath string) error {
+	// Remove scsi device from the node.
+	if !strings.HasPrefix(devicePath, "/dev/") {
+		return fmt.Errorf("fc detach disk: invalid device name: %s", devicePath)
+	}
+	arr := strings.Split(devicePath, "/")
+	dev := arr[len(arr)-1]
+	removeFromScsiSubsystem(dev, c.io)
+	return nil
+}
+
+// GetGlobalMapPath returns global map path and error
+// path: plugins/kubernetes.io/{PluginName}/volumeDevices/{WWID}/{podUid}
+func (fc *fcDisk) GetGlobalMapPath(spec *volume.Spec) (string, error) {
+	return fc.fcGlobalMapPath(spec)
+}
+
+// GetPodDeviceMapPath returns pod device map path and volume name
+// path: pods/{podUid}/volumeDevices/kubernetes.io~fc
+// volumeName: pv0001
+func (fc *fcDisk) GetPodDeviceMapPath() (string, string) {
+	return fc.fcPodDeviceMapPath()
+}
+
 func getVolumeSource(spec *volume.Spec) (*v1.FCVolumeSource, bool, error) {
+	// fc volumes used directly in a pod have a ReadOnly flag set by the pod author.
+	// fc volumes used as a PersistentVolume gets the ReadOnly flag indirectly through the persistent-claim volume used to mount the PV
 	if spec.Volume != nil && spec.Volume.FC != nil {
 		return spec.Volume.FC, spec.Volume.FC.ReadOnly, nil
 	} else if spec.PersistentVolume != nil &&
@@ -256,3 +432,34 @@ func getVolumeSource(spec *volume.Spec) (*v1.FCVolumeSource, bool, error) {
 
 	return nil, false, fmt.Errorf("Spec does not reference a FibreChannel volume type")
 }
+
+func createPersistentVolumeFromFCVolumeSource(volumeName string, fc v1.FCVolumeSource) *v1.PersistentVolume {
+	block := v1.PersistentVolumeBlock
+	return &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: volumeName,
+		},
+		Spec: v1.PersistentVolumeSpec{
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				FC: &fc,
+			},
+			VolumeMode: &block,
+		},
+	}
+}
+
+func getWwnsLunWwids(fc *v1.FCVolumeSource) ([]string, string, []string, error) {
+	var lun string
+	var wwids []string
+	if fc.Lun != nil && len(fc.TargetWWNs) != 0 {
+		lun = strconv.Itoa(int(*fc.Lun))
+		return fc.TargetWWNs, lun, wwids, nil
+	}
+	if len(fc.WWIDs) != 0 {
+		for _, wwid := range fc.WWIDs {
+			wwids = append(wwids, strings.Replace(wwid, " ", "_", -1))
+		}
+		return fc.TargetWWNs, lun, wwids, nil
+	}
+	return nil, "", nil, fmt.Errorf("fc: no fc disk information found")
+}

pkg/volume/fc/fc_test.go

@@ -91,6 +91,11 @@ func (fake *fakeDiskManager) Cleanup() {
 func (fake *fakeDiskManager) MakeGlobalPDName(disk fcDisk) string {
 	return fake.tmpDir
 }
+
+func (fake *fakeDiskManager) MakeGlobalVDPDName(disk fcDisk) string {
+	return fake.tmpDir
+}
+
 func (fake *fakeDiskManager) AttachDisk(b fcDiskMounter) (string, error) {
 	globalPath := b.manager.MakeGlobalPDName(*b.fcDisk)
 	err := os.MkdirAll(globalPath, 0750)
@@ -361,3 +366,40 @@ func TestPersistentClaimReadOnlyFlag(t *testing.T) {
 		t.Errorf("Expected true for mounter.IsReadOnly")
 	}
 }
+
+func Test_getWwnsLun(t *testing.T) {
+	num := int32(0)
+	fc := &v1.FCVolumeSource{
+		TargetWWNs: []string{"500a0981891b8dc5"},
+		FSType:     "ext4",
+		Lun:        &num,
+	}
+	wwn, lun, _, err := getWwnsLunWwids(fc)
+	// if no wwn and lun, exit
+	if (len(wwn) == 0 && lun != "0") || err != nil {
+		t.Errorf("no fc disk found")
+	}
+}
+
+func Test_getWwids(t *testing.T) {
+	fc := &v1.FCVolumeSource{
+		FSType: "ext4",
+		WWIDs:  []string{"3600508b400105e210000900000490000"},
+	}
+	_, _, wwid, err := getWwnsLunWwids(fc)
+	// if no wwn and lun, exit
+	if len(wwid) == 0 || err != nil {
+		t.Errorf("no fc disk found")
+	}
+}
+
+func Test_getWwnsLunWwidsError(t *testing.T) {
+	fc := &v1.FCVolumeSource{
+		FSType: "ext4",
+	}
+	wwn, lun, wwid, err := getWwnsLunWwids(fc)
+	// expected no wwn and lun and wwid
+	if (len(wwn) != 0 && lun != "" && len(wwid) != 0) || err == nil {
+		t.Errorf("unexpected fc disk found")
+	}
+}

pkg/volume/fc/fc_util.go

@@ -25,6 +25,9 @@ import (
 	"strings"
 
 	"github.com/golang/glog"
+	"k8s.io/api/core/v1"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
 )
 
@@ -143,7 +146,7 @@ func scsiHostRescan(io ioHandler) {
 	}
 }
 
-// make a directory like /var/lib/kubelet/plugins/kubernetes.io/pod/fc/target-lun-0
+// make a directory like /var/lib/kubelet/plugins/kubernetes.io/fc/target-lun-0
 func makePDNameInternal(host volume.VolumeHost, wwns []string, lun string, wwids []string) string {
 	if len(wwns) != 0 {
 		return path.Join(host.GetPluginDir(fcPluginName), wwns[0]+"-lun-"+lun)
@@ -152,13 +155,27 @@ func makePDNameInternal(host volume.VolumeHost, wwns []string, lun string, wwids
 	}
 }
 
+// make a directory like /var/lib/kubelet/plugins/kubernetes.io/fc/volumeDevices/target-lun-0
+func makeVDPDNameInternal(host volume.VolumeHost, wwns []string, lun string, wwids []string) string {
+	if len(wwns) != 0 {
+		return path.Join(host.GetVolumeDevicePluginDir(fcPluginName), wwns[0]+"-lun-"+lun)
+	} else {
+		return path.Join(host.GetVolumeDevicePluginDir(fcPluginName), wwids[0])
+	}
+}
+
 type FCUtil struct{}
 
 func (util *FCUtil) MakeGlobalPDName(fc fcDisk) string {
 	return makePDNameInternal(fc.plugin.host, fc.wwns, fc.lun, fc.wwids)
 }
 
-func searchDisk(b fcDiskMounter) (string, string) {
+// Global volume device plugin dir
+func (util *FCUtil) MakeGlobalVDPDName(fc fcDisk) string {
+	return makeVDPDNameInternal(fc.plugin.host, fc.wwns, fc.lun, fc.wwids)
+}
+
+func searchDisk(b fcDiskMounter) (string, error) {
 	var diskIds []string
 	var disk string
 	var dm string
@@ -198,14 +215,6 @@ func searchDisk(b fcDiskMounter) (string, string) {
 		scsiHostRescan(io)
 		rescaned = true
 	}
-	return disk, dm
-}
-
-func (util *FCUtil) AttachDisk(b fcDiskMounter) (string, error) {
-	devicePath := ""
-	var disk, dm string
-
-	disk, dm = searchDisk(b)
 	// if no disk matches input wwn and lun, exit
 	if disk == "" && dm == "" {
 		return "", fmt.Errorf("no fc disk found")
@@ -213,10 +222,26 @@ func (util *FCUtil) AttachDisk(b fcDiskMounter) (string, error) {
 
 	// if multipath devicemapper device is found, use it; otherwise use raw disk
 	if dm != "" {
-		devicePath = dm
-	} else {
-		devicePath = disk
+		return dm, nil
 	}
+	return disk, nil
+}
+
+func (util *FCUtil) AttachDisk(b fcDiskMounter) (string, error) {
+	devicePath, err := searchDisk(b)
+	if err != nil {
+		return "", err
+	}
+	// TODO: remove feature gate check after no longer needed
+	if utilfeature.DefaultFeatureGate.Enabled(features.BlockVolume) {
+		// If the volumeMode is 'Block', plugin don't have to format the volume.
+		// The globalPDPath will be created by operationexecutor. Just return devicePath here.
+		glog.V(5).Infof("fc: AttachDisk volumeMode: %s, devicePath: %s", b.volumeMode, devicePath)
+		if b.volumeMode == v1.PersistentVolumeBlock {
+			return devicePath, nil
+		}
+	}
+
 	// mount it
 	globalPDPath := util.MakeGlobalPDName(*b.fcDisk)
 	if err := os.MkdirAll(globalPDPath, 0750); err != nil {

pkg/volume/fc/fc_util_test.go

@@ -92,9 +92,9 @@ func TestSearchDisk(t *testing.T) {
 			io:   &fakeIOHandler{},
 		},
 	}
-	disk, dm := searchDisk(fakeMounter)
+	devicePath, error := searchDisk(fakeMounter)
 	// if no disk matches input wwn and lun, exit
-	if disk == "" && dm == "" {
+	if devicePath == "" || error != nil {
 		t.Errorf("no fc disk found")
 	}
 }
@@ -106,9 +106,9 @@ func TestSearchDiskWWID(t *testing.T) {
 			io:    &fakeIOHandler{},
 		},
 	}
-	disk, dm := searchDisk(fakeMounter)
+	devicePath, error := searchDisk(fakeMounter)
 	// if no disk matches input wwid, exit
-	if disk == "" && dm == "" {
+	if devicePath == "" || error != nil {
 		t.Errorf("no fc disk found")
 	}
 }