rbd: provide a way to supply mounter specific mapOptions from sc

Uses the below schema to supply mounter specific map/unmapOptions to the
nodeplugin based on the discussion we all had at
https://github.com/ceph/ceph-csi/pull/2636

This should specifically be really helpful with the `tryOthermonters`
set to true, i.e with fallback mechanism settings turned ON.

mapOption: "kbrd:v1,v2,v3;nbd:v1,v2,v3"

- By omitting `krbd:` or `nbd:`, the option(s) apply to
  rbdDefaultMounter which is krbd.
- A user can _override_ the options for a mounter by specifying `krbd:`
  or `nbd:`.
  mapOption: "v1,v2,v3;nbd:v1,v2,v3"
  is effectively the same as the 1st example.
- Sections are split by `;`.
- If users want to specify common options for both `krbd` and `nbd`,
  they should mention them twice.

Signed-off-by: Prasanna Kumar Kalever <prasanna.kalever@redhat.com>

internal/rbd/nodeserver.go

@@ -224,12 +224,14 @@ func populateRbdVol(
 			return nil, status.Errorf(codes.Internal, "unsupported krbd Feature")
 		}
 		// fallback to rbd-nbd,
-		// ignore the mapOptions and unmapOptions as they are meant for krbd use.
 		rv.Mounter = rbdNbdMounter
 	} else {
 		rv.Mounter = req.GetVolumeContext()["mounter"]
-		rv.MapOptions = req.GetVolumeContext()["mapOptions"]
-		rv.UnmapOptions = req.GetVolumeContext()["unmapOptions"]
+	}
+
+	err = getMapOptions(req, rv)
+	if err != nil {
+		return nil, err
 	}
 
 	rv.VolID = volID

internal/rbd/rbd_attach.go

@@ -29,6 +29,7 @@ import (
 	"github.com/ceph/ceph-csi/internal/util"
 	"github.com/ceph/ceph-csi/internal/util/log"
 
+	"github.com/container-storage-interface/spec/lib/go/csi"
 	"k8s.io/apimachinery/pkg/util/wait"
 )
 
@@ -222,6 +223,65 @@ func setRbdNbdToolFeatures() {
 	log.DefaultLog("NBD module loaded: %t, rbd-nbd supported features, cookie: %t", hasNBD, hasNBDCookieSupport)
 }
 
+// parseMapOptions helps parse formatted mapOptions and unmapOptions and
+// returns mounter specific options.
+func parseMapOptions(mapOptions string) (string, string, error) {
+	var krbdMapOptions, nbdMapOptions string
+	const (
+		noKeyLength = 1
+		validLength = 2
+	)
+	for _, item := range strings.Split(mapOptions, ";") {
+		var mounter, options string
+		if item == "" {
+			continue
+		}
+		s := strings.Split(item, ":")
+		switch len(s) {
+		case noKeyLength:
+			options = strings.TrimSpace(s[0])
+			krbdMapOptions = options
+		case validLength:
+			mounter = strings.TrimSpace(s[0])
+			options = strings.TrimSpace(s[1])
+			switch strings.ToLower(mounter) {
+			case accessTypeKRbd:
+				krbdMapOptions = options
+			case accessTypeNbd:
+				nbdMapOptions = options
+			default:
+				return "", "", fmt.Errorf("unknown mounter type: %q", mounter)
+			}
+		default:
+			return "", "", fmt.Errorf("badly formatted map/unmap options: %q", mapOptions)
+		}
+	}
+
+	return krbdMapOptions, nbdMapOptions, nil
+}
+
+// getMapOptions is a wrapper func, calls parse map/unmap funcs and feeds the
+// rbdVolume object.
+func getMapOptions(req *csi.NodeStageVolumeRequest, rv *rbdVolume) error {
+	krbdMapOptions, nbdMapOptions, err := parseMapOptions(req.GetVolumeContext()["mapOptions"])
+	if err != nil {
+		return err
+	}
+	krbdUnmapOptions, nbdUnmapOptions, err := parseMapOptions(req.GetVolumeContext()["unmapOptions"])
+	if err != nil {
+		return err
+	}
+	if rv.Mounter == rbdDefaultMounter {
+		rv.MapOptions = krbdMapOptions
+		rv.UnmapOptions = krbdUnmapOptions
+	} else if rv.Mounter == rbdNbdMounter {
+		rv.MapOptions = nbdMapOptions
+		rv.UnmapOptions = nbdUnmapOptions
+	}
+
+	return nil
+}
+
 func attachRBDImage(ctx context.Context, volOptions *rbdVolume, device string, cr *util.Credentials) (string, error) {
 	var err error
 

internal/rbd/rbd_attach_test.go

@@ -0,0 +1,99 @@
+/*
+Copyright 2021 The Ceph-CSI 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 rbd
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestParseMapOptions(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name              string
+		mapOption         string
+		expectKrbdOptions string
+		expectNbdOptions  string
+		expectErr         string
+	}{
+		{
+			name:              "with old format",
+			mapOption:         "kOp1,kOp2",
+			expectKrbdOptions: "kOp1,kOp2",
+			expectNbdOptions:  "",
+			expectErr:         "",
+		},
+		{
+			name:              "with new format",
+			mapOption:         "krbd:kOp1,kOp2;nbd:nOp1,nOp2",
+			expectKrbdOptions: "kOp1,kOp2",
+			expectNbdOptions:  "nOp1,nOp2",
+			expectErr:         "",
+		},
+		{
+			name:              "without krbd: label",
+			mapOption:         "kOp1,kOp2;nbd:nOp1,nOp2",
+			expectKrbdOptions: "kOp1,kOp2",
+			expectNbdOptions:  "nOp1,nOp2",
+			expectErr:         "",
+		},
+		{
+			name:              "with only nbd label",
+			mapOption:         "nbd:nOp1,nOp2",
+			expectKrbdOptions: "",
+			expectNbdOptions:  "nOp1,nOp2",
+			expectErr:         "",
+		},
+		{
+			name:              "unknown mounter used",
+			mapOption:         "xyz:xOp1,xOp2",
+			expectKrbdOptions: "",
+			expectNbdOptions:  "",
+			expectErr:         "unknown mounter type",
+		},
+		{
+			name:              "bad formatted options",
+			mapOption:         "nbd:nOp1:nOp2;",
+			expectKrbdOptions: "",
+			expectNbdOptions:  "",
+			expectErr:         "badly formatted map/unmap options",
+		},
+	}
+	for _, tt := range tests {
+		tc := tt
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			krbdOpts, nbdOpts, err := parseMapOptions(tc.mapOption)
+			if err != nil && !strings.Contains(err.Error(), tc.expectErr) {
+				// returned error
+				t.Errorf("parseMapOptions(%s) returned error, expected: %v, got: %v",
+					tc.mapOption, tc.expectErr, err)
+			}
+			if krbdOpts != tc.expectKrbdOptions {
+				// unexpected krbd option error
+				t.Errorf("parseMapOptions(%s) returned unexpected krbd options, expected :%q, got: %q",
+					tc.mapOption, tc.expectKrbdOptions, krbdOpts)
+			}
+			if nbdOpts != tc.expectNbdOptions {
+				// unexpected nbd option error
+				t.Errorf("parseMapOptions(%s) returned unexpected nbd options, expected: %q, got: %q",
+					tc.mapOption, tc.expectNbdOptions, nbdOpts)
+			}
+		})
+	}
+}