e2e: handle ceph-csi-operator deployment changes

This commits adds e2e/operator.go containing utility
methods specific to the operator.

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

e2e/cephfs.go

@@ -46,6 +46,9 @@ var (
 	subvolumegroup        = "e2e"
 	fileSystemName        = "myfs"
 	fileSystemPoolName    = "myfs-replicated"
+
+	operatorCephFSDeploymentName = "cephfs.csi.ceph.com-ctrlplugin"
+	operatorCephFSDaemonsetName  = "cephfs.csi.ceph.com-nodeplugin"
 )
 
 func deployCephfsPlugin() {
@@ -175,6 +178,11 @@ var _ = Describe(cephfsType, func() {
 			Skip("Skipping CephFS E2E")
 		}
 		c = f.ClientSet
+		if operatorDeployment {
+			cephFSDeploymentName = operatorCephFSDeploymentName
+			cephFSDeamonSetName = operatorCephFSDaemonsetName
+		}
+
 		if deployCephFS {
 			if cephCSINamespace != defaultNs {
 				err := createNamespace(c, cephCSINamespace)
@@ -209,11 +217,15 @@ var _ = Describe(cephfsType, func() {
 		deployVault(f.ClientSet, deployTimeout)
 
 		// wait for cluster name update in deployment
+		if operatorDeployment {
+			err = setClusterName(defaultClusterName)
+		} else {
 			containers := []string{cephFSContainerName}
 			err = waitForContainersArgsUpdate(c, cephCSINamespace, cephFSDeploymentName,
 				"clustername", defaultClusterName, containers, deployTimeout)
+		}
 		if err != nil {
-			framework.Failf("timeout waiting for deployment update %s/%s: %v", cephCSINamespace, cephFSDeploymentName, err)
+			framework.Failf("timeout waiting for clustername arg update %s/%s: %v", cephCSINamespace, rbdDeploymentName, err)
 		}
 
 		err = createSubvolumegroup(f, fileSystemName, subvolumegroup)

e2e/nfs.go

@@ -50,6 +50,9 @@ var (
 
 	// FIXME: some tests change the subvolumegroup to "e2e".
 	defaultSubvolumegroup = "csi"
+
+	operatorNFSDeploymentName  = "nfs.csi.ceph.com-ctrlplugin"
+	operatorNFSDaemonsetName   = "nfs.csi.ceph.com-nodeplugin"
 )
 
 func deployNFSPlugin(f *framework.Framework) {
@@ -242,6 +245,10 @@ var _ = Describe("nfs", func() {
 			Skip("Skipping NFS E2E")
 		}
 		c = f.ClientSet
+		if operatorDeployment {
+			nfsDeploymentName = operatorNFSDeploymentName
+			nfsDeamonSetName = operatorNFSDaemonsetName
+		}
 		if deployNFS {
 			if cephCSINamespace != defaultNs {
 				err := createNamespace(c, cephCSINamespace)

e2e/operator.go

@@ -0,0 +1,62 @@
+/*
+Copyright 2024 The Ceph-CSI Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package e2e
+
+import (
+	"fmt"
+)
+
+const (
+	OperatorConfigName = "ceph-csi-operator-config"
+	OperatorNamespace  = "ceph-csi-operator-system"
+)
+
+func setEnableMetadata(value bool) error{
+	command := []string{
+		"operatorconfigs.csi.ceph.io",
+		OperatorConfigName,
+		"--type=merge",
+		"-p",
+		fmt.Sprintf(`{"spec": {"driverSpecDefaults": {"enableMetadata": %t}}}`, value),
+	}
+
+	// Patch the operator config
+	err := retryKubectlArgs(OperatorNamespace, kubectlPatch, deployTimeout, command...)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func setClusterName(value string) error {
+	command := []string{
+		"operatorconfigs.csi.ceph.io",
+		OperatorConfigName,
+		"--type=merge",
+		"-p",
+		fmt.Sprintf(`{"spec": {"driverSpecDefaults": {"clusterName": %s}}}`, value),
+	}
+
+	// Patch the operator config
+	err := retryKubectlArgs(OperatorNamespace, kubectlPatch, deployTimeout, command...)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

e2e/rbd.go

@@ -108,6 +108,10 @@ var (
 	volSnapNameKey        = "csi.storage.k8s.io/volumesnapshot/name"
 	volSnapNamespaceKey   = "csi.storage.k8s.io/volumesnapshot/namespace"
 	volSnapContentNameKey = "csi.storage.k8s.io/volumesnapshotcontent/name"
+
+	operatorRBDDeploymentName = "rbd.csi.ceph.com-ctrlplugin"
+	operatorRBDDaemonsetName  = "rbd.csi.ceph.com-nodeplugin"
+	rbdPodSelector            = fmt.Sprintf("app in (ceph-csi-rbd, %s, %s, %s, %s)", rbdDeploymentName, rbdDaemonsetName, operatorRBDDeploymentName, operatorRBDDaemonsetName)
 )
 
 func deployRBDPlugin() {
@@ -272,6 +276,10 @@ var _ = Describe("RBD", func() {
 			Skip("Skipping RBD E2E")
 		}
 		c = f.ClientSet
+		if operatorDeployment {
+			rbdDeploymentName = operatorRBDDeploymentName
+			rbdDaemonsetName = operatorRBDDaemonsetName
+		}
 		if deployRBD {
 			err := addLabelsToNodes(f, map[string]string{
 				nodeRegionLabel:          regionValue,
@@ -344,11 +352,15 @@ var _ = Describe("RBD", func() {
 		}
 
 		// wait for cluster name update in deployment
+		if operatorDeployment {
+			err = setClusterName(defaultClusterName)
+		} else {
 			containers := []string{"csi-rbdplugin", "csi-rbdplugin-controller"}
 			err = waitForContainersArgsUpdate(c, cephCSINamespace, rbdDeploymentName,
 				"clustername", defaultClusterName, containers, deployTimeout)
+		}
 		if err != nil {
-			framework.Failf("timeout waiting for deployment update %s/%s: %v", cephCSINamespace, rbdDeploymentName, err)
+			framework.Failf("timeout waiting for clustername arg update %s/%s: %v", cephCSINamespace, rbdDeploymentName, err)
 		}
 	})
 
@@ -2814,7 +2826,11 @@ var _ = Describe("RBD", func() {
 				validateRBDImageCount(f, 1, defaultRBDPool)
 				validateOmapCount(f, 1, rbdType, defaultRBDPool, volumesType)
 				// delete rbd nodeplugin pods
-				err = deletePodWithLabel("app=csi-rbdplugin", cephCSINamespace, false)
+				selector, err := getDaemonSetLabelSelector(f, cephCSINamespace, rbdDaemonsetName)
+				if err != nil {
+					framework.Failf("failed to get the labels: %v", err)
+				}
+				err = deletePodWithLabel(selector, cephCSINamespace, false)
 				if err != nil {
 					framework.Failf("fail to delete pod: %v", err)
 				}
@@ -3781,8 +3797,7 @@ var _ = Describe("RBD", func() {
 						framework.Failf("failed to create rados namespace: %v", err)
 					}
 					// delete csi pods
-					err = deletePodWithLabel("app in (ceph-csi-rbd, csi-rbdplugin, csi-rbdplugin-provisioner)",
-						cephCSINamespace, false)
+					err = deletePodWithLabel(rbdPodSelector, cephCSINamespace, false)
 					if err != nil {
 						framework.Failf("failed to delete pods with labels: %v", err)
 					}
@@ -4600,10 +4615,14 @@ var _ = Describe("RBD", func() {
 
 				// wait for cluster name update in deployment
 				containers := []string{"csi-rbdplugin", "csi-rbdplugin-controller"}
+				if operatorDeployment {
+					err = setEnableMetadata(false)
+				} else {
 					err = waitForContainersArgsUpdate(c, cephCSINamespace, rbdDeploymentName,
 						"setmetadata", "false", containers, deployTimeout)
+				}
 				if err != nil {
-					framework.Failf("timeout waiting for deployment update %s/%s: %v", cephCSINamespace, rbdDeploymentName, err)
+					framework.Failf("failed to update setmetadata arg in %s/%s: %v", cephCSINamespace, rbdDeploymentName, err)
 				}
 				pvcSmartClone, err := loadPVC(pvcSmartClonePath)
 				if err != nil {
@@ -4703,11 +4722,15 @@ var _ = Describe("RBD", func() {
 				validateRBDImageCount(f, 0, defaultRBDPool)
 				validateOmapCount(f, 0, rbdType, defaultRBDPool, volumesType)
 				validateOmapCount(f, 0, rbdType, defaultRBDPool, snapsType)
+				if operatorDeployment {
+					err = setEnableMetadata(true)
+				} else {
 					// wait for cluster name update in deployment
 					err = waitForContainersArgsUpdate(c, cephCSINamespace, rbdDeploymentName,
 						"setmetadata", "true", containers, deployTimeout)
+				}
 				if err != nil {
-					framework.Failf("timeout waiting for deployment update %s/%s: %v", cephCSINamespace, rbdDeploymentName, err)
+					framework.Failf("failed to update setmetadata arg in %s/%s: %v", cephCSINamespace, rbdDeploymentName, err)
 				}
 			})
 

e2e/utils.go

@@ -1621,6 +1621,8 @@ const (
 	kubectlCreate = kubectlAction("create")
 	// kubectlDelete tells retryKubectlInput() to run "delete".
 	kubectlDelete = kubectlAction("delete")
+	// kubectlPatch tells retryKubectlInput() to run "patch".
+	kubectlPatch = kubectlAction("patch")
 )
 
 // String returns the string format of the kubectlAction, this is automatically