csi: add option to enable read affinity for rbd volumes

This commit adds code to supply variables required
to enable read affinity for rbd volumes.
It leverages `read_from_replica=localize,crush_locations=key:value`
krbd map option from ceph to serve reads from the most local OSD.
refer:
https://docs.ceph.com/en/latest/man/8/rbd/#kernel-rbd-krbd-options

Signed-off-by: Rakshith R <rar@redhat.com>

pkg/operator/ceph/cluster/osd/osd.go

@@ -26,22 +26,23 @@ import (
 	"strings"
 	"time"
 
-	"k8s.io/client-go/kubernetes"
-
-	"github.com/coreos/pkg/capnslog"
-	"github.com/pkg/errors"
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	osdconfig "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+
+	"github.com/coreos/pkg/capnslog"
+	"github.com/pkg/errors"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/kubernetes"
 )
 
 var (
@@ -662,7 +663,7 @@ func getTopologyFromNode(ctx context.Context, clientset kubernetes.Interface, d
 	if err != nil {
 		return "", errors.Wrap(err, "failed to get the node for topology affinity")
 	}
-	_, topologyAffinity := ExtractOSDTopologyFromLabels(node.Labels)
+	_, topologyAffinity := topology.ExtractOSDTopologyFromLabels(node.Labels)
 	logger.Infof("found osd %d topology affinity at %q", osd.ID, topologyAffinity)
 	return topologyAffinity, nil
 }
@@ -722,7 +723,7 @@ func getNode(ctx context.Context, clientset kubernetes.Interface, nodeName strin
 }
 
 func updateLocationWithNodeLabels(location *[]string, nodeLabels map[string]string) string {
-	topology, topologyAffinity := ExtractOSDTopologyFromLabels(nodeLabels)
+	topology, topologyAffinity := topology.ExtractOSDTopologyFromLabels(nodeLabels)
 
 	keys := make([]string, 0, len(topology))
 	for k := range topology {

pkg/operator/ceph/cluster/osd/topology/topology.go

@@ -17,10 +17,11 @@ limitations under the License.
 // Package config provides methods for generating the Ceph config for a Ceph cluster and for
 // producing a "ceph.conf" compatible file from the config as well as Ceph command line-compatible
 // flags.
-package osd
+package topology
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	corev1 "k8s.io/api/core/v1"
@@ -121,3 +122,13 @@ func extractTopologyFromLabels(labels map[string]string) (map[string]string, str
 func formatTopologyAffinity(label, value string) string {
 	return fmt.Sprintf("%s=%s", label, value)
 }
+
+// GetDefaultTopologyLabels returns the supported default topology labels.
+func GetDefaultTopologyLabels() string {
+	Labels := []string{corev1.LabelHostname, corev1.LabelZoneRegionStable, corev1.LabelZoneFailureDomainStable}
+	for _, label := range CRUSHTopologyLabels {
+		Labels = append(Labels, topologyLabelPrefix+label)
+	}
+
+	return strings.Join(Labels, ",")
+}

pkg/operator/ceph/cluster/osd/topology/topology_test.go

@@ -17,7 +17,7 @@ limitations under the License.
 // Package config provides methods for generating the Ceph config for a Ceph cluster and for
 // producing a "ceph.conf" compatible file from the config as well as Ceph command line-compatible
 // flags.
-package osd
+package topology
 
 import (
 	"testing"
@@ -145,3 +145,17 @@ func TestTopologyLabels(t *testing.T) {
 	assert.Equal(t, "zone", topology["zone"])
 	assert.Equal(t, "topology.kubernetes.io/zone=zone", affinity)
 }
+
+func TestGetDefaultTopologyLabels(t *testing.T) {
+	expectedLabels := "kubernetes.io/hostname," +
+		"topology.kubernetes.io/region," +
+		"topology.kubernetes.io/zone," +
+		"topology.rook.io/chassis," +
+		"topology.rook.io/rack," +
+		"topology.rook.io/row," +
+		"topology.rook.io/pdu," +
+		"topology.rook.io/pod," +
+		"topology.rook.io/room," +
+		"topology.rook.io/datacenter"
+	assert.Equal(t, expectedLabels, GetDefaultTopologyLabels())
+}

pkg/operator/ceph/csi/csi.go

@@ -22,8 +22,10 @@ import (
 	"strings"
 	"time"
 
-	"github.com/pkg/errors"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology"
 	"github.com/rook/rook/pkg/operator/k8sutil"
+
+	"github.com/pkg/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/version"
 )
@@ -88,6 +90,11 @@ func (r *ReconcileCSI) setParams(ver *version.Info) error {
 		return errors.Wrap(err, "failed to parse value for 'CSI_ENABLE_HOST_NETWORK'")
 	}
 
+	if CSIParam.EnableReadAffinity, err = strconv.ParseBool(k8sutil.GetValue(r.opConfig.Parameters, "CSI_ENABLE_READ_AFFINITY", "false")); err != nil {
+		return errors.Wrap(err, "failed to parse value for 'CSI_ENABLE_READ_AFFINITY'")
+	}
+	CSIParam.CrushLocationLabels = k8sutil.GetValue(r.opConfig.Parameters, "CSI_CRUSH_LOCATION_LABELS", topology.GetDefaultTopologyLabels())
+
 	// If not set or set to anything but "false", the kernel client will be enabled
 	if strings.EqualFold(k8sutil.GetValue(r.opConfig.Parameters, "CSI_FORCE_CEPHFS_KERNEL_CLIENT", "true"), "false") {
 		CSIParam.ForceCephFSKernelClient = "false"

pkg/operator/ceph/csi/spec.go

@@ -62,6 +62,7 @@ type Param struct {
 	ImagePullPolicy                       string
 	CSIClusterName                        string
 	CSIDomainLabels                       string
+	CrushLocationLabels                   string
 	GRPCTimeout                           time.Duration
 	CSIEnableMetadata                     bool
 	EnablePluginSelinuxHostMount          bool
@@ -76,6 +77,7 @@ type Param struct {
 	EnableCSIEncryption                   bool
 	EnableCSITopology                     bool
 	EnableLiveness                        bool
+	EnableReadAffinity                    bool
 	LogLevel                              uint8
 	SidecarLogLevel                       uint8
 	CephFSGRPCMetricsPort                 uint16

pkg/operator/ceph/csi/template/rbd/csi-rbdplugin.yaml

@@ -79,6 +79,10 @@ spec:
             {{- if .EnableCSITopology }}
             - "--domainlabels={{ .CSIDomainLabels }}"
             {{- end }}
+            {{- if .EnableReadAffinity }}
+            - "--enable-read-affinity=true"
+            - "--crush-location-labels={{ .CrushLocationLabels }}"
+            {{- end }}
           env:
             - name: POD_IP
               valueFrom:

pkg/operator/ceph/disruption/clusterdisruption/pools.go

@@ -19,19 +19,18 @@ package clusterdisruption
 import (
 	"fmt"
 
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/topology"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 
 	"github.com/pkg/errors"
-	"github.com/rook/rook/pkg/operator/ceph/cluster/osd"
-
-	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
-	"github.com/rook/rook/pkg/operator/k8sutil"
 	policyv1 "k8s.io/api/policy/v1"
 	policyv1beta1 "k8s.io/api/policy/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
 func (r *ReconcileClusterDisruption) processPools(request reconcile.Request) (*cephv1.CephObjectStoreList, *cephv1.CephFilesystemList, string, int, error) {
@@ -85,11 +84,11 @@ func getMinimumFailureDomain(poolList []cephv1.PoolSpec) string {
 	}
 
 	//start with max as the min
-	minfailureDomainIndex := len(osd.CRUSHMapLevelsOrdered) - 1
+	minfailureDomainIndex := len(topology.CRUSHMapLevelsOrdered) - 1
 	matched := false
 
 	for _, pool := range poolList {
-		for index, failureDomain := range osd.CRUSHMapLevelsOrdered {
+		for index, failureDomain := range topology.CRUSHMapLevelsOrdered {
 			if index == minfailureDomainIndex {
 				// index is higher-than/equal-to the min
 				break
@@ -105,7 +104,7 @@ func getMinimumFailureDomain(poolList []cephv1.PoolSpec) string {
 		logger.Debugf("could not match failure domain. defaulting to %q", cephv1.DefaultFailureDomain)
 		return cephv1.DefaultFailureDomain
 	}
-	return osd.CRUSHMapLevelsOrdered[minfailureDomainIndex]
+	return topology.CRUSHMapLevelsOrdered[minfailureDomainIndex]
 }
 
 // Setting naive minAvailable for RGW at: n - 1