rbd: add support for rbd map and unmap options

added support for providing map and unmap
options to rbd CLI when mapping rbd image
on the node.

Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>

docs/deploy-rbd.md

@@ -56,6 +56,8 @@ make image-cephcsi
 | `volumeNamePrefix`                                                                                  | no                   | Prefix to use for naming RBD images (defaults to `csi-vol-`).                                                                                                                                                           |
 | `snapshotNamePrefix`                                                                                | no                   | Prefix to use for naming RBD snapshot images (defaults to `csi-snap-`).                                                                                                                                                 |
 | `imageFeatures`                                                                                     | no                   | RBD image features. CSI RBD currently supports only `layering` feature. See [man pages](http://docs.ceph.com/docs/nautilus/man/8/rbd/#cmdoption-rbd-image-feature)                                                      |
+| `mapOptions`                                                                                        | no                   | Map options to use when mapping rbd image. See [krbd](https://docs.ceph.com/docs/master/man/8/rbd/#kernel-rbd-krbd-options) and [nbd](https://docs.ceph.com/docs/master/man/8/rbd-nbd/#options) options.                |
+| `unmapOptions`                                                                                      | no                   | Unmap options to use when unmapping rbd image. See [krbd](https://docs.ceph.com/docs/master/man/8/rbd/#kernel-rbd-krbd-options) and [nbd](https://docs.ceph.com/docs/master/man/8/rbd-nbd/#options) options.            |
 | `csi.storage.k8s.io/provisioner-secret-name`, `csi.storage.k8s.io/node-stage-secret-name`           | yes (for Kubernetes) | name of the Kubernetes Secret object containing Ceph client credentials. Both parameters should have the same value                                                                                                     |
 | `csi.storage.k8s.io/provisioner-secret-namespace`, `csi.storage.k8s.io/node-stage-secret-namespace` | yes (for Kubernetes) | namespaces of the above Secret objects                                                                                                                                                                                  |
 | `mounter`                                                                                           | no                   | if set to `rbd-nbd`, use `rbd-nbd` on nodes that have `rbd-nbd` and `nbd` kernel modules to map rbd images                                                                                                              |

e2e/rbd.go

@@ -1240,6 +1240,31 @@ var _ = Describe("RBD", func() {
 				validateRBDImageCount(f, 0)
 			})
 
+			By("create a PVC and Bind it to an app for mapped rbd image with options", func() {
+				err := deleteResource(rbdExamplePath + "storageclass.yaml")
+				if err != nil {
+					e2elog.Failf("failed to delete storageclass with error %v", err)
+				}
+				err = createRBDStorageClass(f.ClientSet, f, nil, map[string]string{
+					"mapOptions":   "lock_on_read,queue_depth=1024",
+					"unmapOptions": "force"})
+				if err != nil {
+					e2elog.Failf("failed to create storageclass with error %v", err)
+				}
+				err = validatePVCAndAppBinding(pvcPath, appPath, f)
+				if err != nil {
+					e2elog.Failf("failed to validate pvc and application binding with error %v", err)
+				}
+				err = deleteResource(rbdExamplePath + "storageclass.yaml")
+				if err != nil {
+					e2elog.Failf("failed to delete storageclass with error %v", err)
+				}
+				err = createRBDStorageClass(f.ClientSet, f, nil, nil)
+				if err != nil {
+					e2elog.Failf("failed to create storageclass with error %v", err)
+				}
+			})
+
 			// Make sure this should be last testcase in this file, because
 			// it deletes pool
 			By("Create a PVC and delete PVC when backend pool deleted", func() {

examples/rbd/storageclass.yaml

@@ -29,6 +29,20 @@ parameters:
    # CSI RBD currently supports only `layering` feature.
    imageFeatures: layering
 
+   # mapOptions is a comma-separated list of map options.
+   # For krbd options refer
+   # https://docs.ceph.com/docs/master/man/8/rbd/#kernel-rbd-krbd-options
+   # For nbd options refer
+   # https://docs.ceph.com/docs/master/man/8/rbd-nbd/#options
+   # mapOptions: lock_on_read,queue_depth=1024
+
+   # unmapOptions is a comma-separated list of unmap options.
+   # For krbd options refer
+   # https://docs.ceph.com/docs/master/man/8/rbd/#kernel-rbd-krbd-options
+   # For nbd options refer
+   # https://docs.ceph.com/docs/master/man/8/rbd-nbd/#options
+   # unmapOptions: force
+
    # The secrets have to contain Ceph credentials with required access
    # to the 'pool'.
    csi.storage.k8s.io/provisioner-secret-name: csi-rbd-secret

internal/rbd/nodeserver.go

@@ -205,6 +205,9 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 	volOptions.VolID = volID
 	transaction := stageTransaction{}
 
+	volOptions.MapOptions = req.GetVolumeContext()["mapOptions"]
+	volOptions.UnmapOptions = req.GetVolumeContext()["unmapOptions"]
+
 	// Stash image details prior to mapping the image (useful during Unstage as it has no
 	// voloptions passed to the RPC as per the CSI spec)
 	err = stashRBDImageMetadata(volOptions, stagingParentPath)
@@ -213,7 +216,7 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 	}
 	defer func() {
 		if err != nil {
-			ns.undoStagingTransaction(ctx, req, transaction)
+			ns.undoStagingTransaction(ctx, req, transaction, volOptions)
 		}
 	}()
 
@@ -320,7 +323,7 @@ func (ns *NodeServer) stageTransaction(ctx context.Context, req *csi.NodeStageVo
 	return transaction, err
 }
 
-func (ns *NodeServer) undoStagingTransaction(ctx context.Context, req *csi.NodeStageVolumeRequest, transaction stageTransaction) {
+func (ns *NodeServer) undoStagingTransaction(ctx context.Context, req *csi.NodeStageVolumeRequest, transaction stageTransaction, volOptions *rbdVolume) {
 	var err error
 
 	stagingTargetPath := getStagingTargetPath(req)
@@ -345,7 +348,7 @@ func (ns *NodeServer) undoStagingTransaction(ctx context.Context, req *csi.NodeS
 
 	// Unmapping rbd device
 	if transaction.devicePath != "" {
-		err = detachRBDDevice(ctx, transaction.devicePath, volID, transaction.isEncrypted)
+		err = detachRBDDevice(ctx, transaction.devicePath, volID, volOptions.UnmapOptions, transaction.isEncrypted)
 		if err != nil {
 			util.ErrorLog(ctx, "failed to unmap rbd device: %s for volume %s with error: %v", transaction.devicePath, volID, err)
 			// continue on failure to delete the stash file, as kubernetes will fail to delete the staging path otherwise
@@ -679,7 +682,7 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 
 	// Unmapping rbd device
 	imageSpec := imgInfo.String()
-	if err = detachRBDImageOrDeviceSpec(ctx, imageSpec, true, imgInfo.NbdAccess, imgInfo.Encrypted, req.GetVolumeId()); err != nil {
+	if err = detachRBDImageOrDeviceSpec(ctx, imageSpec, true, imgInfo.NbdAccess, imgInfo.Encrypted, req.GetVolumeId(), imgInfo.UnmapOptions); err != nil {
 		util.ErrorLog(ctx, "error unmapping volume (%s) from staging path (%s): (%v)", req.GetVolumeId(), stagingTargetPath, err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}

internal/rbd/rbd_attach.go

@@ -235,13 +235,17 @@ func createPath(ctx context.Context, volOpt *rbdVolume, cr *util.Credentials) (s
 	if volOpt.readOnly {
 		mapOptions = append(mapOptions, "--read-only")
 	}
+
+	if volOpt.MapOptions != "" {
+		mapOptions = append(mapOptions, "--options", volOpt.MapOptions)
+	}
 	// Execute map
 	stdout, stderr, err := util.ExecCommand(ctx, rbd, mapOptions...)
 	if err != nil {
 		util.WarningLog(ctx, "rbd: map error %v, rbd output: %s", err, stderr)
 		// unmap rbd image if connection timeout
 		if strings.Contains(err.Error(), rbdMapConnectionTimeout) {
-			detErr := detachRBDImageOrDeviceSpec(ctx, imagePath, true, isNbd, volOpt.Encrypted, volOpt.VolID)
+			detErr := detachRBDImageOrDeviceSpec(ctx, imagePath, true, isNbd, volOpt.Encrypted, volOpt.VolID, volOpt.UnmapOptions)
 			if detErr != nil {
 				util.WarningLog(ctx, "rbd: %s unmap error %v", imagePath, detErr)
 			}
@@ -275,18 +279,18 @@ func waitForrbdImage(ctx context.Context, backoff wait.Backoff, volOptions *rbdV
 	return err
 }
 
-func detachRBDDevice(ctx context.Context, devicePath, volumeID string, encrypted bool) error {
+func detachRBDDevice(ctx context.Context, devicePath, volumeID, unmapOptions string, encrypted bool) error {
 	nbdType := false
 	if strings.HasPrefix(devicePath, "/dev/nbd") {
 		nbdType = true
 	}
 
-	return detachRBDImageOrDeviceSpec(ctx, devicePath, false, nbdType, encrypted, volumeID)
+	return detachRBDImageOrDeviceSpec(ctx, devicePath, false, nbdType, encrypted, volumeID, unmapOptions)
 }
 
 // detachRBDImageOrDeviceSpec detaches an rbd imageSpec or devicePath, with additional checking
 // when imageSpec is used to decide if image is already unmapped.
-func detachRBDImageOrDeviceSpec(ctx context.Context, imageOrDeviceSpec string, isImageSpec, ndbType, encrypted bool, volumeID string) error {
+func detachRBDImageOrDeviceSpec(ctx context.Context, imageOrDeviceSpec string, isImageSpec, ndbType, encrypted bool, volumeID, unmapOptions string) error {
 	if encrypted {
 		mapperFile, mapperPath := util.VolumeMapper(volumeID)
 		mappedDevice, mapper, err := util.DeviceEncryptionStatus(ctx, mapperPath)
@@ -312,7 +316,9 @@ func detachRBDImageOrDeviceSpec(ctx context.Context, imageOrDeviceSpec string, i
 		accessType = accessTypeNbd
 	}
 	options := []string{"unmap", "--device-type", accessType, imageOrDeviceSpec}
-
+	if unmapOptions != "" {
+		options = append(options, "--options", unmapOptions)
+	}
 	_, stderr, err := util.ExecCommand(ctx, rbd, options...)
 	if err != nil {
 		// Messages for krbd and nbd differ, hence checking either of them for missing mapping

internal/rbd/rbd_util.go

@@ -95,6 +95,8 @@ type rbdVolume struct {
 	ClusterID           string `json:"clusterId"`
 	RequestName         string
 	ReservedID          string
+	MapOptions          string
+	UnmapOptions        string
 	VolName             string `json:"volName"`
 	MonValueFromSecret  string `json:"monValueFromSecret"`
 	VolSize             int64  `json:"volSize"`
@@ -986,6 +988,7 @@ type rbdImageMetadataStash struct {
 	Pool           string `json:"pool"`
 	RadosNamespace string `json:"radosNamespace"`
 	ImageName      string `json:"image"`
+	UnmapOptions   string `json:"unmapOptions"`
 	NbdAccess      bool   `json:"accessType"`
 	Encrypted      bool   `json:"encrypted"`
 }
@@ -1006,11 +1009,12 @@ func (ri *rbdImageMetadataStash) String() string {
 func stashRBDImageMetadata(volOptions *rbdVolume, path string) error {
 	var imgMeta = rbdImageMetadataStash{
 		// there are no checks for this at present
-		Version:        2, // nolint:gomnd // number specifies version.
+		Version:        3, // nolint:gomnd // number specifies version.
 		Pool:           volOptions.Pool,
 		RadosNamespace: volOptions.RadosNamespace,
 		ImageName:      volOptions.RbdImageName,
 		Encrypted:      volOptions.Encrypted,
+		UnmapOptions:   volOptions.UnmapOptions,
 	}
 
 	imgMeta.NbdAccess = false