fix(chk): let healthy keeper reconciles complete

pkg/controller/chk/controller.go

@@ -112,7 +112,9 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 		return ctrl.Result{}, nil
 	}
 
-	w.reconcileCR(ctx, nil, new)
+	if err := w.reconcileCR(ctx, nil, new); err != nil {
+		return ctrl.Result{}, err
+	}
 
 	return ctrl.Result{}, nil
 }

pkg/controller/chk/worker-reconciler-chk.go

@@ -116,7 +116,9 @@ func (w *worker) reconcileCR(ctx context.Context, old, new *apiChk.ClickHouseKee
 		return nil
 	}
 
-	w.markReconcileStart(ctx, new)
+	if err := w.markReconcileStart(ctx, new); err != nil {
+		return err
+	}
 	w.prepareMonitoring(new)
 	w.setHostStatusesPreliminary(ctx, new)
 
@@ -142,7 +144,9 @@ func (w *worker) reconcileCR(ctx context.Context, old, new *apiChk.ClickHouseKee
 		w.clean(ctx, new)
 		w.addToMonitoring(new)
 		w.waitForIPAddresses(ctx, new)
-		w.finalizeReconcileAndMarkCompleted(ctx, new)
+		if err := w.finalizeReconcileAndMarkCompleted(ctx, new); err != nil {
+			return err
+		}
 
 		metrics.CRReconcilesCompleted(ctx, new)
 		metrics.CRReconcilesTimings(ctx, new, time.Since(startTime).Seconds())
@@ -281,19 +285,25 @@ func (w *worker) reconcileCRAuxObjectsPreliminaryDomain(ctx context.Context, cr
 	// Use a context-aware wait so a controller shutdown does not stall the
 	// worker for up to two minutes mid-reconcile. The wait windows below are
 	// best-effort pacing only; the function returns early on ctx cancellation.
-	var d time.Duration
-	switch {
-	case cr.HostsCount() < cr.GetAncestor().HostsCount():
-		d = 120 * time.Second
-	case cr.HostsCount() > cr.GetAncestor().HostsCount():
-		d = 30 * time.Second
-	default:
-		d = 10 * time.Second
+	d := keeperMembershipSettleDelay(cr.HostsCount(), cr.GetAncestor().HostsCount())
+	if d == 0 {
+		return nil
 	}
 	util.WaitContextDoneOrTimeout(ctx, d)
 	return nil
 }
 
+func keeperMembershipSettleDelay(currentHosts, ancestorHosts int) time.Duration {
+	switch {
+	case currentHosts < ancestorHosts:
+		return 120 * time.Second
+	case currentHosts > ancestorHosts:
+		return 30 * time.Second
+	default:
+		return 0
+	}
+}
+
 // reconcileCRServicePreliminary runs first stage of CR reconcile process
 func (w *worker) reconcileCRServicePreliminary(ctx context.Context, cr api.ICustomResource) error {
 	if cr.IsStopped() {

pkg/controller/chk/worker-reconciler-chk_test.go

@@ -0,0 +1,164 @@
+// Copyright 2019 Altinity Ltd and/or its affiliates. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package chk
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	apiChk "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse-keeper.altinity.com/v1"
+	api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1"
+	"github.com/altinity/clickhouse-operator/pkg/apis/common/types"
+	a "github.com/altinity/clickhouse-operator/pkg/controller/common/announcer"
+	"github.com/altinity/clickhouse-operator/pkg/interfaces"
+)
+
+type fakeCRStatusWriter struct {
+	err     error
+	updated api.ICustomResource
+	opts    types.UpdateStatusOptions
+}
+
+func (f *fakeCRStatusWriter) Get(context.Context, string, string) (api.ICustomResource, error) {
+	return nil, nil
+}
+
+func (f *fakeCRStatusWriter) StatusUpdate(_ context.Context, cr api.ICustomResource, opts types.UpdateStatusOptions) error {
+	f.updated = cr
+	f.opts = opts
+	return f.err
+}
+
+type fakeKubeWithCR struct {
+	interfaces.IKube
+	cr interfaces.IKubeCR
+}
+
+func (f *fakeKubeWithCR) CR() interfaces.IKubeCR {
+	return f.cr
+}
+
+func newStatusTestWorker(statusWriter interfaces.IKubeCR) *worker {
+	controller := &Controller{kube: &fakeKubeWithCR{cr: statusWriter}}
+	return &worker{
+		c: controller,
+		a: a.NewAnnouncer(nil, statusWriter),
+	}
+}
+
+func TestKeeperMembershipSettleDelay(t *testing.T) {
+	tests := []struct {
+		name          string
+		currentHosts  int
+		ancestorHosts int
+		want          time.Duration
+	}{
+		{
+			name:          "same size does not wait",
+			currentHosts:  3,
+			ancestorHosts: 3,
+			want:          0,
+		},
+		{
+			name:          "upscale waits for raft membership",
+			currentHosts:  3,
+			ancestorHosts: 1,
+			want:          30 * time.Second,
+		},
+		{
+			name:          "downscale waits for raft membership",
+			currentHosts:  1,
+			ancestorHosts: 3,
+			want:          120 * time.Second,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := keeperMembershipSettleDelay(tt.currentHosts, tt.ancestorHosts); got != tt.want {
+				t.Fatalf("keeperMembershipSettleDelay(%d, %d) = %s, want %s", tt.currentHosts, tt.ancestorHosts, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestPersistReconcileCompleted(t *testing.T) {
+	target := &apiChk.ClickHouseKeeperInstallation{
+		ObjectMeta: meta.ObjectMeta{Namespace: "test", Name: "keeper"},
+	}
+	cr := &apiChk.ClickHouseKeeperInstallation{
+		ObjectMeta: meta.ObjectMeta{Namespace: "test", Name: "keeper"},
+		Status: &apiChk.Status{
+			TaskID:       "task-1",
+			NormalizedCR: target,
+		},
+	}
+	statusWriter := &fakeCRStatusWriter{}
+	w := newStatusTestWorker(statusWriter)
+
+	if err := w.persistReconcileCompleted(context.Background(), cr); err != nil {
+		t.Fatalf("persistReconcileCompleted() error = %v", err)
+	}
+
+	if statusWriter.updated != cr {
+		t.Fatal("completion status was not passed to the status writer")
+	}
+	if got := cr.EnsureStatus().GetStatus(); got != api.StatusCompleted {
+		t.Fatalf("status = %q, want %q", got, api.StatusCompleted)
+	}
+	if cr.GetAncestorT() != target {
+		t.Fatal("normalized target was not promoted to normalizedCompleted")
+	}
+	if cr.GetTarget() != nil {
+		t.Fatal("normalized target was not cleared after completion")
+	}
+	completed := cr.EnsureStatus().GetTaskIDsCompleted()
+	if len(completed) != 1 || completed[0] != "task-1" {
+		t.Fatalf("taskIDsCompleted = %v, want [task-1]", completed)
+	}
+	if !statusWriter.opts.CopyStatusOptions.FieldGroupWholeStatus {
+		t.Fatal("completion did not request a whole-status update")
+	}
+}
+
+func TestPersistReconcileCompletedReturnsStatusUpdateError(t *testing.T) {
+	wantErr := errors.New("status update rejected")
+	statusWriter := &fakeCRStatusWriter{err: wantErr}
+	w := newStatusTestWorker(statusWriter)
+	cr := &apiChk.ClickHouseKeeperInstallation{Status: &apiChk.Status{TaskID: "task-1"}}
+
+	if err := w.persistReconcileCompleted(context.Background(), cr); !errors.Is(err, wantErr) {
+		t.Fatalf("persistReconcileCompleted() error = %v, want %v", err, wantErr)
+	}
+}
+
+func TestMarkReconcileStartReturnsStatusUpdateError(t *testing.T) {
+	wantErr := errors.New("status update rejected")
+	statusWriter := &fakeCRStatusWriter{err: wantErr}
+	w := newStatusTestWorker(statusWriter)
+	cr := &apiChk.ClickHouseKeeperInstallation{Status: &apiChk.Status{TaskID: "task-1"}}
+	cr.EnsureRuntime().ActionPlan = api.MakeActionPlan(nil, cr)
+
+	if err := w.markReconcileStart(context.Background(), cr); !errors.Is(err, wantErr) {
+		t.Fatalf("markReconcileStart() error = %v, want %v", err, wantErr)
+	}
+	if got := cr.EnsureStatus().GetStatus(); got != api.StatusInProgress {
+		t.Fatalf("status = %q, want %q before persistence attempt", got, api.StatusInProgress)
+	}
+}

pkg/controller/chk/worker.go

@@ -200,39 +200,45 @@ func (w *worker) ensureFinalizer(ctx context.Context, chk *apiChk.ClickHouseKeep
 func (w *worker) finalizeCR(
 	ctx context.Context,
 	obj meta.Object,
-	updateStatusOpts types.UpdateStatusOptions,
-	f func(*apiChk.ClickHouseKeeperInstallation),
 ) error {
-	chi, err := w.buildCRFromObj(ctx, obj)
+	cr, err := w.buildCRFromObj(ctx, obj)
 	if err != nil {
 		log.V(1).Error("Unable to finalize CR: %s err: %v", util.NamespacedName(obj), err)
 		return err
 	}
+	return w.persistReconcileCompleted(ctx, cr)
+}
 
-	if f != nil {
-		f(chi)
-	}
-
-	_ = w.c.updateCRObjectStatus(ctx, chi, updateStatusOpts)
-
-	return nil
+func (w *worker) persistReconcileCompleted(ctx context.Context, cr *apiChk.ClickHouseKeeperInstallation) error {
+	cr.SetAncestor(cr.GetTarget())
+	cr.SetTarget(nil)
+	cr.EnsureStatus().ReconcileComplete()
+	return w.c.updateCRObjectStatus(ctx, cr, types.UpdateStatusOptions{
+		CopyStatusOptions: types.CopyStatusOptions{
+			CopyStatusFieldGroup: types.CopyStatusFieldGroup{
+				FieldGroupWholeStatus: true,
+			},
+		},
+	})
 }
 
-func (w *worker) markReconcileStart(ctx context.Context, cr *apiChk.ClickHouseKeeperInstallation) {
+func (w *worker) markReconcileStart(ctx context.Context, cr *apiChk.ClickHouseKeeperInstallation) error {
 	if util.IsContextDone(ctx) {
 		log.V(1).Info("Reconcile is aborted. cr: %s ", cr.GetName())
-		return
+		return ctx.Err()
 	}
 
 	// Write desired normalized CHI with initialized .Status, so it would be possible to monitor progress
 	cr.EnsureStatus().ReconcileStart(cr.EnsureRuntime().ActionPlan)
-	_ = w.c.updateCRObjectStatus(ctx, cr, types.UpdateStatusOptions{
+	if err := w.c.updateCRObjectStatus(ctx, cr, types.UpdateStatusOptions{
 		CopyStatusOptions: types.CopyStatusOptions{
 			CopyStatusFieldGroup: types.CopyStatusFieldGroup{
 				FieldGroupMain: true,
 			},
 		},
-	})
+	}); err != nil {
+		return err
+	}
 
 	w.a.V(1).
 		WithEvent(cr, a.EventActionReconcile, a.EventReasonReconcileStarted).
@@ -241,40 +247,29 @@ func (w *worker) markReconcileStart(ctx context.Context, cr *apiChk.ClickHouseKe
 		M(cr).F().
 		Info("reconcile started, task id: %s", cr.GetSpecT().GetTaskID())
 	w.a.V(2).M(cr).F().Info("action plan\n%s\n", cr.EnsureRuntime().ActionPlan.String())
+	return nil
 }
 
-func (w *worker) finalizeReconcileAndMarkCompleted(ctx context.Context, _cr *apiChk.ClickHouseKeeperInstallation) {
+func (w *worker) finalizeReconcileAndMarkCompleted(ctx context.Context, _cr *apiChk.ClickHouseKeeperInstallation) error {
 	if util.IsContextDone(ctx) {
 		log.V(1).Info("Reconcile is aborted. cr: %s ", _cr.GetName())
-		return
+		return ctx.Err()
 	}
 
 	w.a.V(1).M(_cr).F().S().Info("finalize reconcile")
 
 	// Update CR object
-	_ = w.finalizeCR(
-		ctx,
-		_cr,
-		types.UpdateStatusOptions{
-			CopyStatusOptions: types.CopyStatusOptions{
-				CopyStatusFieldGroup: types.CopyStatusFieldGroup{
-					FieldGroupWholeStatus: true,
-				},
-			},
-		},
-		func(c *apiChk.ClickHouseKeeperInstallation) {
-			c.SetAncestor(c.GetTarget())
-			c.SetTarget(nil)
-			c.EnsureStatus().ReconcileComplete()
-		},
-	)
+	if err := w.finalizeCR(ctx, _cr); err != nil {
+		return err
+	}
 
 	w.a.V(1).
 		WithEvent(_cr, a.EventActionReconcile, a.EventReasonReconcileCompleted).
 		WithAction(_cr).
 		WithActions(_cr).
 		M(_cr).F().
 		Info("reconcile completed successfully, task id: %s", _cr.GetSpecT().GetTaskID())
+	return nil
 }
 
 func (w *worker) markReconcileCompletedUnsuccessfully(ctx context.Context, cr *apiChk.ClickHouseKeeperInstallation, err error) {

tests/e2e/test_operator.py

@@ -8151,6 +8151,23 @@ def test_020003_2(self):
             kubectl.wait_field('pod', 'chk-test-020003-chk-keeper-0-1-0', '.status.containerStatuses[0].ready', 'true', retries=10)
             kubectl.wait_field('pod', 'chk-test-020003-chk-keeper-0-2-0', '.status.containerStatuses[0].ready', 'true', retries=10)
 
+        with And("CHK remains completed after the steady-state reconcile window"):
+            kubectl.wait_chk_status("test-020003-chk", "Completed")
+            task_ids_started = kubectl.get_field("chk", "test-020003-chk", ".status.taskIDsStarted")
+            task_ids_completed = kubectl.get_field("chk", "test-020003-chk", ".status.taskIDsCompleted")
+            assert task_ids_completed, error("CHK must record at least one completed reconcile task")
+
+            # Issue #2035 reproduced as a reconcile roughly every 10 seconds:
+            # status returned to InProgress and taskIDsStarted kept growing.
+            time.sleep(12)
+            assert kubectl.get_field("chk", "test-020003-chk", ".status.status") == "Completed", error()
+            assert kubectl.get_field("chk", "test-020003-chk", ".status.taskIDsStarted") == task_ids_started, error(
+                "steady-state CHK unexpectedly started another reconcile task"
+            )
+            assert kubectl.get_field("chk", "test-020003-chk", ".status.taskIDsCompleted") == task_ids_completed, error(
+                "steady-state CHK unexpectedly changed its completed task history"
+            )
+
     with Finally("I clean up"):
         delete_test_namespace()