cephfs: add read affinity mount option

This commit makes use of crush location labels from node
labels to supply `crush_location` and `read_from_replica=localize`
options during mount. Using these options, cephfs
will be able to redirect reads to the closest OSD,
improving performance.

Signed-off-by: Praveen M <m.praveen@ibm.com>

internal/cephfs/driver.go

@@ -28,6 +28,7 @@ import (
 	hc "github.com/ceph/ceph-csi/internal/health-checker"
 	"github.com/ceph/ceph-csi/internal/journal"
 	"github.com/ceph/ceph-csi/internal/util"
+	"github.com/ceph/ceph-csi/internal/util/k8s"
 	"github.com/ceph/ceph-csi/internal/util/log"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -74,24 +75,29 @@ func NewControllerServer(d *csicommon.CSIDriver) *ControllerServer {
 func NewNodeServer(
 	d *csicommon.CSIDriver,
 	t string,
-	topology map[string]string,
 	kernelMountOptions string,
 	fuseMountOptions string,
+	nodeLabels, topology, crushLocationMap map[string]string,
 ) *NodeServer {
-	return &NodeServer{
-		DefaultNodeServer:  csicommon.NewDefaultNodeServer(d, t, topology),
+	cliReadAffinityMapOptions := util.ConstructReadAffinityMapOption(crushLocationMap)
+	ns := &NodeServer{
+		DefaultNodeServer:  csicommon.NewDefaultNodeServer(d, t, cliReadAffinityMapOptions, topology, nodeLabels),
 		VolumeLocks:        util.NewVolumeLocks(),
 		kernelMountOptions: kernelMountOptions,
 		fuseMountOptions:   fuseMountOptions,
 		healthChecker:      hc.NewHealthCheckManager(),
 	}
+
+	return ns
 }
 
 // Run start a non-blocking grpc controller,node and identityserver for
 // ceph CSI driver which can serve multiple parallel requests.
 func (fs *Driver) Run(conf *util.Config) {
-	var err error
-	var topology map[string]string
+	var (
+		err                                    error
+		nodeLabels, topology, crushLocationMap map[string]string
+	)
 
 	// Configuration
 	if err = mounter.LoadAvailableMounters(conf); err != nil {
@@ -102,6 +108,18 @@ func (fs *Driver) Run(conf *util.Config) {
 	if conf.InstanceID != "" {
 		CSIInstanceID = conf.InstanceID
 	}
+
+	if conf.IsNodeServer && k8s.RunsOnKubernetes() {
+		nodeLabels, err = k8s.GetNodeLabels(conf.NodeID)
+		if err != nil {
+			log.FatalLogMsg(err.Error())
+		}
+	}
+
+	if conf.EnableReadAffinity {
+		crushLocationMap = util.GetCrushLocationMap(conf.CrushLocationLabels, nodeLabels)
+	}
+
 	// Create an instance of the volume journal
 	store.VolJournal = journal.NewCSIVolumeJournalWithNamespace(CSIInstanceID, fsutil.RadosNamespace)
 
@@ -138,7 +156,11 @@ func (fs *Driver) Run(conf *util.Config) {
 		if err != nil {
 			log.FatalLogMsg(err.Error())
 		}
-		fs.ns = NewNodeServer(fs.cd, conf.Vtype, topology, conf.KernelMountOptions, conf.FuseMountOptions)
+		fs.ns = NewNodeServer(
+			fs.cd, conf.Vtype,
+			conf.KernelMountOptions, conf.FuseMountOptions,
+			nodeLabels, topology, crushLocationMap,
+		)
 	}
 
 	if conf.IsControllerServer {
@@ -151,7 +173,11 @@ func (fs *Driver) Run(conf *util.Config) {
 		if err != nil {
 			log.FatalLogMsg(err.Error())
 		}
-		fs.ns = NewNodeServer(fs.cd, conf.Vtype, topology, conf.KernelMountOptions, conf.FuseMountOptions)
+		fs.ns = NewNodeServer(
+			fs.cd, conf.Vtype,
+			conf.KernelMountOptions, conf.FuseMountOptions,
+			nodeLabels, topology, crushLocationMap,
+		)
 		fs.cs = NewControllerServer(fs.cd)
 	}
 

internal/cephfs/nodeserver.go

@@ -766,11 +766,12 @@ func (ns *NodeServer) setMountOptions(
 	csiConfigFile string,
 ) error {
 	var (
-		configuredMountOptions string
-		kernelMountOptions     string
-		fuseMountOptions       string
-		mountOptions           []string
-		err                    error
+		configuredMountOptions   string
+		readAffinityMountOptions string
+		kernelMountOptions       string
+		fuseMountOptions         string
+		mountOptions             []string
+		err                      error
 	)
 	if m := volCap.GetMount(); m != nil {
 		mountOptions = m.GetMountFlags()
@@ -781,6 +782,14 @@ func (ns *NodeServer) setMountOptions(
 		if err != nil {
 			return err
 		}
+
+		// read affinity mount options
+		readAffinityMountOptions, err = util.GetReadAffinityMapOptions(
+			csiConfigFile, volOptions.ClusterID, ns.CLIReadAffinityOptions, ns.NodeLabels,
+		)
+		if err != nil {
+			return err
+		}
 	}
 
 	switch mnt.(type) {
@@ -799,6 +808,7 @@ func (ns *NodeServer) setMountOptions(
 			configuredMountOptions = kernelMountOptions
 		}
 		volOptions.KernelMountOptions = util.MountOptionsAdd(volOptions.KernelMountOptions, configuredMountOptions)
+		volOptions.KernelMountOptions = util.MountOptionsAdd(volOptions.KernelMountOptions, readAffinityMountOptions)
 		volOptions.KernelMountOptions = util.MountOptionsAdd(volOptions.KernelMountOptions, mountOptions...)
 	}
 

internal/cephfs/nodeserver_test.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/ceph/ceph-csi/internal/cephfs/mounter"
 	"github.com/ceph/ceph-csi/internal/cephfs/store"
+	csicommon "github.com/ceph/ceph-csi/internal/csi-common"
 	"github.com/ceph/ceph-csi/internal/util"
 )
 
@@ -63,19 +64,19 @@ func Test_setMountOptions(t *testing.T) {
 	t.Logf("path = %s", tmpConfPath)
 	err = os.WriteFile(tmpConfPath, csiConfigFileContent, 0o600)
 	if err != nil {
-		t.Errorf("failed to write %s file content: %v", util.CsiConfigFile, err)
+		t.Errorf("failed to write %s file content: %v", tmpConfPath, err)
 	}
 
 	tests := []struct {
 		name       string
-		ns         NodeServer
+		ns         *NodeServer
 		mnt        mounter.VolumeMounter
 		volOptions *store.VolumeOptions
 		want       string
 	}{
 		{
 			name: "KernelMountOptions set in cluster-1 config and not set in CLI",
-			ns:   NodeServer{},
+			ns:   &NodeServer{},
 			mnt:  mounter.VolumeMounter(&mounter.KernelMounter{}),
 			volOptions: &store.VolumeOptions{
 				ClusterID: "cluster-1",
@@ -84,7 +85,7 @@ func Test_setMountOptions(t *testing.T) {
 		},
 		{
 			name: "FuseMountOptions set in cluster-1 config and not set in CLI",
-			ns:   NodeServer{},
+			ns:   &NodeServer{},
 			mnt:  mounter.VolumeMounter(&mounter.FuseMounter{}),
 			volOptions: &store.VolumeOptions{
 				ClusterID: "cluster-1",
@@ -93,7 +94,7 @@ func Test_setMountOptions(t *testing.T) {
 		},
 		{
 			name: "KernelMountOptions set in cluster-1 config and set in CLI",
-			ns: NodeServer{
+			ns: &NodeServer{
 				kernelMountOptions: cliKernelMountOptions,
 			},
 			mnt: mounter.VolumeMounter(&mounter.KernelMounter{}),
@@ -104,7 +105,7 @@ func Test_setMountOptions(t *testing.T) {
 		},
 		{
 			name: "FuseMountOptions not set in cluster-2 config and set in CLI",
-			ns: NodeServer{
+			ns: &NodeServer{
 				fuseMountOptions: cliFuseMountOptions,
 			},
 			mnt: mounter.VolumeMounter(&mounter.FuseMounter{}),
@@ -115,7 +116,7 @@ func Test_setMountOptions(t *testing.T) {
 		},
 		{
 			name: "KernelMountOptions not set in cluster-2 config and set in CLI",
-			ns: NodeServer{
+			ns: &NodeServer{
 				kernelMountOptions: cliKernelMountOptions,
 			},
 			mnt: mounter.VolumeMounter(&mounter.KernelMounter{}),
@@ -126,7 +127,7 @@ func Test_setMountOptions(t *testing.T) {
 		},
 		{
 			name: "FuseMountOptions not set in cluster-1 config and set in CLI",
-			ns: NodeServer{
+			ns: &NodeServer{
 				fuseMountOptions: cliFuseMountOptions,
 			},
 			mnt: mounter.VolumeMounter(&mounter.FuseMounter{}),
@@ -146,6 +147,11 @@ func Test_setMountOptions(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 
+			driver := &csicommon.CSIDriver{}
+			tc.ns.DefaultNodeServer = csicommon.NewDefaultNodeServer(
+				driver, "cephfs", "", map[string]string{}, map[string]string{},
+			)
+
 			err := tc.ns.setMountOptions(tc.mnt, tc.volOptions, volCap, tmpConfPath)
 			if err != nil {
 				t.Errorf("setMountOptions() = %v", err)