Add support for using encrypted volumes in block mode

Use the staging path as a parent folder and bind mount device mapper mapped
encrypted volume device into a file inside of that folder.

This allows us to store additional metadata and give us more control over
our tear down for example since we are not sharing control with kubernetes
as is the case for the staging path itself.

Ref: #1613

Longhorn 4883

Signed-off-by: Derek Su <derek.su@suse.com>

csi/node_server.go

@@ -140,7 +140,17 @@ func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 	}
 
 	if volumeCapability.GetBlock() != nil {
-		devicePath := volume.Controllers[0].Endpoint
+		devicePath := filepath.Join(stagingPath, volumeID)
+		_, err := os.Stat(devicePath)
+		if err != nil {
+			if !os.IsNotExist(err) {
+				return nil, status.Errorf(codes.Internal, errors.Wrapf(err, "failed to stat device %s", devicePath).Error())
+			}
+			// Fall back to the controller endpoint if the device path under the stagingPath doesn't exist
+			log.Infof("Device path %s doesn't exist, falling back to controller endpoint %s", devicePath, volume.Controllers[0].Endpoint)
+			devicePath = volume.Controllers[0].Endpoint
+		}
+
 		if err := ns.nodePublishBlockVolume(volumeID, devicePath, targetPath, mounter); err != nil {
 			log.WithError(err).Errorf("Failed to publish BlockVolume %s", volumeID)
 			return nil, err
@@ -268,6 +278,13 @@ func (ns *NodeServer) nodeStageMountVolume(volumeID, devicePath, targetPath, fsT
 	return nil
 }
 
+// nodeStageBlockVolume utilizes the stagingPath to create a volumeID file to bind mount the devicePath
+// this is valid since the csi plugin is in control of the staging path
+func (ns *NodeServer) nodeStageBlockVolume(volumeID, devicePath, stagingPath string, mounter mount.Interface) error {
+	stagingPath = filepath.Join(stagingPath, volumeID)
+	return ns.nodePublishBlockVolume(volumeID, devicePath, stagingPath, mounter)
+}
+
 func (ns *NodeServer) nodePublishBlockVolume(volumeID, devicePath, targetPath string, mounter mount.Interface) error {
 	log := getLoggerForCSINodeServer()
 	log = log.WithFields(logrus.Fields{"function": "nodePublishBlockVolume"})
@@ -368,13 +385,6 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 		return nil, status.Errorf(codes.Aborted, "volume %s is not ready for workloads", volumeID)
 	}
 
-	devicePath := volume.Controllers[0].Endpoint
-
-	// do nothing for block devices, since they are handled by publish
-	if volumeCapability.GetBlock() != nil {
-		return &csi.NodeStageVolumeResponse{}, nil
-	}
-
 	if requiresSharedAccess(volume, volumeCapability) && !volume.Migratable {
 		if volume.AccessMode != string(longhorn.AccessModeReadWriteMany) {
 			return nil, status.Errorf(codes.FailedPrecondition, "volume %s requires shared access but is not marked for shared use", volumeID)
@@ -399,18 +409,8 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 		return &csi.NodeStageVolumeResponse{}, nil
 	}
 
-	options := volumeCapability.GetMount().GetMountFlags()
-	fsType := volumeCapability.GetMount().GetFsType()
-	if fsType == "" {
-		fsType = defaultFsType
-	}
-
-	formatMounter, ok := mounter.(*mount.SafeFormatAndMount)
-	if !ok {
-		return nil, status.Errorf(codes.Internal, "volume %v cannot get format mounter that support filesystem %v creation", volumeID, fsType)
-	}
-
-	diskFormat, err := formatMounter.GetDiskFormat(devicePath)
+	devicePath := volume.Controllers[0].Endpoint
+	diskFormat, err := getDiskFormat(devicePath)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, errors.Wrapf(err, "failed to evaluate device filesystem %v format", devicePath).Error())
 	}
@@ -453,6 +453,26 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 		devicePath = cryptoDevice
 	}
 
+	if volumeCapability.GetBlock() != nil {
+		if err := ns.nodeStageBlockVolume(volumeID, devicePath, targetPath, mounter); err != nil {
+			return nil, err
+		}
+
+		logrus.Infof("Volume %v device %v available for usage as block device", volumeID, devicePath)
+		return &csi.NodeStageVolumeResponse{}, nil
+	}
+
+	options := volumeCapability.GetMount().GetMountFlags()
+	fsType := volumeCapability.GetMount().GetFsType()
+	if fsType == "" {
+		fsType = defaultFsType
+	}
+
+	formatMounter, ok := mounter.(*mount.SafeFormatAndMount)
+	if !ok {
+		return nil, status.Errorf(codes.Internal, "volume %v cannot get format mounter that support filesystem %v creation", volumeID, fsType)
+	}
+
 	if err := ns.nodeStageMountVolume(volumeID, devicePath, targetPath, fsType, options, formatMounter); err != nil {
 		return nil, err
 	}
@@ -497,16 +517,29 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 		return nil, status.Error(codes.InvalidArgument, "volume id missing in request")
 	}
 
+	mounter := mount.New("")
+
 	// CO owns the staging_path so we only unmount but not remove the path
-	if err := unmount(targetPath, mount.New("")); err != nil {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("failed to unmount volume %s mount point %v error %v", volumeID, targetPath, err))
+	if err := unmount(targetPath, mounter); err != nil {
+		return nil, status.Error(codes.Internal, errors.Wrapf(err, "failed to unmount volume %s mount point %v", volumeID, targetPath).Error())
+	}
+
+	// For block mode we use the staging path as parent, so we have to do additional cleanup of the subfolder/files
+	// we should transition the regular fs mounts to also use the same sub folder, this allows us to store additional
+	// metadata as well as do more forcefully removals since we no longer share the control of the staging_path with kubernetes
+	//
+	// The unmount of the parent is a no op for block mode, this is also important for backwards compatibility of the existing block devices.
+	deviceFilePath := filepath.Join(targetPath, volumeID)
+	if err := cleanupMountPoint(deviceFilePath, mounter); err != nil {
+		return nil, status.Error(codes.Internal, errors.Wrapf(err, "failed to clean up volume %s device mount point %v", volumeID, deviceFilePath).Error())
 	}
 
 	// optionally try to retrieve the volume and check if it's an RWX volume
 	// if it is we let the share-manager clean up the crypto device
 	volume, _ := ns.apiClient.Volume.ById(volumeID)
-	cleanupCryptoDevice := !requiresSharedAccess(volume, nil)
 
+	// Currently, only "RWO volumes" and "block device with volume.Migratable is true" supports encryption.
+	cleanupCryptoDevice := !requiresSharedAccess(volume, nil) || (volume != nil && volume.Migratable)
 	if cleanupCryptoDevice {
 		cryptoDevice := crypto.VolumeMapper(volumeID)
 		if isOpen, err := crypto.IsDeviceOpen(cryptoDevice); err != nil {

csi/util.go

@@ -309,6 +309,11 @@ func isBlockDevice(volumePath string) (bool, error) {
 	return false, nil
 }
 
+func getDiskFormat(devicePath string) (string, error) {
+	m := mount.SafeFormatAndMount{Interface: mount.New(""), Exec: utilexec.New()}
+	return m.GetDiskFormat(devicePath)
+}
+
 func getFilesystemStatistics(volumePath string) (*volumeFilesystemStatistics, error) {
 	var statfs unix.Statfs_t
 	// See http://man7.org/linux/man-pages/man2/statfs.2.html for details.