Add e2e tests with updates and rw checks

GrigoryPervakov · GrigoryPervakov · commit 9bfe69aae05c · 2025-06-03T20:53:11.000+02:00
diff --git a/Tiltfile b/Tiltfile
@@ -11,7 +11,7 @@ local_resource(
     labels=["makefile"],
 )
 
-docker_build('controller', '.',
+docker_build('clickhouse.com/clickhouse-operator', '.',
     dockerfile='Dockerfile',
     ignore=[
         'config',
diff --git a/api/v1alpha1/keepercluster_conditions.go b/api/v1alpha1/keepercluster_conditions.go
@@ -17,6 +17,17 @@ const (
 	KeeperConditionTypeReady KeeperConditionType = "Ready"
 )
 
+var (
+	AllConditionTypes = []KeeperConditionType{
+		KeeperConditionTypeReconcileSucceeded,
+		KeeperConditionTypeReplicaStartupSucceeded,
+		KeeperConditionTypeHealthy,
+		KeeperConditionTypeClusterSizeAligned,
+		KeeperConditionTypeConfigurationInSync,
+		KeeperConditionTypeReady,
+	}
+)
+
 type KeeperConditionReason string
 
 const (
diff --git a/api/v1alpha1/keepercluster_types.go b/api/v1alpha1/keepercluster_types.go
@@ -142,42 +142,6 @@ func (s *KeeperClusterSpec) WithDefaults() {
 	}
 }
 
-// KeeperReplicaState
-// +kubebuilder:validation:Enum=Pending;Provisioning;Ready;Degraded;OutOfSync;Error
-type KeeperReplicaState string
-
-const (
-	// ReplicaStatePending - Waiting for workload creation to be initiated.
-	ReplicaStatePending KeeperReplicaState = "Pending"
-	// ReplicaStateProvisioning - Starting StatefulSet, replica may not be able to accept connections.
-	ReplicaStateProvisioning KeeperReplicaState = "Provisioning"
-	// ReplicaStateDegraded - Pod is running, but replica is not healthy.
-	ReplicaStateDegraded KeeperReplicaState = "Degraded"
-	// ReplicaStateOutOfSync - Some changes were not applied yet.
-	ReplicaStateOutOfSync KeeperReplicaState = "OutOfSync"
-	// ReplicaStateReady - Pods are running and replica should be able to take connections.
-	ReplicaStateReady KeeperReplicaState = "Ready"
-	// ReplicaStateDeprovisioning - Replica is being deleted from cluster.
-	ReplicaStateDeprovisioning KeeperReplicaState = "Deprovisioning"
-	// ReplicaStateError - A persistent error is causing pods to fail.
-	ReplicaStateError KeeperReplicaState = "Error"
-)
-
-var activeReplicaStates = map[KeeperReplicaState]bool{
-	ReplicaStatePending:        false,
-	ReplicaStateDeprovisioning: false,
-	ReplicaStateProvisioning:   true,
-	ReplicaStateDegraded:       true,
-	ReplicaStateOutOfSync:      true,
-	ReplicaStateReady:          true,
-	ReplicaStateError:          true,
-}
-
-// Active replica state means that pod is expected to exist and other replicas aware of this replica.
-func (s KeeperReplicaState) Active() bool {
-	return activeReplicaStates[s]
-}
-
 // KeeperClusterStatus defines the observed state of KeeperCluster.
 type KeeperClusterStatus struct {
 	// +listType=map
diff --git a/go.mod b/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 	github.com/go-logr/logr v1.4.2
 	github.com/go-logr/zapr v1.3.0
+	github.com/go-zookeeper/zk v1.0.4
 	github.com/google/go-cmp v0.6.0
 	github.com/onsi/ginkgo/v2 v2.19.0
 	github.com/onsi/gomega v1.33.1
diff --git a/go.sum b/go.sum
@@ -44,6 +44,8 @@ github.com/go-openapi/swag v0.22.4 h1:QLMzNJnMGPRNDCbySlcj1x01tzU8/9LTTL9hZZZogB
 github.com/go-openapi/swag v0.22.4/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14=
 github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
 github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
+github.com/go-zookeeper/zk v1.0.4 h1:DPzxraQx7OrPyXq2phlGlNSIyWEsAox0RJmjTseMV6I=
+github.com/go-zookeeper/zk v1.0.4/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE=
diff --git a/internal/controller/keeper/constants.go b/internal/controller/keeper/constants.go
@@ -20,14 +20,15 @@ const (
 	QuorumConfigFileName   = "config.yaml"
 	QuorumConfigVolumeName = "clickhouse-keeper-quorum-config-volume"
 
+	PersistentVolumeName = "keeper-storage-volume"
+
 	ConfigPath       = QuorumConfigPath + "config.d/"
 	ConfigFileName   = "00-config.yaml"
 	ConfigVolumeName = "clickhouse-keeper-config-volume"
 
-	BasePath            = "/var/lib/clickhouse-keeper/"
-	StoragePath         = BasePath + "storage/"
-	StorageLogPath      = BasePath + "coordination/log/"
-	StorageSnapshotPath = BasePath + "coordination/snapshots/"
+	BaseDataPath        = "/var/lib/clickhouse/"
+	StorageLogPath      = BaseDataPath + "coordination/log/"
+	StorageSnapshotPath = BaseDataPath + "coordination/snapshots/"
 
 	ContainerName          = "clickhouse-keeper"
 	DefaultRevisionHistory = 10
diff --git a/internal/controller/keeper/sync.go b/internal/controller/keeper/sync.go
@@ -80,7 +80,7 @@ func (r replicaState) Ready(ctx reconcileContext) bool {
 	}
 
 	stsReady := r.StatefulSet.Status.ReadyReplicas == 1 // Not reliable, but allows to wait until pod is `green`
-	if ctx.KeeperCluster.Replicas() == 1 {
+	if len(ctx.KeeperCluster.Status.Replicas) == 1 {
 		return stsReady && r.Status.ServerState == ModeStandalone
 	}
 
@@ -704,6 +704,13 @@ func (r *ClusterReconciler) reconcileConditions(ctx reconcileContext) (*ctrl.Res
 	}
 
 	setCondition(ctx, v1.KeeperConditionTypeReady, status, reason, message)
+
+	for _, condition := range ctx.KeeperCluster.Status.Conditions {
+		if condition.Status != metav1.ConditionTrue {
+			return &ctrl.Result{RequeueAfter: RequeueOnRefreshTimeout}, nil
+		}
+	}
+
 	return &ctrl.Result{}, nil
 }
 
diff --git a/internal/controller/keeper/templates.go b/internal/controller/keeper/templates.go
@@ -353,7 +353,7 @@ func TemplateStatefulSet(cr *v1.KeeperCluster, replicaID string) *appsv1.Statefu
 		VolumeClaimTemplates: []corev1.PersistentVolumeClaim{
 			{
 				ObjectMeta: metav1.ObjectMeta{
-					Name: "ch-storage-volume",
+					Name: PersistentVolumeName,
 				},
 				Spec: cr.Spec.Storage,
 			},
@@ -386,7 +386,7 @@ func TemplateStatefulSet(cr *v1.KeeperCluster, replicaID string) *appsv1.Statefu
 func generateConfigForSingleReplica(cr *v1.KeeperCluster, replicaID string) Config {
 	config := Config{
 		ListenHost: "0.0.0.0",
-		Path:       BasePath,
+		Path:       BaseDataPath,
 		Prometheus: PrometheusConfig{
 			Endpoint:            "/metrics",
 			Port:                PortPrometheusScrape,
@@ -401,7 +401,7 @@ func generateConfigForSingleReplica(cr *v1.KeeperCluster, replicaID string) Conf
 		KeeperServer: KeeperServer{
 			TcpPort:             PortNative,
 			ServerID:            replicaID,
-			StoragePath:         StoragePath,
+			StoragePath:         BaseDataPath,
 			DigestEnabled:       true,
 			LogStoragePath:      StorageLogPath,
 			SnapshotStoragePath: StorageSnapshotPath,
@@ -441,12 +441,12 @@ func buildVolumes(cr *v1.KeeperCluster, replicaID string) ([]corev1.Volume, []co
 			MountPath: ConfigPath,
 		},
 		{
-			Name:      "ch-storage-volume",
-			MountPath: StoragePath,
+			Name:      PersistentVolumeName,
+			MountPath: BaseDataPath,
 			SubPath:   "var-lib-clickhouse",
 		},
 		{
-			Name:      "ch-storage-volume",
+			Name:      PersistentVolumeName,
 			MountPath: "/var/log/clickhouse-keeper",
 			SubPath:   "var-log-clickhouse",
 		},
diff --git a/internal/util/common.go b/internal/util/common.go
@@ -56,6 +56,9 @@ func ApplyDefault[T any](source *T, defaults T) error {
 func applyDefaultRecursive(sourceValue reflect.Value, defaults reflect.Value) error {
 	if sourceValue.Kind() == reflect.Struct {
 		for i := range sourceValue.NumField() {
+			if !sourceValue.Field(i).CanSet() {
+				continue
+			}
 			if err := applyDefaultRecursive(sourceValue.Field(i), defaults.Field(i)); err != nil {
 				return fmt.Errorf("apply default value for field %s: %w", sourceValue.Type().Field(i).Name, err)
 			}
diff --git a/test/e2e/e2e_suite_test.go b/test/e2e/e2e_suite_test.go
@@ -73,8 +73,8 @@ var _ = BeforeSuite(func() {
 	By("installing the cert-manager")
 	Expect(utils.InstallCertManager()).To(Succeed())
 
-	By("installing prometheus operator")
-	Expect(utils.InstallPrometheusOperator()).To(Succeed())
+	// By("installing prometheus operator")
+	// Expect(utils.InstallPrometheusOperator()).To(Succeed())
 
 	By("creating manager namespace")
 	cmd := exec.Command("kubectl", "create", "ns", namespace)
@@ -98,7 +98,7 @@ var _ = BeforeSuite(func() {
 	err = utils.LoadImageToKindClusterWithName(projectimage)
 	Expect(err).NotTo(HaveOccurred())
 
-	// In minukube cert-manager root CA may not be injected at this moment.
+	// In kind cert-manager root CA may not be injected at this moment.
 	By("deploying the controller-manager")
 	Eventually(func() error {
 		cmd = exec.Command("make", "deploy", fmt.Sprintf("IMG=%s", projectimage))
@@ -153,11 +153,8 @@ var _ = BeforeSuite(func() {
 var _ = AfterSuite(func() {
 	cancel()
 
-	By("uninstalling the Prometheus manager bundle")
-	utils.UninstallPrometheusOperator()
-
-	By("uninstalling the cert-manager bundle")
-	utils.UninstallCertManager()
+	// By("uninstalling the Prometheus manager bundle")
+	// utils.UninstallPrometheusOperator()
 
 	By("removing manager namespace")
 	cmd := exec.Command("kubectl", "delete", "ns", namespace)
diff --git a/test/e2e/keeper_e2e_test.go b/test/e2e/keeper_e2e_test.go
@@ -20,40 +20,122 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/clickhouse-operator/api/v1alpha1"
+	v1 "github.com/clickhouse-operator/api/v1alpha1"
+	"github.com/clickhouse-operator/internal/util"
+	"github.com/clickhouse-operator/test/utils"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"golang.org/x/exp/rand"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 )
 
+const (
+	KeeperBaseVersion   = "25.3"
+	KeeperUpdateVersion = "25.5"
+)
+
 var _ = Describe("Keeper controller", func() {
-	When("operate standalone cluster", Ordered, func() {
-		cr := v1alpha1.KeeperCluster{
+	DescribeTable("standalone keeper updates", func(specUpdate v1.KeeperClusterSpec) {
+		cr := v1.KeeperCluster{
 			ObjectMeta: metav1.ObjectMeta{
 				Namespace: testNamespace,
-				Name:      fmt.Sprintf("standalone-%d", rand.Uint32()),
+				Name:      fmt.Sprintf("keeper-%d", rand.Uint32()),
 			},
-			Spec: v1alpha1.KeeperClusterSpec{
+			Spec: v1.KeeperClusterSpec{
 				Replicas: ptr.To[int32](1),
+				Image: v1.ContainerImage{
+					Tag: KeeperBaseVersion,
+				},
+			},
+		}
+		checks := 0
+
+		By("creating cluster CR")
+		Expect(k8sClient.Create(ctx, &cr)).To(Succeed())
+		WaitUpdatedAndReady(&cr, time.Minute)
+		RWChecks(&cr, &checks)
+
+		By("updating cluster CR")
+		Expect(k8sClient.Get(ctx, cr.GetNamespacedName(), &cr)).To(Succeed())
+		Expect(util.ApplyDefault(&specUpdate, cr.Spec)).To(Succeed())
+		cr.Spec = specUpdate
+		Expect(k8sClient.Update(ctx, &cr)).To(Succeed())
+
+		WaitUpdatedAndReady(&cr, 3*time.Minute)
+		RWChecks(&cr, &checks)
+
+		Expect(k8sClient.Delete(ctx, &cr)).To(Succeed())
+	},
+		Entry("update log level", v1.KeeperClusterSpec{LoggerConfig: v1.LoggerConfig{LoggerLevel: "warning"}}),
+		Entry("upgrade version", v1.KeeperClusterSpec{Image: v1.ContainerImage{Tag: KeeperUpdateVersion}}),
+		Entry("scale up to 3 replicas", v1.KeeperClusterSpec{Replicas: ptr.To[int32](3)}),
+	)
+
+	DescribeTable("keeper cluster updates", func(baseReplicas int, specUpdate v1.KeeperClusterSpec) {
+		cr := v1.KeeperCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: testNamespace,
+				Name:      fmt.Sprintf("keeper-%d", rand.Uint32()),
+			},
+			Spec: v1.KeeperClusterSpec{
+				Replicas: ptr.To(int32(baseReplicas)),
+				Image: v1.ContainerImage{
+					Tag: KeeperBaseVersion,
+				},
 			},
 		}
+		checks := 0
+
+		By("creating cluster CR")
+		Expect(k8sClient.Create(ctx, &cr)).To(Succeed())
+		WaitUpdatedAndReady(&cr, time.Minute)
+		RWChecks(&cr, &checks)
+
+		// TODO ensure updates one-by-one
+		By("updating cluster CR")
+		Expect(k8sClient.Get(ctx, cr.GetNamespacedName(), &cr)).To(Succeed())
+		Expect(util.ApplyDefault(&specUpdate, cr.Spec)).To(Succeed())
+		cr.Spec = specUpdate
+		Expect(k8sClient.Update(ctx, &cr)).To(Succeed())
+
+		WaitUpdatedAndReady(&cr, 3*time.Minute)
+		RWChecks(&cr, &checks)
 
-		AfterAll(func() {
-			Expect(k8sClient.Delete(ctx, &cr)).To(Succeed())
-		})
-
-		It("should successfully create standalone cluster", func() {
-			By("creating cluster CR")
-			Expect(k8sClient.Create(ctx, &cr)).To(Succeed())
-
-			By("waiting for cluster to be ready")
-			Eventually(func() bool {
-				var cluster v1alpha1.KeeperCluster
-				Expect(k8sClient.Get(ctx, cr.GetNamespacedName(), &cluster)).To(Succeed())
-				return cluster.Status.ReadyReplicas == 1
-			}, time.Minute).Should(BeTrue())
-		})
-	})
+		Expect(k8sClient.Delete(ctx, &cr)).To(Succeed())
+	},
+		Entry("update log level", 3, v1.KeeperClusterSpec{LoggerConfig: v1.LoggerConfig{LoggerLevel: "warning"}}),
+		Entry("upgrade version", 3, v1.KeeperClusterSpec{Image: v1.ContainerImage{Tag: KeeperUpdateVersion}}),
+		Entry("scale up to 5 replicas", 3, v1.KeeperClusterSpec{Replicas: ptr.To[int32](5)}),
+		Entry("scale down to 3 replicas", 5, v1.KeeperClusterSpec{Replicas: ptr.To[int32](3)}),
+	)
 })
+
+func WaitUpdatedAndReady(cr *v1.KeeperCluster, timeout time.Duration) {
+	By("waiting for cluster to be ready")
+	Eventually(func() bool {
+		var cluster v1.KeeperCluster
+		Expect(k8sClient.Get(ctx, cr.GetNamespacedName(), &cluster)).To(Succeed())
+		return cluster.Generation == cluster.Status.ObservedGeneration &&
+			cluster.Status.CurrentRevision == cluster.Status.UpdateRevision &&
+			cluster.Status.ReadyReplicas == cluster.Replicas()
+	}, timeout).Should(BeTrue())
+}
+
+func RWChecks(cr *v1.KeeperCluster, checksDone *int) {
+	Expect(k8sClient.Get(ctx, cr.GetNamespacedName(), cr)).To(Succeed())
+
+	By("connecting to cluster")
+	client, err := utils.NewKeeperClient(ctx, cr)
+	Expect(err).NotTo(HaveOccurred())
+	defer client.Close()
+
+	By("writing new test data")
+	Expect(client.CheckWrite(*checksDone)).To(Succeed())
+	*checksDone++
+
+	By("reading all test data")
+	for i := range *checksDone {
+		Expect(client.CheckRead(i)).To(Succeed(), "check read %d failed", i)
+	}
+}
diff --git a/test/utils/keeper_client.go b/test/utils/keeper_client.go
diff --git a/test/utils/utils.go b/test/utils/utils.go

Original file line number	Diff line number	Diff line change
`@@ -11,7 +11,7 @@ local_resource(`
`11`	`11`	`labels=["makefile"],`
`12`	`12`	`)`
`13`	`13`
`14`		`-docker_build('controller', '.',`
	`14`	`+docker_build('clickhouse.com/clickhouse-operator', '.',`
`15`	`15`	`dockerfile='Dockerfile',`
`16`	`16`	`ignore=[`
`17`	`17`	`'config',`
Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,7 @@ func (r replicaState) Ready(ctx reconcileContext) bool {`
`80`	`80`	`}`
`81`	`81`
`82`	`82`	stsReady := r.StatefulSet.Status.ReadyReplicas == 1 // Not reliable, but allows to wait until pod is `green`
`83`		`- if ctx.KeeperCluster.Replicas() == 1 {`
	`83`	`+ if len(ctx.KeeperCluster.Status.Replicas) == 1 {`
`84`	`84`	`return stsReady && r.Status.ServerState == ModeStandalone`
`85`	`85`	`}`
`86`	`86`
`@@ -704,6 +704,13 @@ func (r ClusterReconciler) reconcileConditions(ctx reconcileContext) (ctrl.Res`
`704`	`704`	`}`
`705`	`705`
`706`	`706`	`setCondition(ctx, v1.KeeperConditionTypeReady, status, reason, message)`
	`707`	`+`
	`708`	`+ for _, condition := range ctx.KeeperCluster.Status.Conditions {`
	`709`	`+ if condition.Status != metav1.ConditionTrue {`
	`710`	`+ return &ctrl.Result{RequeueAfter: RequeueOnRefreshTimeout}, nil`
	`711`	`+ }`
	`712`	`+ }`
	`713`	`+`
`707`	`714`	`return &ctrl.Result{}, nil`
`708`	`715`	`}`
`709`	`716`