e2e: added test to verify read affinity functionality

e2e test case is added to test if read affinity is enabled by
verifying read_from_replica=localize option is passed

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

e2e/deployment.go

@@ -231,15 +231,19 @@ func (yr *yamlResource) Do(action kubectlAction) error {
 // replaceNamespaceInTemplate() on it. There are several options for adjusting
 // templates, each has their own comment.
 type yamlResourceNamespaced struct {
-	filename  string
-	namespace string
+	filename            string
+	namespace           string
+	domainLabel         string
+	crushLocationLabels string
 
 	// set the number of replicas in a Deployment to 1.
 	oneReplica bool
 
 	// enable topology support (for RBD)
 	enableTopology bool
-	domainLabel    string
+
+	// enable read affinity support (for RBD)
+	enableReadAffinity bool
 }
 
 func (yrn *yamlResourceNamespaced) Do(action kubectlAction) error {
@@ -260,6 +264,14 @@ func (yrn *yamlResourceNamespaced) Do(action kubectlAction) error {
 		data = addTopologyDomainsToDSYaml(data, yrn.domainLabel)
 	}
 
+	if yrn.enableReadAffinity {
+		data = enableReadAffinityInTemplate(data)
+	}
+
+	if yrn.crushLocationLabels != "" {
+		data = addCrsuhLocationLabels(data, yrn.crushLocationLabels)
+	}
+
 	err = retryKubectlInput(yrn.namespace, action, data, deployTimeout)
 	if err != nil {
 		return fmt.Errorf("failed to %s resource %q in namespace %q: %w", action, yrn.filename, yrn.namespace, err)

e2e/pod.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"regexp"
 	"strings"
 	"time"
 
@@ -623,3 +624,83 @@ func verifySeLinuxMountOption(
 
 	return nil
 }
+
+func verifyReadAffinity(
+	f *framework.Framework,
+	pvcPath, appPath, daemonSetName, cn, ns string,
+) error {
+	readFromReplicaOption := "read_from_replica=localize"
+	expectedCrushLocationValues := map[string]string{
+		strings.Split(crushLocationRegionLabel, "/")[1]: crushLocationRegionValue,
+		strings.Split(crushLocationZoneLabel, "/")[1]:   crushLocationZoneValue,
+	}
+	actualCrushLocationValues := make(map[string]string)
+
+	// create PVC
+	pvc, err := loadPVC(pvcPath)
+	if err != nil {
+		return fmt.Errorf("failed to load pvc: %w", err)
+	}
+	pvc.Namespace = f.UniqueName
+	err = createPVCAndvalidatePV(f.ClientSet, pvc, deployTimeout)
+	if err != nil {
+		return fmt.Errorf("failed to create PVC: %w", err)
+	}
+	app, err := loadApp(appPath)
+	if err != nil {
+		return fmt.Errorf("failed to load application: %w", err)
+	}
+	app.Namespace = f.UniqueName
+	err = createApp(f.ClientSet, app, deployTimeout)
+	if err != nil {
+		return fmt.Errorf("failed to create application: %w", err)
+	}
+
+	pod, err := f.ClientSet.CoreV1().Pods(f.UniqueName).Get(context.TODO(), app.Name, metav1.GetOptions{})
+	if err != nil {
+		framework.Logf("Error occurred getting pod %s in namespace %s", app.Name, f.UniqueName)
+
+		return fmt.Errorf("failed to get pod: %w", err)
+	}
+
+	nodepluginPodName, err := getDaemonsetPodOnNode(f, daemonSetName, pod.Spec.NodeName, ns)
+	if err != nil {
+		return fmt.Errorf("failed to get daemonset pod on node: %w", err)
+	}
+	logs, err := frameworkPod.GetPodLogs(context.TODO(), f.ClientSet, ns, nodepluginPodName, cn)
+	if err != nil {
+		return fmt.Errorf("failed to get pod logs from container %s/%s/%s : %w", ns, nodepluginPodName, cn, err)
+	}
+
+	if !strings.Contains(logs, readFromReplicaOption) {
+		return fmt.Errorf("option %s not found in logs: %s", readFromReplicaOption, logs)
+	}
+
+	crushLocationPattern := "crush_location=[^]\\s]+"
+	regex := regexp.MustCompile(crushLocationPattern)
+	match := regex.FindString(logs)
+
+	if match == "" {
+		return fmt.Errorf("option crush_location not found in logs: %s", logs)
+	}
+
+	crushLocationValue := strings.Split(match, "=")[1]
+	keyValues := strings.Split(crushLocationValue, "|")
+
+	for _, keyValue := range keyValues {
+		s := strings.Split(keyValue, ":")
+		actualCrushLocationValues[s[0]] = s[1]
+	}
+	for key, expectedValue := range expectedCrushLocationValues {
+		if actualValue, exists := actualCrushLocationValues[key]; !(exists && actualValue == expectedValue) {
+			return fmt.Errorf("option crush_location=%s:%s not found in logs: %s", key, expectedValue, logs)
+		}
+	}
+
+	err = deletePVCAndApp("", f, pvc, app)
+	if err != nil {
+		return fmt.Errorf("failed to delete PVC and application: %w", err)
+	}
+
+	return nil
+}

e2e/rbd.go

@@ -65,6 +65,12 @@ var (
 	rbdTopologyPool     = "newrbdpool"
 	rbdTopologyDataPool = "replicapool" // NOTE: should be different than rbdTopologyPool for test to be effective
 
+	// CRUSH location node labels & values.
+	crushLocationRegionLabel = "topology.kubernetes.io/region"
+	crushLocationRegionValue = "east"
+	crushLocationZoneLabel   = "topology.kubernetes.io/zone"
+	crushLocationZoneValue   = "east-zone1"
+
 	// yaml files required for deployment.
 	pvcPath                = rbdExamplePath + "pvc.yaml"
 	appPath                = rbdExamplePath + "pod.yaml"
@@ -161,9 +167,11 @@ func createORDeleteRbdResources(action kubectlAction) {
 		},
 		// the node-plugin itself
 		&yamlResourceNamespaced{
-			filename:    rbdDirPath + rbdNodePlugin,
-			namespace:   cephCSINamespace,
-			domainLabel: nodeRegionLabel + "," + nodeZoneLabel,
+			filename:            rbdDirPath + rbdNodePlugin,
+			namespace:           cephCSINamespace,
+			domainLabel:         nodeRegionLabel + "," + nodeZoneLabel,
+			enableReadAffinity:  true,
+			crushLocationLabels: crushLocationRegionLabel + "," + crushLocationZoneLabel,
 		},
 	}
 
@@ -275,6 +283,14 @@ var _ = Describe("RBD", func() {
 			if err != nil {
 				framework.Failf("failed to create node label: %v", err)
 			}
+			err = createNodeLabel(f, crushLocationRegionLabel, crushLocationRegionValue)
+			if err != nil {
+				framework.Failf("failed to create node label: %v", err)
+			}
+			err = createNodeLabel(f, crushLocationZoneLabel, crushLocationZoneValue)
+			if err != nil {
+				framework.Failf("failed to create node label: %v", err)
+			}
 			if cephCSINamespace != defaultNs {
 				err = createNamespace(c, cephCSINamespace)
 				if err != nil {
@@ -444,6 +460,14 @@ var _ = Describe("RBD", func() {
 				})
 			}
 
+			By("verify readAffinity support", func() {
+				err := verifyReadAffinity(f, pvcPath, appPath,
+					rbdDaemonsetName, rbdContainerName, cephCSINamespace)
+				if err != nil {
+					framework.Failf("failed to verify readAffinity: %v", err)
+				}
+			})
+
 			By("verify mountOptions support", func() {
 				err := verifySeLinuxMountOption(f, pvcPath, appPath,
 					rbdDaemonsetName, rbdContainerName, cephCSINamespace)

e2e/utils.go

@@ -827,6 +827,15 @@ func enableTopologyInTemplate(data string) string {
 	return strings.ReplaceAll(data, "--feature-gates=Topology=false", "--feature-gates=Topology=true")
 }
 
+func enableReadAffinityInTemplate(template string) string {
+	return strings.ReplaceAll(template, "# - \"--enable-read-affinity=true\"", "- \"--enable-read-affinity=true\"")
+}
+
+func addCrsuhLocationLabels(template, labels string) string {
+	return strings.ReplaceAll(template, "# - \"--crush-location-labels=topology.io/zone,topology.io/rack\"",
+		"- \"--crush-location-labels="+labels+"\"")
+}
+
 func writeDataAndCalChecksum(app *v1.Pod, opt *metav1.ListOptions, f *framework.Framework) (string, error) {
 	filePath := app.Spec.Containers[0].VolumeMounts[0].MountPath + "/test"
 	// write data in PVC