diff --git a/pkg/controller/chk/controller.go b/pkg/controller/chk/controller.go index bef2cd13f..2942efdd3 100644 --- a/pkg/controller/chk/controller.go +++ b/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 } diff --git a/pkg/controller/chk/worker-reconciler-chk.go b/pkg/controller/chk/worker-reconciler-chk.go index 1bd95a4e7..7aa59eed4 100644 --- a/pkg/controller/chk/worker-reconciler-chk.go +++ b/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() { diff --git a/pkg/controller/chk/worker-reconciler-chk_test.go b/pkg/controller/chk/worker-reconciler-chk_test.go new file mode 100644 index 000000000..bddbc8818 --- /dev/null +++ b/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) + } +} diff --git a/pkg/controller/chk/worker.go b/pkg/controller/chk/worker.go index 61fa1b613..e86c0939c 100644 --- a/pkg/controller/chk/worker.go +++ b/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,33 +247,21 @@ 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). @@ -275,6 +269,7 @@ func (w *worker) finalizeReconcileAndMarkCompleted(ctx context.Context, _cr *api 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) { diff --git a/tests/e2e/test_operator.py b/tests/e2e/test_operator.py index ca305eca2..a9945135e 100644 --- a/tests/e2e/test_operator.py +++ b/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()