fix: replace the blocking StatefulSet-ready wait with reconcile-loop gates

internal/controller/etcdcluster_controller.go

@@ -106,6 +106,13 @@ func (r *EtcdClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	}
 
 	if err = r.performHealthChecks(ctx, state); err != nil {
+		// While the StatefulSet is still converging (e.g. image pull after bootstrap)
+		// an unreachable member is expected; requeue instead of counting an error.
+		if !isStatefulSetSettled(state.sts) {
+			log.FromContext(ctx).Info("Health check failed while StatefulSet is converging, requeuing",
+				"reason", err.Error())
+			return ctrl.Result{RequeueAfter: requeueDuration}, nil
+		}
 		return ctrl.Result{}, err
 	}
 
@@ -277,6 +284,14 @@ func (r *EtcdClusterReconciler) reconcileClusterState(ctx context.Context, s *re
 	if s.memberListResp != nil {
 		memberCnt = len(s.memberListResp.Members)
 	}
+
+	// Don't act on a spec the StatefulSet controller hasn't observed yet (stale cache after our own write).
+	if !statefulSetSpecObserved(s.sts) {
+		logger.Info("StatefulSet spec not observed by its controller yet, requeuing",
+			"generation", s.sts.Generation, "observedGeneration", s.sts.Status.ObservedGeneration)
+		return ctrl.Result{RequeueAfter: requeueDuration}, nil
+	}
+
 	targetReplica := *s.sts.Spec.Replicas
 	var err error
 
@@ -301,6 +316,25 @@ func (r *EtcdClusterReconciler) reconcileClusterState(ctx context.Context, s *re
 		return ctrl.Result{RequeueAfter: requeueDuration}, nil
 	}
 
+	// Membership mutations below must only run against a settled StatefulSet. This gate sits
+	// after mismatch recovery on purpose: an interrupted scale-in leaves an orphaned pod whose
+	// etcd has exited, so the StatefulSet can never settle until recovery shrinks it.
+	// Spec-driven scale-in is exempt: reaching here means every member passed health checks,
+	// so removal is quorum-safe, and the member removed first is the last ordinal — often the
+	// very pod whose unreadiness prevents settling. Gating it would wedge a shrink away from
+	// a broken pod.
+	scaleIn := targetReplica > int32(s.cluster.Spec.Size)
+	if !scaleIn && !isStatefulSetSettled(s.sts) {
+		logger.Info("StatefulSet replicas not ready yet, requeuing",
+			"readyReplicas", s.sts.Status.ReadyReplicas, "replicas", targetReplica)
+		// Health checks passed but pods lag: normal briefly after a scale-out, anomalous
+		// if persistent (e.g. NotReady kubelet with etcd still serving).
+		r.warnEvent(s.cluster, "StatefulSetNotSettled", "Requeue",
+			"waiting for %d/%d ready replicas before mutating etcd membership",
+			s.sts.Status.ReadyReplicas, targetReplica)
+		return ctrl.Result{RequeueAfter: requeueDuration}, nil
+	}
+
 	var (
 		learnerStatus *clientv3.StatusResponse
 		learner       uint64
@@ -402,6 +436,13 @@ func (r *EtcdClusterReconciler) reconcileClusterState(ctx context.Context, s *re
 	return ctrl.Result{}, nil
 }
 
+// warnEvent emits a Warning event on the cluster; no-op when no Recorder is wired (unit tests).
+func (r *EtcdClusterReconciler) warnEvent(obj runtime.Object, reason, action, note string, args ...interface{}) {
+	if r.Recorder != nil {
+		r.Recorder.Eventf(obj, nil, corev1.EventTypeWarning, reason, action, note, args...)
+	}
+}
+
 // updateStatus updates the EtcdCluster status based on observed state.
 // It is called at the end of each reconciliation cycle.
 func (r *EtcdClusterReconciler) updateStatus(ctx context.Context, s *reconcileState) error {

internal/controller/etcdcluster_controller_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package controller
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -26,11 +27,15 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/events"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	ecv1alpha1 "go.etcd.io/etcd-operator/api/v1alpha1"
+	"go.etcd.io/etcd-operator/internal/etcdutils"
+	"go.etcd.io/etcd/api/v3/etcdserverpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
 // TestFetchAndValidateState describes the scenarios for the fetchAndValidateState
@@ -571,3 +576,194 @@ func TestBootstrapStatefulSet(t *testing.T) {
 		assert.Equal(t, storedCM.Data, fetchedCM.Data)
 	})
 }
+
+func memberListRespWithCount(n int) *clientv3.MemberListResponse {
+	members := make([]*etcdserverpb.Member, n)
+	for i := range members {
+		members[i] = &etcdserverpb.Member{ID: uint64(i + 1), Name: fmt.Sprintf("etcd-%d", i)}
+	}
+	return &clientv3.MemberListResponse{Members: members}
+}
+
+// TestReconcileClusterStateGates verifies the observed-generation/settled gates
+// in reconcileClusterState. The tests run without a live etcd cluster: passing
+// without one proves the gates fire before any membership call.
+func TestReconcileClusterStateGates(t *testing.T) {
+	scheme := runtime.NewScheme()
+	_ = corev1.AddToScheme(scheme)
+	_ = appsv1.AddToScheme(scheme)
+	_ = ecv1alpha1.AddToScheme(scheme)
+
+	newCluster := func(size int) *ecv1alpha1.EtcdCluster {
+		return &ecv1alpha1.EtcdCluster{
+			ObjectMeta: metav1.ObjectMeta{Name: "etcd", Namespace: "default", UID: "1"},
+			Spec:       ecv1alpha1.EtcdClusterSpec{Size: size, Version: "3.5.17"},
+		}
+	}
+	newSTS := func(generation, observedGen int64, replicas, readyReplicas int32) *appsv1.StatefulSet {
+		return &appsv1.StatefulSet{
+			ObjectMeta: metav1.ObjectMeta{Name: "etcd", Namespace: "default", Generation: generation},
+			Spec:       appsv1.StatefulSetSpec{Replicas: pointerToInt32(replicas)},
+			Status: appsv1.StatefulSetStatus{
+				ObservedGeneration: observedGen,
+				ReadyReplicas:      readyReplicas,
+			},
+		}
+	}
+
+	t.Run("stale generation requeues without writes", func(t *testing.T) {
+		ctx := t.Context()
+		ec := newCluster(5)
+		sts := newSTS(2, 1, 3, 3)
+		fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(ec, sts).Build()
+		r := &EtcdClusterReconciler{Client: fakeClient, Scheme: scheme}
+
+		stored := &appsv1.StatefulSet{}
+		require.NoError(t, fakeClient.Get(ctx, client.ObjectKey{Name: ec.Name, Namespace: ec.Namespace}, stored))
+		oldRV := stored.ResourceVersion
+
+		state := &reconcileState{cluster: ec, sts: stored, memberListResp: memberListRespWithCount(3)}
+		res, err := r.reconcileClusterState(ctx, state)
+		assert.NoError(t, err)
+		assert.Equal(t, ctrl.Result{RequeueAfter: requeueDuration}, res)
+
+		after := &appsv1.StatefulSet{}
+		require.NoError(t, fakeClient.Get(ctx, client.ObjectKey{Name: ec.Name, Namespace: ec.Namespace}, after))
+		assert.Equal(t, oldRV, after.ResourceVersion)
+	})
+
+	t.Run("unsettled replicas requeue before scale-out", func(t *testing.T) {
+		ctx := t.Context()
+		ec := newCluster(5)
+		sts := newSTS(1, 1, 3, 2)
+		fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(ec, sts).Build()
+		recorder := events.NewFakeRecorder(10)
+		r := &EtcdClusterReconciler{Client: fakeClient, Scheme: scheme, Recorder: recorder}
+
+		stored := &appsv1.StatefulSet{}
+		require.NoError(t, fakeClient.Get(ctx, client.ObjectKey{Name: ec.Name, Namespace: ec.Namespace}, stored))
+		oldRV := stored.ResourceVersion
+
+		state := &reconcileState{cluster: ec, sts: stored, memberListResp: memberListRespWithCount(3)}
+		res, err := r.reconcileClusterState(ctx, state)
+		assert.NoError(t, err)
+		assert.Equal(t, ctrl.Result{RequeueAfter: requeueDuration}, res)
+
+		after := &appsv1.StatefulSet{}
+		require.NoError(t, fakeClient.Get(ctx, client.ObjectKey{Name: ec.Name, Namespace: ec.Namespace}, after))
+		assert.Equal(t, oldRV, after.ResourceVersion)
+
+		select {
+		case ev := <-recorder.Events:
+			assert.Contains(t, ev, "StatefulSetNotSettled")
+		default:
+			t.Fatal("expected a StatefulSetNotSettled warning event")
+		}
+	})
+
+	t.Run("spec-driven scale-in is exempt from the settled gate", func(t *testing.T) {
+		ctx := t.Context()
+		ec := newCluster(1)
+		sts := newSTS(1, 1, 3, 2) // e.g. last pod NotReady while its etcd still answers
+		fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(ec, sts).Build()
+		r := &EtcdClusterReconciler{Client: fakeClient, Scheme: scheme}
+
+		stored := &appsv1.StatefulSet{}
+		require.NoError(t, fakeClient.Get(ctx, client.ObjectKey{Name: ec.Name, Namespace: ec.Namespace}, stored))
+
+		// Leaderless health data: reconcile must pass the gate and fail on the leader
+		// lookup, proving the gate did not requeue the scale-in.
+		health := make([]etcdutils.EpHealth, 3)
+		for i := range health {
+			health[i] = etcdutils.EpHealth{
+				Health: true,
+				Status: &clientv3.StatusResponse{
+					Header: &etcdserverpb.ResponseHeader{MemberId: uint64(i + 1)},
+					Leader: 99,
+				},
+			}
+		}
+
+		state := &reconcileState{cluster: ec, sts: stored, memberListResp: memberListRespWithCount(3), memberHealth: health}
+		_, err := r.reconcileClusterState(ctx, state)
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "couldn't find leader")
+	})
+
+	t.Run("interrupted scale-in recovery is not gated", func(t *testing.T) {
+		ctx := t.Context()
+		ec := newCluster(2)
+		// Member already removed but StatefulSet not yet shrunk: the orphaned pod can
+		// never become ready, so recovery must run before the settled gate.
+		sts := newSTS(1, 1, 3, 2)
+		fakeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(ec, sts).Build()
+		r := &EtcdClusterReconciler{Client: fakeClient, Scheme: scheme}
+
+		stored := &appsv1.StatefulSet{}
+		require.NoError(t, fakeClient.Get(ctx, client.ObjectKey{Name: ec.Name, Namespace: ec.Namespace}, stored))
+
+		state := &reconcileState{cluster: ec, sts: stored, memberListResp: memberListRespWithCount(2)}
+		res, err := r.reconcileClusterState(ctx, state)
+		assert.NoError(t, err)
+		assert.Equal(t, ctrl.Result{RequeueAfter: requeueDuration}, res)
+
+		after := &appsv1.StatefulSet{}
+		require.NoError(t, fakeClient.Get(ctx, client.ObjectKey{Name: ec.Name, Namespace: ec.Namespace}, after))
+		assert.Equal(t, int32(2), *after.Spec.Replicas)
+	})
+}
+
+// TestReconcileClusterStateStaleGenerationEnvtest exercises the stale-generation
+// gate against real API-server generation semantics: envtest runs no
+// kube-controller-manager, so Status.ObservedGeneration stays 0 while
+// Generation is 1 after creation.
+func TestReconcileClusterStateStaleGenerationEnvtest(t *testing.T) {
+	ctx := t.Context()
+
+	ec := &ecv1alpha1.EtcdCluster{
+		ObjectMeta: metav1.ObjectMeta{Name: "gate-envtest", Namespace: "default"},
+		Spec:       ecv1alpha1.EtcdClusterSpec{Size: 1, Version: "3.5.17"},
+	}
+	require.NoError(t, k8sClient.Create(ctx, ec))
+	t.Cleanup(func() { _ = k8sClient.Delete(ctx, ec) })
+
+	labels := map[string]string{"app": ec.Name}
+	sts := &appsv1.StatefulSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      ec.Name,
+			Namespace: ec.Namespace,
+			OwnerReferences: []metav1.OwnerReference{{
+				APIVersion: ecv1alpha1.GroupVersion.String(),
+				Kind:       "EtcdCluster",
+				Name:       ec.Name,
+				UID:        ec.UID,
+				Controller: pointerToBool(true),
+			}},
+		},
+		Spec: appsv1.StatefulSetSpec{
+			Replicas:    pointerToInt32(1),
+			ServiceName: ec.Name,
+			Selector:    &metav1.LabelSelector{MatchLabels: labels},
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{Labels: labels},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{{Name: "etcd", Image: "gcr.io/etcd-development/etcd:3.5.17"}},
+				},
+			},
+		},
+	}
+	require.NoError(t, k8sClient.Create(ctx, sts))
+	t.Cleanup(func() { _ = k8sClient.Delete(ctx, sts) })
+
+	fetched := &appsv1.StatefulSet{}
+	require.NoError(t, k8sClient.Get(ctx, client.ObjectKey{Name: ec.Name, Namespace: ec.Namespace}, fetched))
+	require.Equal(t, int64(1), fetched.Generation)
+	require.Equal(t, int64(0), fetched.Status.ObservedGeneration)
+
+	r := &EtcdClusterReconciler{Client: k8sClient, Scheme: k8sClient.Scheme()}
+	state := &reconcileState{cluster: ec, sts: fetched, memberListResp: memberListRespWithCount(1)}
+
+	res, err := r.reconcileClusterState(ctx, state)
+	assert.NoError(t, err)
+	assert.Equal(t, ctrl.Result{RequeueAfter: requeueDuration}, res)
+}