rbd: detect krbd features in runtime and fallback to nbd

Currently, we recognize and warn for the provided image features based on
our prior intelligence at ceph-csi (i.e based on supportedFeatures map
and validateImageFeatures) at image/PV creation time. It might be very
much possible that the cluster is heterogeneous i.e. the PV creation and
application container might both be on different nodes with different
kernel versions (krbd driver versions).

This PR adds a mechanism to check for the supported krbd features during
mount time, if the krbd driver doesn't have the specified image feature
then it will fall back to rbd-nbd mounter.

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

internal/rbd/driver.go

@@ -157,6 +157,15 @@ func (r *Driver) Run(conf *util.Config) {
 		if err != nil {
 			log.FatalLogMsg("failed to start node server, err %v\n", err)
 		}
+		var attr string
+		attr, err = getKrbdSupportedFeatures()
+		if err != nil {
+			log.FatalLogMsg(err.Error())
+		}
+		krbdFeatures, err = hexStringToInteger(attr)
+		if err != nil {
+			log.FatalLogMsg(err.Error())
+		}
 	}
 
 	if conf.IsControllerServer {

internal/rbd/nodeserver.go

@@ -226,10 +226,22 @@ func populateRbdVol(
 		rv.RbdImageName = imageAttributes.ImageName
 	}
 
+	krbdSupported := false
+	if req.GetVolumeContext()["mounter"] != rbdNbdMounter {
+		krbdSupported = isKrbdFeatureSupported(ctx, req.GetVolumeContext()["imageFeatures"])
+	}
+	if krbdSupported {
+		rv.MapOptions = req.GetVolumeContext()["mapOptions"]
+		rv.UnmapOptions = req.GetVolumeContext()["unmapOptions"]
+		rv.Mounter = req.GetVolumeContext()["mounter"]
+	} else {
+		// fallback to rbd-nbd,
+		// ignore the mapOptions and unmapOptions as they are meant for krbd use.
+		rv.Mounter = rbdNbdMounter
+	}
+
 	rv.VolID = volID
-	rv.MapOptions = req.GetVolumeContext()["mapOptions"]
-	rv.UnmapOptions = req.GetVolumeContext()["unmapOptions"]
-	rv.Mounter = req.GetVolumeContext()["mounter"]
+
 	rv.LogDir = req.GetVolumeContext()["cephLogDir"]
 	if rv.LogDir == "" {
 		rv.LogDir = defaultLogDir

internal/rbd/rbd_util.go

@@ -91,6 +91,9 @@ const (
 	migImageNamePrefix = "image-"
 	// prefix in the handle for monitors field.
 	migMonPrefix = "mons-"
+
+	// krbd attribute file to check supported features.
+	krbdSupportedFeaturesFile = "/sys/bus/rbd/supported_features"
 )
 
 // rbdImage contains common attributes and methods for the rbdVolume and
@@ -197,17 +200,85 @@ type migrationVolID struct {
 	clusterID string
 }
 
-var supportedFeatures = map[string]imageFeature{
-	librbd.FeatureNameLayering: {
-		needRbdNbd: false,
-	},
-	librbd.FeatureNameExclusiveLock: {
-		needRbdNbd: true,
-	},
-	librbd.FeatureNameJournaling: {
-		needRbdNbd: true,
-		dependsOn:  []string{librbd.FeatureNameExclusiveLock},
-	},
+var (
+	supportedFeatures = map[string]imageFeature{
+		librbd.FeatureNameLayering: {
+			needRbdNbd: false,
+		},
+		librbd.FeatureNameExclusiveLock: {
+			needRbdNbd: true,
+		},
+		librbd.FeatureNameJournaling: {
+			needRbdNbd: true,
+			dependsOn:  []string{librbd.FeatureNameExclusiveLock},
+		},
+	}
+
+	krbdFeatures uint64 // krbd features supported by the loaded driver.
+)
+
+// getKrbdSupportedFeatures load the module if needed and return supported
+// features attribute as a string.
+func getKrbdSupportedFeatures() (string, error) {
+	// check if the module is loaded or compiled in
+	_, err := os.Stat(krbdSupportedFeaturesFile)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			log.ErrorLogMsg("stat on %q failed: %v", krbdSupportedFeaturesFile, err)
+
+			return "", err
+		}
+		// try to load the module
+		_, _, err = util.ExecCommand(context.TODO(), "modprobe", rbdDefaultMounter)
+		if err != nil {
+			log.ErrorLogMsg("modprobe failed: %v", err)
+
+			return "", err
+		}
+	}
+	val, err := ioutil.ReadFile(krbdSupportedFeaturesFile)
+	if err != nil {
+		log.ErrorLogMsg("reading file %q failed: %v", krbdSupportedFeaturesFile, err)
+
+		return "", err
+	}
+
+	return strings.TrimSuffix(string(val), "\n"), nil
+}
+
+// hexStringToInteger convert hex value to uint.
+func hexStringToInteger(hexString string) (uint64, error) {
+	// trim 0x prefix
+	numberStr := strings.TrimPrefix(strings.ToLower(hexString), "0x")
+
+	output, err := strconv.ParseUint(numberStr, 16, 64)
+	if err != nil {
+		log.ErrorLogMsg("converting string %q to integer failed: %v", numberStr, err)
+
+		return 0, err
+	}
+
+	return output, nil
+}
+
+// isKrbdFeatureSupported checks if a given Image Feature is supported by krbd
+// driver or not.
+func isKrbdFeatureSupported(ctx context.Context, imageFeatures string) bool {
+	arr := strings.Split(imageFeatures, ",")
+	log.UsefulLog(ctx, "checking for ImageFeatures: %v", arr)
+	imageFeatureSet := librbd.FeatureSetFromNames(arr)
+
+	supported := true
+	for _, featureName := range imageFeatureSet.Names() {
+		if (uint64(librbd.FeatureSetFromNames(strings.Split(featureName, " "))) & krbdFeatures) == 0 {
+			supported = false
+			log.ErrorLog(ctx, "krbd feature %q not supported", featureName)
+
+			break
+		}
+	}
+
+	return supported
 }
 
 // Connect an rbdVolume to the Ceph cluster.

internal/rbd/rbd_util_test.go

@@ -283,3 +283,42 @@ func TestStrategicActionOnLogFile(t *testing.T) {
 		})
 	}
 }
+
+func TestIsKrbdFeatureSupported(t *testing.T) {
+	t.Parallel()
+	ctx := context.TODO()
+
+	tests := []struct {
+		name        string
+		featureName string
+		isSupported bool
+	}{
+		{
+			name:        "supported feature",
+			featureName: "layering",
+			isSupported: true,
+		},
+		{
+			name:        "not supported feature",
+			featureName: "journaling",
+			isSupported: false,
+		},
+	}
+	for _, tt := range tests {
+		tc := tt
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			var err error
+			krbdSupportedFeaturesAttr := "0x1"
+			krbdFeatures, err = hexStringToInteger(krbdSupportedFeaturesAttr) // initialize krbdFeatures
+			if err != nil {
+				t.Errorf("hexStringToInteger(%s) failed", krbdSupportedFeaturesAttr)
+			}
+			supported := isKrbdFeatureSupported(ctx, tc.featureName)
+			if supported != tc.isSupported {
+				t.Errorf("isKrbdFeatureSupported(%s) returned supported status, expected: %t, got: %t",
+					tc.featureName, tc.isSupported, supported)
+			}
+		})
+	}
+}