[Draft] CHK Raft safety planner: observation, admission, and disruption

pkg/controller/chk/controller.go

@@ -118,6 +118,19 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		return ctrl.Result{}, nil
 	}
 
+	// A staged joiner whose Pod never started has already exhausted the
+	// configured startup timeout and persisted Aborted. Child events and
+	// status updates must not restart the same failed generation. A spec
+	// edit (e.g. fixing the broken image) increments metadata.Generation
+	// and permits a new attempt.
+	if shouldHoldRaftStagedStartupAbort(new) {
+		log.V(1).M(new).F().Info(
+			"Keeper staged-joiner startup Aborted for generation %d; wait for CHK spec change",
+			new.GetGeneration(),
+		)
+		return ctrl.Result{}, nil
+	}
+
 	return c.reconcileResult(new, w.reconcileCR(ctx, nil, new))
 }
 
@@ -131,6 +144,12 @@ func (c *Controller) reconcileResult(cr *apiChk.ClickHouseKeeperInstallation, er
 	switch {
 	case err == nil:
 		return ctrl.Result{}, nil
+	case errors.Is(err, errRaftStagedJoinerStartupTimeout):
+		// Status was persisted as Aborted by reconcileCR. Do not return an
+		// error (controller-runtime retries it), and do not explicitly requeue.
+		// The generation guard above also suppresses status/child-event retries.
+		log.V(1).M(cr).F().Info("Keeper staged-joiner startup timed out; wait for spec edit")
+		return ctrl.Result{}, nil
 	case errors.Is(err, common.ErrCRUDDeferred):
 		// Soft defer: quorum headroom not available yet. Status already
 		// records [RaftQuorumUnsafe]; retry on a fixed interval instead of

pkg/controller/chk/controller_test.go

@@ -71,6 +71,15 @@ func TestReconcileResult(t *testing.T) {
 		require.Zero(t, res.RequeueAfter)
 	})
 
+	t.Run("terminal staged startup abort is not requeued", func(t *testing.T) {
+		err := fmt.Errorf("%w: %w",
+			common.ErrCRUDAbort, errRaftStagedJoinerStartupTimeout)
+		res, outErr := c.reconcileResult(cr, err)
+		require.NoError(t, outErr,
+			"terminal startup abort was already persisted as Aborted, so runtime must not retry it")
+		require.Equal(t, ctrl.Result{}, res)
+	})
+
 	t.Run("arbitrary error is returned verbatim", func(t *testing.T) {
 		boom := errors.New("boom")
 		res, err := c.reconcileResult(cr, boom)
@@ -84,3 +93,49 @@ func TestReconcileResult(t *testing.T) {
 		require.Equal(t, raftQuorumDeferredRequeueAfter, res.RequeueAfter)
 	})
 }
+
+// TestHoldRaftStagedStartupAbort ensures status and child-object watch events
+// cannot turn a persisted terminal abort back into InProgress. Only a spec
+// edit (new Kubernetes generation) should resume an aborted image pull.
+func TestHoldRaftStagedStartupAbort(t *testing.T) {
+	target := &apiChk.ClickHouseKeeperInstallation{
+		ObjectMeta: meta.ObjectMeta{Generation: 7},
+	}
+	cr := &apiChk.ClickHouseKeeperInstallation{
+		ObjectMeta: meta.ObjectMeta{Namespace: "test", Name: "keeper", Generation: 7},
+		Status: &apiChk.Status{
+			Status:       apiChk.StatusAborted,
+			NormalizedCR: target,
+			Errors:       []string{raftStagedJoinerStartupTimeoutStatusMarker + " failed to start"},
+		},
+	}
+
+	t.Run("same spec generation stays aborted after resync", func(t *testing.T) {
+		require.True(t, shouldHoldRaftStagedStartupAbort(cr))
+	})
+	t.Run("spec edit permits image recovery", func(t *testing.T) {
+		updated := cr.DeepCopy()
+		updated.Generation = 8
+		require.False(t, shouldHoldRaftStagedStartupAbort(updated))
+	})
+	t.Run("unrelated abort retains legacy behavior", func(t *testing.T) {
+		updated := cr.DeepCopy()
+		updated.Status.Errors = []string{"unrelated error"}
+		require.False(t, shouldHoldRaftStagedStartupAbort(updated))
+	})
+	t.Run("in progress is never held", func(t *testing.T) {
+		updated := cr.DeepCopy()
+		updated.Status.Status = apiChk.StatusInProgress
+		require.False(t, shouldHoldRaftStagedStartupAbort(updated))
+	})
+	t.Run("missing last-attempt generation is not a permanent lock", func(t *testing.T) {
+		updated := cr.DeepCopy()
+		updated.Status.NormalizedCR = nil
+		require.False(t, shouldHoldRaftStagedStartupAbort(updated))
+	})
+	t.Run("legacy zero generation is not a permanent lock", func(t *testing.T) {
+		updated := cr.DeepCopy()
+		updated.Status.NormalizedCR.Generation = 0
+		require.False(t, shouldHoldRaftStagedStartupAbort(updated))
+	})
+}

pkg/controller/chk/raftplan/admission.go

@@ -0,0 +1,171 @@
+// Copyright 2019 Altinity Ltd and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+package raftplan
+
+// PlanAdmission determines which Keeper hosts may be published into
+// raft_configuration and which missing StatefulSets may be created.
+//
+// It preserves the legacy staged-admission policy for established ensembles
+// with fewer than three ancestor members. Existing published members are
+// never removed merely because their StatefulSet has disappeared. Deletion
+// and scale-down are outside this policy and require a separate safety gate.
+//
+// The caller supplies DesiredHosts in stable server-ID order and must
+// provide an independently obtained, live snapshot. The function never
+// modifies any of its inputs.
+func PlanAdmission(s Snapshot) AdmissionPlan {
+	plan := AdmissionPlan{
+		MayCreate: make(map[HostID]bool),
+		Reason:    AdmissionAllowed,
+	}
+
+	// Preservation is the safe default. An unreadable ConfigMap cannot be
+	// interpreted as an empty member set.
+	included := unique(s.PublishedHosts)
+	if !s.PublishedKnown {
+		plan.Reason = ObservationUnknown
+		plan.IncludedHosts = included
+		return plan
+	}
+
+	// A stopped CHK may reconcile existing resources but must not create a
+	// new member. Preserve precisely what was already published.
+	if s.Stopped {
+		plan.IncludedHosts = included
+		plan.Reason = CHKStopped
+		return plan
+	}
+
+	// Unknown StatefulSet existence for any requested member must not be
+	// converted into an admission or a membership removal.
+	for _, id := range unique(s.DesiredHosts) {
+		host, ok := s.Hosts[id]
+		if !ok || host.StatefulSet == PresenceUnknown {
+			plan.IncludedHosts = included
+			plan.Reason = ObservationUnknown
+			return plan
+		}
+	}
+
+	// Existing voters and members with live StatefulSets are preserved,
+	// including after an interrupted creation and when normalizedCompleted
+	// hasn't yet caught up with the current live topology.
+	included = appendUnique(included, s.AncestorHosts...)
+	for _, id := range unique(s.DesiredHosts) {
+		if s.Hosts[id].StatefulSet == PresencePresent &&
+			(!s.Established || len(unique(s.AncestorHosts)) >= 3 ||
+				(s.Hosts[id].PodStarted && s.Hosts[id].StartedKnown) ||
+				(s.Hosts[id].Ready && s.Hosts[id].ReadyKnown)) {
+			included = appendUnique(included, id)
+		}
+	}
+
+	// Fresh bootstrap and 3->N intentionally retain the existing fan-out
+	// policy. The safe serial join policy applies only to an established
+	// ensemble with fewer than three ancestor members.
+	if !s.Established || len(unique(s.AncestorHosts)) >= 3 {
+		for _, id := range unique(s.DesiredHosts) {
+			included = appendUnique(included, id)
+			if s.Hosts[id].StatefulSet == PresenceAbsent {
+				plan.MayCreate[id] = true
+			}
+		}
+		return finalizeAdmission(plan, included, s.PublishedHosts)
+	}
+
+	ancestor := make(map[HostID]bool, len(s.AncestorHosts))
+	for _, id := range s.AncestorHosts {
+		ancestor[id] = true
+	}
+
+	// A previous joiner with an existing unready StatefulSet blocks the
+	// next absent joiner. The existing hosts remain eligible for safe
+	// reconciliation so they can adopt the membership configuration.
+	for _, id := range unique(s.DesiredHosts) {
+		if ancestor[id] {
+			continue
+		}
+		h := s.Hosts[id]
+		if h.StatefulSet == PresencePresent && (!h.ReadyKnown || !h.Ready) {
+			plan.BlockedBy = id
+			plan.Reason = WaitingForJoiner
+			return finalizeAdmission(plan, included, s.PublishedHosts)
+		}
+	}
+
+	// If a previously published joiner has no StatefulSet, repair that
+	// member before admitting a different one. Do not retract membership on
+	// an ambiguous create response or skip over a stranded admission.
+	for _, id := range unique(s.DesiredHosts) {
+		if ancestor[id] {
+			continue
+		}
+		if s.Hosts[id].StatefulSet == PresenceAbsent && contains(included, id) {
+			plan.MayCreate[id] = true
+			return finalizeAdmission(plan, included, s.PublishedHosts)
+		}
+	}
+
+	// Admit at most one new absent member in stable ID order.
+	for _, id := range unique(s.DesiredHosts) {
+		if ancestor[id] {
+			continue
+		}
+		if s.Hosts[id].StatefulSet == PresenceAbsent {
+			// Create with a host-scoped bootstrap config first; publishing
+			// now could exhaust Keeper's retries while the Pod is offline.
+			plan.MayCreate[id] = true
+			break
+		}
+	}
+	return finalizeAdmission(plan, included, s.PublishedHosts)
+}
+
+func finalizeAdmission(plan AdmissionPlan, included, published []HostID) AdmissionPlan {
+	plan.IncludedHosts = unique(included)
+	plan.ShouldPublish = !equalMembers(plan.IncludedHosts, published)
+	return plan
+}
+
+func contains(ids []HostID, want HostID) bool {
+	for _, id := range ids {
+		if id == want {
+			return true
+		}
+	}
+	return false
+}
+
+func unique(ids []HostID) []HostID {
+	seen := make(map[HostID]bool, len(ids))
+	result := make([]HostID, 0, len(ids))
+	for _, id := range ids {
+		if !seen[id] {
+			seen[id] = true
+			result = append(result, id)
+		}
+	}
+	return result
+}
+
+func appendUnique(ids []HostID, extra ...HostID) []HostID {
+	for _, id := range extra {
+		if !contains(ids, id) {
+			ids = append(ids, id)
+		}
+	}
+	return ids
+}
+
+func equalMembers(left, right []HostID) bool {
+	left, right = unique(left), unique(right)
+	if len(left) != len(right) {
+		return false
+	}
+	for _, id := range left {
+		if !contains(right, id) {
+			return false
+		}
+	}
+	return true
+}