rbd: add exclusive-lock and journaling image features for rbd image

Current rbd plugin only supports the layering feature
for rbd image. Add exclusive-lock and journaling image
features for the rbd.

Signed-off-by: Madhu Rajanna <madhupr007@gmail.com>
Signed-off-by: woohhan <woohyung_han@tmax.co.kr>

internal/rbd/rbd_util.go

@@ -46,6 +46,7 @@ const (
 	rbdImageWatcherFactor    = 1.4
 	rbdImageWatcherSteps     = 10
 	rbdDefaultMounter        = "rbd"
+	rbdNbdMounter            = "rbd-nbd"
 
 	// Output strings returned during invocation of "ceph rbd task add remove <imagespec>" when
 	// command is not supported by ceph manager. Used to check errors and recover when the command
@@ -142,8 +143,27 @@ type rbdSnapshot struct {
 	SizeBytes      int64
 }
 
+// imageFeature represents required image features and value.
+type imageFeature struct {
+	// needRbdNbd indicates whether this image feature requires an rbd-nbd mounter
+	needRbdNbd bool
+	// dependsOn is the image features required for this imageFeature
+	dependsOn []string
+}
+
 var (
-	supportedFeatures = sets.NewString(librbd.FeatureNameLayering)
+	supportedFeatures = map[string]imageFeature{
+		librbd.FeatureNameLayering: {
+			needRbdNbd: false,
+		},
+		librbd.FeatureNameExclusiveLock: {
+			needRbdNbd: true,
+		},
+		librbd.FeatureNameJournaling: {
+			needRbdNbd: true,
+			dependsOn:  []string{librbd.FeatureNameExclusiveLock},
+		},
+	}
 )
 
 // Connect an rbdVolume to the Ceph cluster.
@@ -889,28 +909,18 @@ func genVolFromVolumeOptions(ctx context.Context, volOptions, credentials map[st
 	if err != nil {
 		return nil, err
 	}
-	// if no image features is provided, it results in empty string
-	// which disable all RBD image features as we expected
-
-	imageFeatures, found := volOptions["imageFeatures"]
-	if found {
-		arr := strings.Split(imageFeatures, ",")
-		for _, f := range arr {
-			if !supportedFeatures.Has(f) {
-				return nil, fmt.Errorf("invalid feature %q for volume csi-rbdplugin, supported"+
-					" features are: %v", f, supportedFeatures)
-			}
-		}
-		rbdVol.imageFeatureSet = librbd.FeatureSetFromNames(arr)
-	}
-
-	util.ExtendedLog(ctx, "setting disableInUseChecks on rbd volume to: %v", disableInUseChecks)
-	rbdVol.DisableInUseChecks = disableInUseChecks
-
-	rbdVol.Mounter, ok = volOptions["mounter"]
-	if !ok {
+	if rbdVol.Mounter, ok = volOptions["mounter"]; !ok {
 		rbdVol.Mounter = rbdDefaultMounter
 	}
+	// if no image features is provided, it results in empty string
+	// which disable all RBD image features as we expected
+	if err = rbdVol.validateImageFeatures(volOptions["imageFeatures"]); err != nil {
+		util.ErrorLog(ctx, "failed to validate image features %v", err)
+		return nil, err
+	}
+
+	util.ExtendedLog(ctx, "setting disableInUseChecks: %t image features: %v mounter: %s", disableInUseChecks, rbdVol.imageFeatureSet.Names(), rbdVol.Mounter)
+	rbdVol.DisableInUseChecks = disableInUseChecks
 
 	err = rbdVol.initKMS(ctx, volOptions, credentials)
 	if err != nil {
@@ -920,6 +930,29 @@ func genVolFromVolumeOptions(ctx context.Context, volOptions, credentials map[st
 	return rbdVol, nil
 }
 
+func (rv *rbdVolume) validateImageFeatures(imageFeatures string) error {
+	arr := strings.Split(imageFeatures, ",")
+	featureSet := sets.NewString(arr...)
+	for _, f := range arr {
+		sf, found := supportedFeatures[f]
+		if !found {
+			return fmt.Errorf("invalid feature %s", f)
+		}
+
+		for _, r := range sf.dependsOn {
+			if !featureSet.Has(r) {
+				return fmt.Errorf("feature %s requires %s to be set", f, r)
+			}
+		}
+
+		if sf.needRbdNbd && rv.Mounter != rbdNbdMounter {
+			return fmt.Errorf("feature %s requires rbd-nbd for mounter", f)
+		}
+	}
+	rv.imageFeatureSet = librbd.FeatureSetFromNames(arr)
+	return nil
+}
+
 func genSnapFromOptions(ctx context.Context, rbdVol *rbdVolume, snapOptions map[string]string) (*rbdSnapshot, error) {
 	var err error
 

internal/rbd/rbd_util_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	librbd "github.com/ceph/go-ceph/rbd"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestHasSnapshotFeature(t *testing.T) {
@@ -42,3 +43,86 @@ func TestHasSnapshotFeature(t *testing.T) {
 		}
 	}
 }
+
+func TestValidateImageFeatures(t *testing.T) {
+	tests := []struct {
+		imageFeatures string
+		rbdVol        *rbdVolume
+		isErr         bool
+		errMsg        string
+	}{
+		{
+			"layering",
+			&rbdVolume{
+				Mounter: rbdDefaultMounter,
+			},
+			false,
+			"",
+		},
+		{
+			"layering",
+			&rbdVolume{
+				Mounter: rbdNbdMounter,
+			},
+			false,
+			"",
+		},
+		{
+			"layering,exclusive-lock,journaling",
+			&rbdVolume{
+				Mounter: rbdNbdMounter,
+			},
+			false,
+			"",
+		},
+		{
+			"layering,journaling",
+			&rbdVolume{
+				Mounter: rbdNbdMounter,
+			},
+			true,
+			"feature journaling requires exclusive-lock to be set",
+		},
+		{
+			"layering,exclusive-lock,journaling",
+			&rbdVolume{
+				Mounter: rbdDefaultMounter,
+			},
+			true,
+			"feature exclusive-lock requires rbd-nbd for mounter",
+		},
+		{
+			"layering,exclusive-lock,journaling",
+			&rbdVolume{
+				Mounter: rbdDefaultMounter,
+			},
+			true,
+			"feature exclusive-lock requires rbd-nbd for mounter",
+		},
+		{
+			"layering,exclusive-loc,journaling",
+			&rbdVolume{
+				Mounter: rbdNbdMounter,
+			},
+			true,
+			"invalid feature exclusive-loc",
+		},
+		{
+			"ayering",
+			&rbdVolume{
+				Mounter: rbdDefaultMounter,
+			},
+			true,
+			"invalid feature ayering",
+		},
+	}
+
+	for _, test := range tests {
+		err := test.rbdVol.validateImageFeatures(test.imageFeatures)
+		if test.isErr {
+			assert.EqualError(t, err, test.errMsg)
+			continue
+		}
+		assert.Nil(t, err)
+	}
+}