e2e: ceph-csi-operator deployment support

build.env

@@ -14,6 +14,9 @@ CSI_IMAGE_VERSION=canary
 # cephcsi upgrade version
 CSI_UPGRADE_VERSION=v3.12.1
 
+# ceph-csi-operator version used for e2e test
+CEPH_CSI_OPERATOR_VERSION=v0.2.0
+
 # Ceph version to use
 BASE_IMAGE=quay.io/ceph/ceph:v19
 CEPH_VERSION=squid

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,13 +178,19 @@ var _ = Describe(cephfsType, func() {
 			Skip("Skipping CephFS E2E")
 		}
 		c = f.ClientSet
-		if deployCephFS {
-			if cephCSINamespace != defaultNs {
-				err := createNamespace(c, cephCSINamespace)
-				if err != nil {
-					framework.Failf("failed to create namespace %s: %v", cephCSINamespace, err)
-				}
+		if operatorDeployment {
+			cephFSDeploymentName = operatorCephFSDeploymentName
+			cephFSDeamonSetName = operatorCephFSDaemonsetName
+		}
+
+		if cephCSINamespace != defaultNs && !operatorDeployment {
+			err := createNamespace(c, cephCSINamespace)
+			if err != nil {
+				framework.Failf("failed to create namespace %s: %v", cephCSINamespace, err)
 			}
+		}
+
+		if deployCephFS {
 			deployCephfsPlugin()
 		}
 		err := createConfigMap(cephFSDirPath, f.ClientSet, f)
@@ -209,11 +218,15 @@ var _ = Describe(cephfsType, func() {
 		deployVault(f.ClientSet, deployTimeout)
 
 		// wait for cluster name update in deployment
-		containers := []string{cephFSContainerName}
-		err = waitForContainersArgsUpdate(c, cephCSINamespace, cephFSDeploymentName,
-			"clustername", defaultClusterName, containers, deployTimeout)
+		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)
@@ -226,13 +239,14 @@ var _ = Describe(cephfsType, func() {
 		if !testCephFS || upgradeTesting {
 			Skip("Skipping CephFS E2E")
 		}
+
 		if CurrentSpecReport().Failed() {
 			// log pods created by helm chart
 			logsCSIPods("app=ceph-csi-cephfs", c)
 			// log provisioner
-			logsCSIPods("app=csi-cephfsplugin-provisioner", c)
+			logsCSIPods("app="+cephFSDeploymentName, c)
 			// log node plugin
-			logsCSIPods("app=csi-cephfsplugin", c)
+			logsCSIPods("app="+cephFSDeamonSetName, c)
 
 			// log all details from the namespace where Ceph-CSI is deployed
 			e2edebug.DumpAllNamespaceInfo(context.TODO(), c, cephCSINamespace)
@@ -266,11 +280,11 @@ var _ = Describe(cephfsType, func() {
 
 		if deployCephFS {
 			deleteCephfsPlugin()
-			if cephCSINamespace != defaultNs {
-				err = deleteNamespace(c, cephCSINamespace)
-				if err != nil {
-					framework.Failf("failed to delete namespace %s: %v", cephCSINamespace, err)
-				}
+		}
+		if cephCSINamespace != defaultNs && !operatorDeployment {
+			err = deleteNamespace(c, cephCSINamespace)
+			if err != nil {
+				framework.Failf("failed to delete namespace %s: %v", cephCSINamespace, err)
 			}
 		}
 	})

e2e/deployment.go

@@ -232,16 +232,12 @@ 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
-	domainLabel         string
-	crushLocationLabels string
+	filename    string
+	namespace   string
+	domainLabel string
 
 	// set the number of replicas in a Deployment to 1.
 	oneReplica bool
-
-	// enable read affinity support (for RBD)
-	enableReadAffinity bool
 }
 
 func (yrn *yamlResourceNamespaced) Do(action kubectlAction) error {
@@ -263,14 +259,6 @@ 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/e2e_test.go

@@ -52,6 +52,7 @@ func init() {
 	flag.StringVar(&fileSystemName, "filesystem", "myfs", "CephFS filesystem to use")
 	flag.StringVar(&clusterID, "clusterid", "", "Ceph cluster ID to use (defaults to `ceph fsid` detection)")
 	flag.StringVar(&nfsDriverName, "nfs-driver", "nfs.csi.ceph.com", "name of the driver for NFS-volumes")
+	flag.BoolVar(&operatorDeployment, "operator-deployment", false, "test running on deployment via operator")
 	setDefaultKubeconfig()
 
 	// Register framework flags, then handle flags
@@ -91,4 +92,8 @@ func handleFlags() {
 		testNFS = testCephFS
 		deployNFS = deployCephFS
 	}
+
+	if operatorDeployment {
+		cephCSINamespace = "ceph-csi-operator-system"
+	}
 }

e2e/nfs.go

@@ -50,9 +50,12 @@ 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) {
+func deployNFSPlugin() {
 	// delete objects deployed by rook
 
 	err := deleteResource(nfsDirPath + nfsProvisionerRBAC)
@@ -65,18 +68,35 @@ func deployNFSPlugin(f *framework.Framework) {
 		framework.Failf("failed to delete nodeplugin rbac %s: %v", nfsDirPath+nfsNodePluginRBAC, err)
 	}
 
+	createORDeleteNFSResources(kubectlCreate)
+}
+
+func deleteNFSPlugin() {
+	createORDeleteNFSResources(kubectlDelete)
+}
+
+func createNFSPool(f *framework.Framework) {
 	// the pool should not be deleted, as it may contain configurations
 	// from non-e2e related CephNFS objects
-	err = createPool(f, nfsPoolName)
+	err := createPool(f, nfsPoolName)
 	if err != nil {
 		framework.Failf("failed to create pool for NFS config %q: %v", nfsPoolName, err)
 	}
 
-	createORDeleteNFSResources(kubectlCreate)
-}
+	resources := []ResourceDeployer{
+		// NFS server deployment
+		&yamlResourceNamespaced{
+			filename:  nfsExamplePath + nfsRookCephNFS,
+			namespace: rookNamespace,
+		},
+	}
 
-func deleteNFSPlugin() {
-	createORDeleteNFSResources(kubectlDelete)
+	for _, r := range resources {
+		err := r.Do(kubectlCreate)
+		if err != nil {
+			framework.Failf("failed to %s resource: %v", kubectlCreate, err)
+		}
+	}
 }
 
 func createORDeleteNFSResources(action kubectlAction) {
@@ -242,14 +262,21 @@ var _ = Describe("nfs", func() {
 			Skip("Skipping NFS E2E")
 		}
 		c = f.ClientSet
-		if deployNFS {
-			if cephCSINamespace != defaultNs {
-				err := createNamespace(c, cephCSINamespace)
-				if err != nil {
-					framework.Failf("failed to create namespace %s: %v", cephCSINamespace, err)
-				}
+		if operatorDeployment {
+			nfsDeploymentName = operatorNFSDeploymentName
+			nfsDeamonSetName = operatorNFSDaemonsetName
+		}
+
+		if cephCSINamespace != defaultNs && !operatorDeployment {
+			err := createNamespace(c, cephCSINamespace)
+			if err != nil {
+				framework.Failf("failed to create namespace %s: %v", cephCSINamespace, err)
 			}
-			deployNFSPlugin(f)
+		}
+
+		createNFSPool(f)
+		if deployNFS {
+			deployNFSPlugin()
 		}
 
 		// cephfs testing might have changed the default subvolumegroup
@@ -287,13 +314,14 @@ var _ = Describe("nfs", func() {
 		if !testNFS || upgradeTesting {
 			Skip("Skipping NFS E2E")
 		}
+
 		if CurrentSpecReport().Failed() {
 			// log pods created by helm chart
 			logsCSIPods("app=ceph-csi-nfs", c)
 			// log provisioner
-			logsCSIPods("app=csi-nfsplugin-provisioner", c)
+			logsCSIPods("app="+nfsDeploymentName, c)
 			// log node plugin
-			logsCSIPods("app=csi-nfsplugin", c)
+			logsCSIPods("app="+nfsDeamonSetName, c)
 
 			// log all details from the namespace where Ceph-CSI is deployed
 			e2edebug.DumpAllNamespaceInfo(context.TODO(), c, cephCSINamespace)
@@ -325,11 +353,12 @@ var _ = Describe("nfs", func() {
 
 		if deployNFS {
 			deleteNFSPlugin()
-			if cephCSINamespace != defaultNs {
-				err = deleteNamespace(c, cephCSINamespace)
-				if err != nil {
-					framework.Failf("failed to delete namespace %s: %v", cephCSINamespace, err)
-				}
+		}
+
+		if cephCSINamespace != defaultNs && !operatorDeployment {
+			err = deleteNamespace(c, cephCSINamespace)
+			if err != nil {
+				framework.Failf("failed to delete namespace %s: %v", cephCSINamespace, err)
 			}
 		}
 	})

e2e/operator.go

@@ -0,0 +1,94 @@
+/*
+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 (
+	"encoding/json"
+	"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": %q}}}`, value),
+	}
+
+	// Patch the operator config
+	err := retryKubectlArgs(OperatorNamespace, kubectlPatch, deployTimeout, command...)
+	if err != nil {
+		return fmt.Errorf("failed to set cluster name: %w", err)
+	}
+
+	return nil
+}
+
+func setDomainLabels(labels []string) error {
+	// Define the patch operations
+	patchOps := []map[string]interface{}{
+		{"op": "add", "path": "/spec/driverSpecDefaults/nodePlugin", "value": map[string]interface{}{}},
+		{"op": "add", "path": "/spec/driverSpecDefaults/nodePlugin/topology", "value": map[string]interface{}{}},
+		{"op": "add", "path": "/spec/driverSpecDefaults/nodePlugin/topology/domainLabels", "value": labels},
+	}
+
+	// Serialize to JSON
+	patchJSON, err := json.Marshal(patchOps)
+	if err != nil {
+		return fmt.Errorf("failed to marshal patch JSON: %w", err)
+	}
+
+	command := []string{
+		"operatorconfigs.csi.ceph.io",
+		OperatorConfigName,
+		"--type=json",
+		"-p",
+		string(patchJSON),
+	}
+
+	// Patch the operator config
+	err = retryKubectlArgs(OperatorNamespace, kubectlPatch, deployTimeout, command...)
+	if err != nil {
+		return fmt.Errorf("failed to set domain labels: %w", err)
+	}
+
+	return nil
+}