Stable order of replicas removal, wait for complete deletion. Remove misleading structures from Status

Author: GrigoryPervakov

api/v1alpha1/keepercluster_types.go

@@ -166,18 +166,6 @@ func (s KeeperReplicaState) Active() bool {
 	return activeReplicaStates[s]
 }
 
-type KeeperReplica struct {
-	// Ready indicates that replica is ready to accept incoming connections.
-	Ready bool `json:"ready"`
-	// Error indicates that replica has a persistent error causing Pod startup failure.
-	Error bool `json:"error"`
-	// Updated indicates that replica Pod is updated to the latest StatefulSet revision.
-	Updated bool `json:"updated"`
-	// Mode indicates replica role, during latest reconcile.
-	// +optional
-	Mode string `json:"mode"`
-}
-
 // KeeperClusterStatus defines the observed state of KeeperCluster.
 type KeeperClusterStatus struct {
 	// +listType=map
@@ -186,9 +174,10 @@ type KeeperClusterStatus struct {
 	// +patchMergeKey=type
 	// +optional
 	Conditions []metav1.Condition `json:"conditions,omitempty"`
-	// Replicas Current replica state by Keeper ID.
+	// Replicas that should be present in config
+	// TODO probably can be tracked by rereading quorum config.
 	// +optional
-	Replicas map[string]KeeperReplica `json:"replicas"`
+	Replicas []string `json:"replicas"`
 	// ReadyReplicas Total number of replicas ready to server requests.
 	// +optional
 	ReadyReplicas int32 `json:"readyReplicas"`
@@ -272,7 +261,7 @@ func (v *KeeperCluster) HostnameById(replicaID string) string {
 
 func (v *KeeperCluster) Hostnames() []string {
 	hostnames := make([]string, 0, len(v.Status.Replicas))
-	for id := range v.Status.Replicas {
+	for _, id := range v.Status.Replicas {
 		hostnames = append(hostnames, v.HostnameById(id))
 	}
 
@@ -281,7 +270,7 @@ func (v *KeeperCluster) Hostnames() []string {
 
 func (v *KeeperCluster) HostnamesByID() map[string]string {
 	hostnameByID := map[string]string{}
-	for id := range v.Status.Replicas {
+	for _, id := range v.Status.Replicas {
 		hostnameByID[id] = v.HostnameById(id)
 	}
 

api/v1alpha1/zz_generated.deepcopy.go

@@ -1664,30 +1664,10 @@ spec:
                 format: int32
                 type: integer
               replicas:
-                additionalProperties:
-                  properties:
-                    error:
-                      description: Error indicates that replica has a persistent error
-                        causing Pod startup failure.
-                      type: boolean
-                    mode:
-                      description: Mode indicates replica role, during latest reconcile.
-                      type: string
-                    ready:
-                      description: Ready indicates that replica is ready to accept
-                        incoming connections.
-                      type: boolean
-                    updated:
-                      description: Updated indicates that replica Pod is updated to
-                        the latest StatefulSet revision.
-                      type: boolean
-                  required:
-                  - error
-                  - ready
-                  - updated
-                  type: object
-                description: Replicas Current replica state by Keeper ID.
-                type: object
+                description: Replicas that should be present in config
+                items:
+                  type: string
+                type: array
               updateRevision:
                 description: CurrentRevision indicates latest requested KeeperCluster
                   spec revision.

config/crd/bases/clickhouse.com_keeperclusters.yaml

@@ -10,8 +10,6 @@ spec:
   image:
     repository: "docker.io/clickhouse/clickhouse-keeper"
     tag: latest
-  loggerConfig:
-    loggerLevel: warning
   storage:
     accessModes:
       - ReadWriteOncePod

config/samples/v1alpha1_keepercluster.yaml

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -14,21 +15,21 @@ import (
 )
 
 const (
-	FLWCommand = "srvr"
+	FLWCommand = "mntr"
 
 	ModeLeader     = "leader"
 	ModeFollower   = "follower"
 	ModeStandalone = "standalone"
 )
 
 var (
-	ClusterModes    = []string{ModeLeader, ModeFollower}
-	StandaloneModes = []string{ModeStandalone}
+	ClusterModes = []string{ModeLeader, ModeFollower}
 )
 
-// ServerState holds parsed fields of the "srvr" command response.
-type ServerState struct {
-	Mode string
+// ServerStatus holds parsed fields of the "mntr" command response.
+type ServerStatus struct {
+	ServerState string
+	Followers   int
 }
 
 type Connections map[string]net.Conn
@@ -59,54 +60,68 @@ func getConnections(ctx context.Context, log util.Logger, hostnamesByID map[stri
 	return result, connErrs
 }
 
-func queryPod(ctx context.Context, log util.Logger, conn net.Conn) (ServerState, error) {
+func queryPod(ctx context.Context, log util.Logger, conn net.Conn) (ServerStatus, error) {
 	log.Debug(fmt.Sprintf("querying keeper pod: %s", conn.RemoteAddr().String()))
 	if dl, ok := ctx.Deadline(); ok {
 		if err := conn.SetDeadline(dl); err != nil {
-			return ServerState{}, fmt.Errorf("set deadline: %w", err)
+			return ServerStatus{}, fmt.Errorf("set deadline: %w", err)
 		}
 	}
 
 	n, err := io.WriteString(conn, FLWCommand)
 	if err != nil {
-		return ServerState{}, fmt.Errorf("write command: %w", err)
+		return ServerStatus{}, fmt.Errorf("write command: %w", err)
 	}
 	if n != len(FLWCommand) {
-		return ServerState{}, fmt.Errorf("can't write the whole string to socket expected: %d; actual: %d", len(FLWCommand), n)
+		return ServerStatus{}, fmt.Errorf("can't write the whole string to socket expected: %d; actual: %d", len(FLWCommand), n)
 	}
 
 	reader := bufio.NewReader(conn)
 	data, err := io.ReadAll(reader)
 	if err != nil {
-		return ServerState{}, fmt.Errorf("got error while reading from socket: %w", err)
+		return ServerStatus{}, fmt.Errorf("got error while reading from socket: %w", err)
 	}
 
 	statMap := map[string]string{}
 	for i, stat := range strings.Split(string(data), "\n") {
 		if len(stat) == 0 {
 			continue
 		}
-		parts := strings.Split(stat, ": ")
+		parts := strings.Split(stat, "\t")
 		if len(parts) != 2 {
-			return ServerState{}, fmt.Errorf("failed to parse response line %d: %q", i, stat)
+			return ServerStatus{}, fmt.Errorf("failed to parse response line %d: %q", i, stat)
 		}
 
 		statMap[parts[0]] = parts[1]
 	}
 
-	result := ServerState{Mode: statMap["Mode"]}
-	if result.Mode == "" {
-		return ServerState{}, fmt.Errorf("response missing required field 'Mode': %q", string(data))
+	result := ServerStatus{
+		ServerState: statMap["zk_server_state"],
+	}
+	if result.ServerState == "" {
+		return ServerStatus{}, fmt.Errorf("response missing required field 'Mode': %q", string(data))
+	}
+
+	if result.ServerState == ModeLeader {
+		if followers, ok := statMap["zk_followers"]; ok {
+			result.Followers, err = strconv.Atoi(followers)
+			if err != nil {
+				return ServerStatus{}, fmt.Errorf("failed to parse field 'zk_followers': %w", err)
+			}
+		} else {
+			log.Warn("'zk_followers' is missing in keeper response")
+			return ServerStatus{}, fmt.Errorf("response missing required field 'Followers': %q", string(data))
+		}
 	}
 
 	return result, nil
 }
 
-func queryAllPods(ctx context.Context, log util.Logger, connections Connections) map[string]ServerState {
+func queryAllPods(ctx context.Context, log util.Logger, connections Connections) map[string]ServerStatus {
 	// Contains the hostName along with parsed response from it
 	type NamedFourLetterCommandResponse struct {
 		id    string
-		state ServerState
+		state ServerStatus
 	}
 
 	// Buffered channels will block only if buffer is full
@@ -119,7 +134,7 @@ func queryAllPods(ctx context.Context, log util.Logger, connections Connections)
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			resp, err := queryPod(ctx, log, conn)
+			resp, err := queryPod(ctx, log.With("replica_id", id), conn)
 			if err != nil {
 				fails <- err
 			} else {
@@ -128,7 +143,7 @@ func queryAllPods(ctx context.Context, log util.Logger, connections Connections)
 		}()
 	}
 
-	result := map[string]ServerState{}
+	result := map[string]ServerStatus{}
 	errs := []error{}
 
 	// Wait for all goroutines to return error or result
@@ -152,7 +167,7 @@ func queryAllPods(ctx context.Context, log util.Logger, connections Connections)
 	return result
 }
 
-func getServersStates(ctx context.Context, log util.Logger, hostnamesByID map[string]string) map[string]ServerState {
+func getServersStates(ctx context.Context, log util.Logger, hostnamesByID map[string]string) map[string]ServerStatus {
 	connections, errs := getConnections(ctx, log, hostnamesByID)
 	for _, err := range errs {
 		log.Info("error getting keeper connection", "error", err)