Support data persistence: integrate with storageClass

api/v1alpha1/etcdcluster_types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -32,6 +33,10 @@ type EtcdClusterSpec struct {
 	Size int `json:"size"`
 	// Version is the expected version of the etcd container image.
 	Version string `json:"version"`
+	// StorageClassName is the name of the StorageClass to use for the etcd cluster.
+	StorageClassName string `json:"storageClassName"`
+	// VolumeSize is the size of the volume to use for the etcd cluster.
+	VolumeSize resource.Quantity `json:"volumeSize"`
 }
 
 // EtcdClusterStatus defines the observed state of EtcdCluster.

config/crd/bases/operator.etcd.io_etcdclusters.yaml

@@ -42,13 +42,27 @@ spec:
               size:
                 description: Size is the expected size of the etcd cluster.
                 type: integer
+              storageClassName:
+                description: StorageClassName is the name of the StorageClass to use
+                  for the etcd cluster.
+                type: string
               version:
                 description: Version is the expected version of the etcd container
                   image.
                 type: string
+              volumeSize:
+                anyOf:
+                - type: integer
+                - type: string
+                description: VolumeSize is the size of the volume to use for the etcd
+                  cluster.
+                pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                x-kubernetes-int-or-string: true
             required:
             - size
+            - storageClassName
             - version
+            - volumeSize
             type: object
           status:
             description: EtcdClusterStatus defines the observed state of EtcdCluster.

internal/controller/utils.go

@@ -26,6 +26,11 @@ import (
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
+const (
+	etcdDataDir = "/var/lib/etcd"
+	volumeName  = "etcd-data"
+)
+
 func prepareOwnerReference(ec *ecv1alpha1.EtcdCluster, scheme *runtime.Scheme) ([]metav1.OwnerReference, error) {
 	gvk, err := apiutil.GVKForObject(ec, scheme)
 	if err != nil {
@@ -87,7 +92,6 @@ func createOrPatchStatefulSet(ctx context.Context, logger logr.Logger, ec *ecv1a
 	if err != nil {
 		return err
 	}
-
 	podSpec := corev1.PodSpec{
 		Containers: []corev1.Container{
 			{
@@ -101,6 +105,10 @@ func createOrPatchStatefulSet(ctx context.Context, logger logr.Logger, ec *ecv1a
 					fmt.Sprintf("--advertise-client-urls=http://$(POD_NAME).%s.$(POD_NAMESPACE).svc.cluster.local:2379", ec.Name),
 				},
 				Image: fmt.Sprintf("gcr.io/etcd-development/etcd:%s", ec.Spec.Version),
+				VolumeMounts: []corev1.VolumeMount{{
+					Name:      volumeName,
+					MountPath: etcdDataDir,
+				}},
 				Env: []corev1.EnvVar{
 					{
 						Name: "POD_NAME",
@@ -142,6 +150,22 @@ func createOrPatchStatefulSet(ctx context.Context, logger logr.Logger, ec *ecv1a
 		},
 	}
 
+	volumeClaimTemplate := corev1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            volumeName,
+			OwnerReferences: owners,
+		},
+		Spec: corev1.PersistentVolumeClaimSpec{
+			AccessModes:      []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOncePod},
+			StorageClassName: &ec.Spec.StorageClassName,
+			Resources: corev1.VolumeResourceRequirements{
+				Requests: corev1.ResourceList{
+					corev1.ResourceStorage: ec.Spec.VolumeSize,
+				},
+			},
+		},
+	}
+
 	logger.Info("Now creating/updating statefulset", "name", ec.Name, "namespace", ec.Namespace, "replicas", replicas)
 	_, err = controllerutil.CreateOrPatch(ctx, c, sts, func() error {
 		// Define or update the desired spec
@@ -156,6 +180,7 @@ func createOrPatchStatefulSet(ctx context.Context, logger logr.Logger, ec *ecv1a
 			Selector: &metav1.LabelSelector{
 				MatchLabels: labels,
 			},
+			VolumeClaimTemplates: []corev1.PersistentVolumeClaim{volumeClaimTemplate},
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: labels,
@@ -292,6 +317,7 @@ func newEtcdClusterState(ec *ecv1alpha1.EtcdCluster, replica int) *corev1.Config
 		Data: map[string]string{
 			"ETCD_INITIAL_CLUSTER_STATE": state,
 			"ETCD_INITIAL_CLUSTER":       strings.Join(initialCluster, ","),
+			"ETCD_DATA_DIR":              etcdDataDir,
 		},
 	}
 }

test/e2e/e2e_test.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -42,6 +43,13 @@ const metricsServiceName = "etcd-operator-controller-manager-metrics-service"
 // metricsRoleBindingName is the name of the RBAC that will be created to allow get the metrics data
 const metricsRoleBindingName = "etcd-operator-metrics-binding"
 
+const (
+	etcdClusterName      = "test-etcd-cluster"
+	etcdClusterNamespace = "default"
+	expectedReplicaCount = 3
+	storageClassName     = "standard"
+)
+
 var _ = Describe("Manager", Ordered, func() {
 	var controllerPodName string
 
@@ -245,6 +253,95 @@ var _ = Describe("Manager", Ordered, func() {
 		//    strings.ToLower(<Kind>),
 		// ))
 	})
+	Context("Storage class check before EtcdCluster creation", func() {
+		It("should have the storage class available", func() {
+			By("checking if the storage class is available")
+			cmd := exec.Command("kubectl", "get", "storageclass", storageClassName)
+			_, err := utils.Run(cmd)
+			Expect(err).NotTo(HaveOccurred(), "Storage class not found")
+		})
+
+	})
+
+	Context("EtcdCluster Setup and Validation", func() {
+		It("should create and verify an EtcdCluster Custom Resource", func() {
+			By("creating an EtcdCluster custom resource")
+
+			// Define the EtcdCluster YAML manifest
+			etcdClusterYAML := `
+apiVersion: operator.etcd.io/v1alpha1
+kind: EtcdCluster
+metadata:
+  name: ` + etcdClusterName + `
+  namespace: ` + etcdClusterNamespace + `
+spec:
+  size: 3
+  version: "v3.5.17"
+  storageClassName: ` + storageClassName + `
+  volumeSize: 100M
+`
+
+			cmd := exec.Command("kubectl", "apply", "-f", "-")
+			cmd.Stdin = strings.NewReader(etcdClusterYAML)
+			_, err := utils.Run(cmd)
+			Expect(err).NotTo(HaveOccurred(), "Failed to apply EtcdCluster resource")
+
+			By("waiting for the EtcdCluster to be reconciled")
+			verifyEtcdClusterReady := func(g Gomega) {
+				cmd := exec.Command("kubectl", "logs", controllerPodName, "-n", namespace)
+				output, err := utils.Run(cmd)
+				g.Expect(err).NotTo(HaveOccurred())
+				g.Expect(output).To(ContainSubstring("EtcdCluster is already up-to-date\t{\"controller\": \"etcdcluster\", \"controllerGroup\": \"operator.etcd.io\", \"controllerKind\": \"EtcdCluster\", \"EtcdCluster\": {\"name\":\"" + etcdClusterName + "\",\"namespace\":\"" + etcdClusterNamespace + "\"}"))
+			}
+			Eventually(verifyEtcdClusterReady, 5*time.Minute).Should(Succeed())
+		})
+
+		It("should create a StatefulSet with the correct number of replicas", func() {
+			validateStatefulSet := func(g Gomega) {
+				cmd := exec.Command("kubectl", "get", "statefulset",
+					etcdClusterName, "-o", "jsonpath={.status.replicas}",
+					"-n", etcdClusterNamespace)
+				output, err := utils.Run(cmd)
+				g.Expect(err).NotTo(HaveOccurred(), "Failed to get StatefulSet details")
+				g.Expect(output).To(Equal(fmt.Sprintf("%d", expectedReplicaCount)), "StatefulSet replicas mismatch")
+			}
+			Eventually(validateStatefulSet, 5*time.Minute, 10*time.Second).Should(Succeed())
+		})
+
+		It("should create PVCs for each StatefulSet replica and bind them", func() {
+			validatePVCs := func(g Gomega) {
+				cmd := exec.Command("kubectl", "get", "pvc",
+					"-l", fmt.Sprintf("app=%s", etcdClusterName),
+					"-o", "jsonpath={.items[*].status.phase}",
+					"-n", etcdClusterNamespace)
+				output, err := utils.Run(cmd)
+				g.Expect(err).NotTo(HaveOccurred(), "Failed to get PVC status")
+				pvcStatuses := utils.GetSeparatedDelimited(output, " ")
+				g.Expect(pvcStatuses).To(HaveLen(expectedReplicaCount), "Incorrect number of PVCs created")
+				for _, status := range pvcStatuses {
+					g.Expect(status).To(Equal("Bound"), "PVC is not in 'Bound' state")
+				}
+			}
+			Eventually(validatePVCs, 5*time.Minute, 10*time.Second).Should(Succeed())
+		})
+
+		It("should ensure all Pods are created and ready", func() {
+			validatePods := func(g Gomega) {
+				cmd := exec.Command("kubectl", "get", "pods",
+					"-l", fmt.Sprintf("app=%s", etcdClusterName),
+					"-o", "jsonpath={.items[*].status.conditions[?(@.type==\"Ready\")].status}",
+					"-n", etcdClusterNamespace)
+				output, err := utils.Run(cmd)
+				g.Expect(err).NotTo(HaveOccurred(), "Failed to get Pod readiness status")
+				podStatuses := utils.GetSeparatedDelimited(output, " ")
+				g.Expect(podStatuses).To(HaveLen(expectedReplicaCount), "Incorrect number of Pods created")
+				for _, status := range podStatuses {
+					g.Expect(status).To(Equal("True"), "Pod is not in 'Ready' state")
+				}
+			}
+			Eventually(validatePods, 5*time.Minute, 10*time.Second).Should(Succeed())
+		})
+	})
 })
 
 // serviceAccountToken returns a token for the specified service account in the given namespace.

test/utils/utils.go

@@ -191,6 +191,20 @@ func GetNonEmptyLines(output string) []string {
 	return res
 }
 
+// GetSeparatedDelimited converts given command output string into individual objects
+// according to delimiter, and ignores the empty elements in it.
+func GetSeparatedDelimited(output string, delimiter string) []string {
+	var res []string
+	elements := strings.Split(output, delimiter)
+	for _, element := range elements {
+		if element != "" {
+			res = append(res, element)
+		}
+	}
+
+	return res
+}
+
 // GetProjectDir will return the directory where the project is
 func GetProjectDir() (string, error) {
 	wd, err := os.Getwd()