chores: split reconcilerBase on more specialized helpers (#162)

Author: GrigoryPervakov

api/v1alpha1/clickhousecluster_types.go

@@ -220,8 +220,6 @@ type ClickHouseCluster struct {
 
 	Spec   ClickHouseClusterSpec   `json:"spec,omitempty"`
 	Status ClickHouseClusterStatus `json:"status,omitempty"`
-
-	specificName string `json:"-"`
 }
 
 // ClickHouseReplicaID identifies a ClickHouse replica within the cluster.
@@ -291,22 +289,18 @@ func (v *ClickHouseCluster) GetStatus() *ClickHouseClusterStatus {
 	return &v.Status
 }
 
-// Conditions returns pointer to the conditions slice.
-func (v *ClickHouseCluster) Conditions() *[]metav1.Condition {
-	if v.Status.Conditions == nil {
-		v.Status.Conditions = []metav1.Condition{}
+// GetConditions returns pointer to the conditions slice.
+func (v *ClickHouseClusterStatus) GetConditions() *[]metav1.Condition {
+	if v.Conditions == nil {
+		v.Conditions = []metav1.Condition{}
 	}
 
-	return &v.Status.Conditions
+	return &v.Conditions
 }
 
 // SpecificName returns cluster name with resource suffix. Used to generate resource names that may be used in DNS.
 func (v *ClickHouseCluster) SpecificName() string {
-	if v.specificName == "" {
-		v.specificName = normalizeName(v.Name) + "-clickhouse"
-	}
-
-	return v.specificName
+	return normalizeName(v.Name) + "-clickhouse"
 }
 
 // Shards returns requested number of shards in the ClickHouseCluster.

api/v1alpha1/conditions.go

@@ -1,10 +1,10 @@
 package v1alpha1
 
 // ConditionType represents the type of condition.
-type ConditionType string
+type ConditionType = string
 
 // ConditionReason represents the reason for a condition's Status.
-type ConditionReason string
+type ConditionReason = string
 
 // Common condition types and reasons for all resources.
 const (

api/v1alpha1/keepercluster_types.go

@@ -184,8 +184,6 @@ type KeeperCluster struct {
 
 	Spec   KeeperClusterSpec   `json:"spec,omitempty"`
 	Status KeeperClusterStatus `json:"status,omitempty"`
-
-	specificName string `json:"-"`
 }
 
 // KeeperReplicaID represents ClickHouse Keeper replica ID. Used for naming resources and RAFT configuration.
@@ -226,22 +224,18 @@ func (v *KeeperCluster) GetStatus() *KeeperClusterStatus {
 	return &v.Status
 }
 
-// Conditions returns pointer to the conditions slice.
-func (v *KeeperCluster) Conditions() *[]metav1.Condition {
-	if v.Status.Conditions == nil {
-		v.Status.Conditions = []metav1.Condition{}
+// GetConditions returns pointer to the conditions slice.
+func (v *KeeperClusterStatus) GetConditions() *[]metav1.Condition {
+	if v.Conditions == nil {
+		v.Conditions = []metav1.Condition{}
 	}
 
-	return &v.Status.Conditions
+	return &v.Conditions
 }
 
 // SpecificName returns cluster name with resource suffix. Used to generate resource names.
 func (v *KeeperCluster) SpecificName() string {
-	if v.specificName == "" {
-		v.specificName = normalizeName(v.Name) + "-keeper"
-	}
-
-	return v.specificName
+	return normalizeName(v.Name) + "-keeper"
 }
 
 // Replicas returns requested number of replicas in the cluster.

internal/controller/clickhouse/commands.go

@@ -106,6 +106,52 @@ func (cmd *commander) Version(ctx context.Context, id v1.ClickHouseReplicaID) (s
 	return version, nil
 }
 
+func (cmd *commander) SyncDatabases(ctx context.Context, log controllerutil.Logger, replicas []v1.ClickHouseReplicaID) bool {
+	result := true
+
+	replicaDatabases := controllerutil.ExecuteParallel(replicas, func(id v1.ClickHouseReplicaID) (v1.ClickHouseReplicaID, map[string]databaseDescriptor, error) {
+		databases, err := cmd.Databases(ctx, id)
+		return id, databases, err
+	})
+
+	databases := map[string]databaseDescriptor{}
+	for id, dbs := range replicaDatabases {
+		if dbs.Err != nil {
+			log.Warn("failed to get databases from replica", "replica_id", id, "error", dbs.Err)
+
+			result = false
+			continue
+		}
+
+		databases = controllerutil.MergeMaps(databases, dbs.Result)
+	}
+
+	results := controllerutil.ExecuteParallel(replicas, func(id v1.ClickHouseReplicaID) (v1.ClickHouseReplicaID, struct{}, error) {
+		if len(databases) == len(replicaDatabases[id].Result) {
+			return id, struct{}{}, nil
+		}
+
+		dbsToSync := map[string]databaseDescriptor{}
+		for name, desc := range databases {
+			if _, ok := replicaDatabases[id].Result[name]; !ok {
+				dbsToSync[name] = desc
+			}
+		}
+
+		return id, struct{}{}, cmd.CreateDatabases(ctx, log, id, dbsToSync)
+	})
+
+	for id, res := range results {
+		if res.Err != nil {
+			log.Info("failed to create databases", "replica_id", id, "error", res.Err)
+
+			result = false
+		}
+	}
+
+	return result
+}
+
 func (cmd *commander) Databases(ctx context.Context, id v1.ClickHouseReplicaID) (map[string]databaseDescriptor, error) {
 	conn, err := cmd.getConn(id)
 	if err != nil {
@@ -169,55 +215,23 @@ func (cmd *commander) CreateDatabases(ctx context.Context, log controllerutil.Lo
 	return nil
 }
 
-// EnsureDefaultDatabaseEngine ensures that the default database engine is set to the Selected one.
-func (cmd *commander) EnsureDefaultDatabaseEngine(ctx context.Context, log controllerutil.Logger, id v1.ClickHouseReplicaID) error {
-	log = log.With("replica_id", id)
-
-	conn, err := cmd.getConn(id)
-	if err != nil {
-		return fmt.Errorf("failed to get connection for replica %s: %w", id, err)
-	}
-
-	var engine string
-
-	rows := conn.QueryRow(ctx, "SELECT engine FROM system.databases WHERE name='default' ")
-	if err = rows.Scan(&engine); err != nil {
-		if !errors.Is(err, sql.ErrNoRows) {
-			return fmt.Errorf("failed to scan default database engine for replica %s: %w", id, err)
-		}
-
-		log.Debug("no default database found")
-	} else {
-		if engine == "Replicated" {
-			log.Debug("default database already has the Replicated engine")
-			return nil
-		}
-
-		var count uint64
-		if err = conn.QueryRow(ctx, "SELECT COUNT() FROM system.tables WHERE database='default'").Scan(&count); err != nil {
-			log.Error(err, "error checking if database 'default' has tables")
-			return fmt.Errorf("check tables in  %s: %w", id, err)
-		}
-
-		if count > 0 {
-			log.Warn("database `default` has tables, but its engine is not Replicated, data loss is possible")
-		}
+// EnsureDefaultDatabaseEngine ensures that the default database engine is set to the Replicated.
+func (cmd *commander) EnsureDefaultDatabaseEngine(ctx context.Context, log controllerutil.Logger, replicas []v1.ClickHouseReplicaID) bool {
+	res := controllerutil.ExecuteParallel(replicas, func(id v1.ClickHouseReplicaID) (v1.ClickHouseReplicaID, struct{}, error) {
+		err := cmd.ensureReplicaDefaultDatabaseEngine(ctx, log, id)
+		return id, struct{}{}, err
+	})
 
-		log.Debug("dropping default database")
+	result := true
+	for id, repl := range res {
+		if repl.Err != nil {
+			log.Info("failed to recreate default database as Replicated", "replica_id", id, "error", repl.Err)
 
-		if err := conn.Exec(ctx, "DROP DATABASE default SYNC"); err != nil {
-			return fmt.Errorf("failed to drop default database on replica %s: %w", id, err)
+			result = false
 		}
 	}
 
-	log.Debug("creating replicated default database")
-
-	defaultDatabaseUUID := uuid.NewSHA1(uuid.Nil, []byte(cmd.cluster.SpecificName())).String()
-	if err = conn.Exec(ctx, createDefaultDatabaseQuery, defaultDatabaseUUID); err != nil {
-		return fmt.Errorf("create default replicated database %s: %w", id, err)
-	}
-
-	return nil
+	return result
 }
 
 func (cmd *commander) SyncShard(ctx context.Context, log controllerutil.Logger, shardID int32) error {
@@ -322,7 +336,7 @@ func (cmd *commander) SyncReplica(ctx context.Context, log controllerutil.Logger
 func (cmd *commander) CleanupDatabaseReplicas(
 	ctx context.Context,
 	log controllerutil.Logger,
-	notInSync map[v1.ClickHouseReplicaID]struct{},
+	running map[v1.ClickHouseReplicaID]struct{},
 ) error {
 	id, conn, err := cmd.getAnyConn(ctx)
 	if err != nil {
@@ -357,8 +371,8 @@ func (cmd *commander) CleanupDatabaseReplicas(
 			continue
 		}
 
-		if _, ok := notInSync[toDrop]; ok {
-			log.Debug("skipping stale database replica cleanup that is not in sync", "database", database, "replica_id", toDrop)
+		if _, ok := running[toDrop]; ok {
+			log.Debug("skipping stale database replica cleanup that is still running", "database", database, "replica_id", toDrop)
 			continue
 		}
 
@@ -442,3 +456,52 @@ func (cmd *commander) getAnyConn(ctx context.Context) (v1.ClickHouseReplicaID, c
 
 	return v1.ClickHouseReplicaID{}, nil, errors.New("no available connections")
 }
+
+func (cmd *commander) ensureReplicaDefaultDatabaseEngine(ctx context.Context, log controllerutil.Logger, id v1.ClickHouseReplicaID) error {
+	log = log.With("replica_id", id)
+
+	conn, err := cmd.getConn(id)
+	if err != nil {
+		return fmt.Errorf("failed to get connection for replica %s: %w", id, err)
+	}
+
+	var engine string
+
+	rows := conn.QueryRow(ctx, "SELECT engine FROM system.databases WHERE name='default' ")
+	if err = rows.Scan(&engine); err != nil {
+		if !errors.Is(err, sql.ErrNoRows) {
+			return fmt.Errorf("failed to scan default database engine for replica %s: %w", id, err)
+		}
+
+		log.Debug("no default database found")
+	} else {
+		if engine == "Replicated" {
+			return nil
+		}
+
+		var count uint64
+		if err = conn.QueryRow(ctx, "SELECT COUNT() FROM system.tables WHERE database='default'").Scan(&count); err != nil {
+			log.Error(err, "error checking if database 'default' has tables")
+			return fmt.Errorf("check tables in  %s: %w", id, err)
+		}
+
+		if count > 0 {
+			log.Warn("database `default` has tables, but its engine is not Replicated, data loss is possible")
+		}
+
+		log.Debug("dropping default database")
+
+		if err := conn.Exec(ctx, "DROP DATABASE default SYNC"); err != nil {
+			return fmt.Errorf("failed to drop default database on replica %s: %w", id, err)
+		}
+	}
+
+	log.Debug("creating replicated default database")
+
+	defaultDatabaseUUID := uuid.NewSHA1(uuid.Nil, []byte(cmd.cluster.SpecificName())).String()
+	if err = conn.Exec(ctx, createDefaultDatabaseQuery, defaultDatabaseUUID); err != nil {
+		return fmt.Errorf("create default replicated database %s: %w", id, err)
+	}
+
+	return nil
+}

internal/controller/clickhouse/commands_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -228,24 +229,19 @@ var _ = Describe("commander", Ordered, Label("integration"), func() {
 		}
 	})
 
-	Describe("EnsureDefaultDatabaseEngine", func() {
-		for i := range testReplicas {
-			It(fmt.Sprintf("converts default database from Atomic to Replicated on %d", i), func(ctx context.Context) {
-				id := v1.ClickHouseReplicaID{ShardID: 0, Index: i}
-
-				By("running EnsureDefaultDatabaseEngine")
+	It("converts default database from Atomic to Replicated", func(ctx context.Context) {
+		By("running EnsureDefaultDatabaseEngine")
 
-				err := cmd.EnsureDefaultDatabaseEngine(ctx, cmd.log, id)
-				Expect(err).NotTo(HaveOccurred())
+		Expect(cmd.EnsureDefaultDatabaseEngine(ctx, cmd.log, slices.Collect(cmd.cluster.ReplicaIDs()))).To(BeTrue())
 
-				By("verifying default database is now Replicated")
+		By("verifying default database is now Replicated")
 
-				dbs, err := cmd.Databases(ctx, id)
-				Expect(err).NotTo(HaveOccurred())
-				Expect(dbs).To(HaveKey("default"))
-				Expect(dbs["default"].IsReplicated).To(BeTrue())
-				Expect(dbs["default"].UUID).To(Equal(uuid.NewSHA1(uuid.Nil, []byte(cmd.cluster.SpecificName())).String()))
-			})
+		for id := range cmd.cluster.ReplicaIDs() {
+			dbs, err := cmd.Databases(ctx, id)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(dbs).To(HaveKey("default"))
+			Expect(dbs["default"].IsReplicated).To(BeTrue())
+			Expect(dbs["default"].UUID).To(Equal(uuid.NewSHA1(uuid.Nil, []byte(cmd.cluster.SpecificName())).String()))
 		}
 	})
 

internal/controller/clickhouse/config_test.go

@@ -12,22 +12,20 @@ import (
 
 var _ = Describe("ConfigGenerator", func() {
 	ctx := clickhouseReconciler{
-		reconcilerBase: reconcilerBase{
-			Cluster: &v1.ClickHouseCluster{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "test-cluster",
-					Namespace: "test-namespace",
-				},
-				Spec: v1.ClickHouseClusterSpec{
-					Replicas: new(int32(3)),
-					Shards:   new(int32(2)),
-					Settings: v1.ClickHouseSettings{
-						ExtraConfig: runtime.RawExtension{
-							Raw: []byte(`{"test": "value"}`),
-						},
-						ExtraUsersConfig: runtime.RawExtension{
-							Raw: []byte(`{}`),
-						},
+		Cluster: &v1.ClickHouseCluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-cluster",
+				Namespace: "test-namespace",
+			},
+			Spec: v1.ClickHouseClusterSpec{
+				Replicas: new(int32(3)),
+				Shards:   new(int32(2)),
+				Settings: v1.ClickHouseSettings{
+					ExtraConfig: runtime.RawExtension{
+						Raw: []byte(`{"test": "value"}`),
+					},
+					ExtraUsersConfig: runtime.RawExtension{
+						Raw: []byte(`{}`),
 					},
 				},
 			},