revert aggressive cleanup of live keeper version

Author: ashishch432

internal/controller/keeper/commands.go

@@ -29,6 +29,7 @@ var (
 type serverStatus struct {
 	ServerState string
 	Followers   int
+	Version     string
 }
 
 func getConnection(ctx context.Context, dialer controllerutil.DialContextFunc, hostname string, tlsRequired bool) (net.Conn, error) {
@@ -103,6 +104,13 @@ func queryKeeper(ctx context.Context, log controllerutil.Logger, conn net.Conn)
 		ServerState: statMap["zk_server_state"],
 	}
 
+	version, err := controllerutil.ParseVersion(statMap["zk_version"])
+	if err != nil {
+		log.Warn("failed to parse keeper version", "raw", statMap["zk_version"], "error", err)
+	} else {
+		result.Version = version
+	}
+
 	if result.ServerState == "" {
 		return serverStatus{}, fmt.Errorf("response missing required field 'Mode': %q", string(data))
 	}

internal/controller/keeper/controller_test.go

@@ -9,8 +9,10 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	policyv1 "k8s.io/api/policy/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -112,6 +114,11 @@ var _ = When("reconciling standalone KeeperCluster resource", Ordered, func() {
 		Expect(suite.Client.List(ctx, &statefulsets, listOpts)).To(Succeed())
 		Expect(statefulsets.Items).To(HaveLen(1))
 
+		var jobs batchv1.JobList
+		Expect(suite.Client.List(ctx, &jobs, listOpts)).To(Succeed())
+		Expect(jobs.Items).To(BeEmpty())
+		Expect(meta.FindStatusCondition(cr.Status.Conditions, v1.ConditionTypeVersionUpgraded)).To(BeNil())
+
 		testutil.AssertEvents(recorder.Events, map[string]int{
 			"ReplicaCreated":  1,
 			"ClusterNotReady": 1,

internal/controller/keeper/sync.go

@@ -7,6 +7,7 @@ import (
 	"math"
 	"slices"
 	"strconv"
+	"strings"
 	"time"
 
 	"gopkg.in/yaml.v2"
@@ -92,7 +93,6 @@ type keeperReconciler struct {
 func (r *keeperReconciler) sync(ctx context.Context, log ctrlutil.Logger) (ctrl.Result, error) {
 	log.Info("Enter Keeper Reconcile", "spec", r.Cluster.Spec, "status", r.Cluster.Status)
 
-	meta.RemoveStatusCondition(&r.Cluster.Status.Conditions, v1.ConditionTypeVersionInSync)
 	meta.RemoveStatusCondition(&r.Cluster.Status.Conditions, v1.ConditionTypeVersionUpgraded)
 
 	r.SetUnknownConditions(v1.ConditionReasonStepFailed, "Reconcile stopped before condition evaluation",
@@ -101,6 +101,7 @@ func (r *keeperReconciler) sync(ctx context.Context, log ctrlutil.Logger) (ctrl.
 			v1.ConditionTypeHealthy,
 			v1.ConditionTypeClusterSizeAligned,
 			v1.ConditionTypeConfigurationInSync,
+			v1.ConditionTypeVersionInSync,
 			v1.ConditionTypeReady,
 			v1.KeeperConditionTypeScaleAllowed,
 		})
@@ -565,6 +566,7 @@ func (r *keeperReconciler) evaluateReplicaConditions() {
 	var errorIDs, notReadyIDs, notUpdatedIDs []string
 
 	replicasByMode := map[string][]v1.KeeperReplicaID{}
+	replicaVersions := map[string]string{}
 
 	r.Cluster.Status.ReadyReplicas = 0
 	for id, replica := range r.ReplicaState {
@@ -584,11 +586,17 @@ func (r *keeperReconciler) evaluateReplicaConditions() {
 		if replica.HasDiff(r.revs) || !replica.Updated() {
 			notUpdatedIDs = append(notUpdatedIDs, idStr)
 		}
+
+		if replica.Status.Version != "" {
+			replicaVersions[idStr] = replica.Status.Version
+		}
 	}
 
 	r.SetCondition(chctrl.ReplicaStartupCondition(errorIDs))
 	r.SetCondition(chctrl.HealthyCondition(notReadyIDs))
 	r.SetCondition(chctrl.ConfigSyncCondition(nil, notUpdatedIDs, nil))
+	versionCond, versionEvents := keeperVersionSyncCondition(replicaVersions, len(notUpdatedIDs) > 0)
+	r.SetCondition(versionCond, versionEvents...)
 
 	// Ready condition — keeper-specific logic.
 	exists := len(r.ReplicaState)
@@ -667,6 +675,54 @@ func (r *keeperReconciler) evaluateReplicaConditions() {
 	)
 }
 
+func keeperVersionSyncCondition(replicaVersions map[string]string, isUpdating bool) (metav1.Condition, []chctrl.EventSpec) {
+	newCond := func(status metav1.ConditionStatus, reason v1.ConditionReason, message string) metav1.Condition {
+		return metav1.Condition{
+			Type:    v1.ConditionTypeVersionInSync,
+			Status:  status,
+			Reason:  reason,
+			Message: message,
+		}
+	}
+
+	if len(replicaVersions) == 0 {
+		return newCond(metav1.ConditionUnknown, v1.ConditionReasonVersionPending, "No Keeper replica version has been observed yet"), nil
+	}
+
+	versions := map[string]struct{}{}
+	var observed []string
+	for id, version := range replicaVersions {
+		versions[version] = struct{}{}
+		observed = append(observed, fmt.Sprintf("%s: %s", id, version))
+	}
+
+	if len(versions) == 1 {
+		var version string
+		for _, v := range replicaVersions {
+			version = v
+			break
+		}
+
+		return newCond(metav1.ConditionTrue, v1.ConditionReasonVersionMatch,
+			fmt.Sprintf("All observed Keeper replicas report version %s", version)), nil
+	}
+
+	slices.Sort(observed)
+	cond := newCond(metav1.ConditionFalse, v1.ConditionReasonVersionMismatch,
+		fmt.Sprintf("Keeper replica versions differ: %s", strings.Join(observed, ", ")))
+
+	if isUpdating {
+		return cond, nil
+	}
+
+	return cond, []chctrl.EventSpec{{
+		Type:    corev1.EventTypeWarning,
+		Reason:  v1.EventReasonVersionDiverge,
+		Action:  v1.EventActionVersionCheck,
+		Message: cond.Message,
+	}}
+}
+
 func (r *keeperReconciler) updateReplica(ctx context.Context, log ctrlutil.Logger, replicaID v1.KeeperReplicaID) (*ctrl.Result, error) {
 	log = log.With("replica_id", replicaID)
 	log.Info("updating replica")