From 2e9186b28606bfeb514ba7349a4fc82b08d2c659 Mon Sep 17 00:00:00 2001 From: alz Date: Fri, 21 Aug 2026 16:52:04 +0300 Subject: [PATCH 1/6] Protect CHK recreate with live Raft quorum safety. Refuse disruptive STS changes when Ready members are at majority, wait Ready only when live quorum exists, abort on STS wait failure, and drop the same-size 10s settle sleep while propagating status persist errors. Fixes #2069; partial #2059. Co-authored-by: Cursor --- pkg/controller/chk/controller.go | 4 +- pkg/controller/chk/worker-raft-safety.go | 191 ++++++++++++++++++ pkg/controller/chk/worker-raft-safety_test.go | 158 +++++++++++++++ pkg/controller/chk/worker-reconciler-chk.go | 121 ++++++----- .../chk/worker-reconciler-chk_test.go | 164 +++++++++++++++ pkg/controller/chk/worker.go | 70 +++---- tests/regression.py | 2 +- 7 files changed, 616 insertions(+), 94 deletions(-) create mode 100644 pkg/controller/chk/worker-raft-safety.go create mode 100644 pkg/controller/chk/worker-raft-safety_test.go create mode 100644 pkg/controller/chk/worker-reconciler-chk_test.go 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-raft-safety.go b/pkg/controller/chk/worker-raft-safety.go new file mode 100644 index 000000000..80ba543d8 --- /dev/null +++ b/pkg/controller/chk/worker-raft-safety.go @@ -0,0 +1,191 @@ +// 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" + "fmt" + + apps "k8s.io/api/apps/v1" + + api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1" + "github.com/altinity/clickhouse-operator/pkg/controller/common" + "github.com/altinity/clickhouse-operator/pkg/controller/common/statefulset" + "github.com/altinity/clickhouse-operator/pkg/interfaces" +) + +// Raft / ensemble safety for CHK (#2069). +// +// Live Ready count decides the mode — not CR ancestor / host inventory: +// +// - Live quorum (Ready members >= majority): rolling recreate of a healthy +// ensemble — wait Ready before the next host, and refuse to disrupt a host +// if doing so would drop below quorum. +// - No live quorum (fresh install, resume-from-stopped, or already broken): +// bootstrap / recovery — wait Started only so siblings can come up together. +// +// Hooks below stay thin so a fuller Raft-membership barrier (committed +// /keeper/config + mntr, as in PR #2041) can replace verifyHostEnsembleMembership +// without reshaping the reconcile loop. + +// chkStatefulSetFallback aborts the reconcile on STS create/update wait failure. +// DefaultFallback returns ErrCRUDIgnore, which lets the host loop recreate the +// next replica while the previous one never rejoined — the #2069 failure mode. +type chkStatefulSetFallback struct{} + +func newChkStatefulSetFallback() *chkStatefulSetFallback { + return &chkStatefulSetFallback{} +} + +func (f *chkStatefulSetFallback) OnStatefulSetCreateFailed(ctx context.Context, host *api.Host) common.ErrorCRUD { + return common.ErrCRUDAbort +} + +func (f *chkStatefulSetFallback) OnStatefulSetUpdateFailed( + ctx context.Context, + oldStatefulSet *apps.StatefulSet, + host *api.Host, + sts interfaces.IKubeSTS, +) common.ErrorCRUD { + return common.ErrCRUDAbort +} + +// raftQuorumSize is Raft majority for an ensemble of n members (n/2 + 1). +func raftQuorumSize(members int) int { + if members <= 0 { + return 0 + } + return members/2 + 1 +} + +// countReadyEnsembleMembers counts Keeper hosts whose StatefulSet reports +// ReadyReplicas > 0 (each host is typically a 1-replica STS). +// +// countReadyEnsembleMembersFn, when set on the worker, overrides live lookup +// (tests inject fixed Ready counts). +func (w *worker) countReadyEnsembleMembers(ctx context.Context, cr api.ICustomResource) int { + if w.countReadyEnsembleMembersFn != nil { + return w.countReadyEnsembleMembersFn(ctx, cr) + } + if cr == nil { + return 0 + } + ready := 0 + _ = cr.WalkHosts(func(host *api.Host) error { + sts := host.Runtime.CurStatefulSet + if sts == nil && w.c != nil { + sts, _ = w.c.kube.STS().Get(ctx, host) + } + if sts != nil && sts.Status.ReadyReplicas > 0 { + ready++ + } + return nil + }) + return ready +} + +// ensembleHasLiveQuorum reports whether enough hosts are Ready to form a Raft +// majority. Resume-from-stopped and fresh bootstrap both yield false (0 Ready) +// and therefore take the fast startup path. +func (w *worker) ensembleHasLiveQuorum(ctx context.Context, cr api.ICustomResource) bool { + if cr == nil { + return false + } + n := cr.HostsCount() + return w.countReadyEnsembleMembers(ctx, cr) >= raftQuorumSize(n) +} + +// hostContributesReady reports whether this host currently counts toward live quorum. +func hostContributesReady(host *api.Host) bool { + if host == nil || host.Runtime.CurStatefulSet == nil { + return false + } + return host.Runtime.CurStatefulSet.Status.ReadyReplicas > 0 +} + +// ensureQuorumSafeToDisruptHost refuses to take down a Ready host when the +// remaining Ready members would fall below quorum. No-op when the ensemble +// already lacks live quorum (bootstrap / resume-from-stopped / recovery). +func (w *worker) ensureQuorumSafeToDisruptHost(ctx context.Context, host *api.Host) error { + cr := host.GetCR() + if cr == nil { + return nil + } + n := cr.HostsCount() + q := raftQuorumSize(n) + ready := w.countReadyEnsembleMembers(ctx, cr) + if ready < q { + return nil + } + remaining := ready + if hostContributesReady(host) { + remaining-- + } + if remaining < q { + return fmt.Errorf( + "refusing to disrupt host %s: ready=%d remaining=%d quorum=%d (would drop below Raft quorum)", + host.GetName(), ready, remaining, q, + ) + } + return nil +} + +// prepareStsReconcileOptsWaitSection sets STS launch waits for Keeper. +// +// Live quorum present: wait until Ready before moving on. +// No live quorum: wait Started only — Ready would deadlock until siblings exist +// (bootstrap, resume-from-stopped, or recovery). +func (w *worker) prepareStsReconcileOptsWaitSection(ctx context.Context, host *api.Host, opts *statefulset.ReconcileOptions) *statefulset.ReconcileOptions { + if opts == nil { + opts = statefulset.NewReconcileStatefulSetOptions() + } + probes := host.GetCluster().GetReconcile().Host.Wait.Probes + liveQuorum := w.ensembleHasLiveQuorum(ctx, host.GetCR()) + + if probes.GetStartup().IsTrue() || !liveQuorum { + opts = opts.SetWaitUntilStarted() + w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilStarted") + } + + switch { + case liveQuorum && !probes.GetReadiness().IsFalse(): + opts = opts.SetWaitUntilReady() + w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilReady (live Raft quorum)") + case !liveQuorum: + w.a.V(1).M(host).F().Info("Skip WaitUntilReady — no live quorum (bootstrap / resume-from-stopped / recovery)") + } + + return opts +} + +// waitHostRaftJoined is the post-host hook for confirming the replica is part of +// the live ensemble before the reconciler advances. +// +// Today a no-op beyond what STS Ready wait already enforced. Replace or extend +// verifyHostEnsembleMembership with committed Raft membership checks as in +// Altinity/clickhouse-operator#2041. +func (w *worker) waitHostRaftJoined(ctx context.Context, host *api.Host) error { + return w.verifyHostEnsembleMembership(ctx, host) +} + +// verifyHostEnsembleMembership is the extension point for Raft membership +// verification. Currently a no-op: STS Ready wait already ran when live quorum +// mode applied. Implement committed-config / leader sync barriers here when +// adopting the fuller rescale design from PR #2041. +func (w *worker) verifyHostEnsembleMembership(ctx context.Context, host *api.Host) error { + _ = ctx + _ = host + return nil +} diff --git a/pkg/controller/chk/worker-raft-safety_test.go b/pkg/controller/chk/worker-raft-safety_test.go new file mode 100644 index 000000000..2134fff69 --- /dev/null +++ b/pkg/controller/chk/worker-raft-safety_test.go @@ -0,0 +1,158 @@ +// 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" + "testing" + + "github.com/stretchr/testify/require" + apps "k8s.io/api/apps/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" + "github.com/altinity/clickhouse-operator/pkg/controller/common" + "github.com/altinity/clickhouse-operator/pkg/controller/common/statefulset" +) + +func TestRaftQuorumSize(t *testing.T) { + require.Equal(t, 0, raftQuorumSize(0)) + require.Equal(t, 1, raftQuorumSize(1)) + require.Equal(t, 2, raftQuorumSize(3)) + require.Equal(t, 3, raftQuorumSize(5)) +} + +func TestEnsembleHasLiveQuorum(t *testing.T) { + cr := chkWithHosts(3) + w := &worker{} + + t.Run("no ready members — bootstrap / resume-from-stopped", func(t *testing.T) { + w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 0 } + require.False(t, w.ensembleHasLiveQuorum(context.Background(), cr)) + }) + + t.Run("below quorum", func(t *testing.T) { + w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 1 } + require.False(t, w.ensembleHasLiveQuorum(context.Background(), cr)) + }) + + t.Run("at quorum", func(t *testing.T) { + w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 2 } + require.True(t, w.ensembleHasLiveQuorum(context.Background(), cr)) + }) +} + +func TestPrepareStsReconcileOptsWaitSection(t *testing.T) { + ctx := context.Background() + + t.Run("no live quorum skips Ready", func(t *testing.T) { + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, + } + host := hostOnCR(chkWithHosts(3)) + opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil) + require.True(t, opts.WaitUntilStarted()) + require.False(t, opts.WaitUntilReady()) + }) + + t.Run("live quorum waits Ready", func(t *testing.T) { + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 3 }, + } + host := hostOnCR(chkWithHosts(3)) + opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil) + require.True(t, opts.WaitUntilReady()) + }) + + t.Run("live quorum can opt out of Ready", func(t *testing.T) { + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 3 }, + } + host := hostOnCR(chkWithHosts(3)) + host.GetCluster().GetReconcile().Host.Wait.Probes.Readiness = types.NewStringBool(false) + opts := w.prepareStsReconcileOptsWaitSection(ctx, host, statefulset.NewReconcileStatefulSetOptions()) + require.False(t, opts.WaitUntilReady()) + }) +} + +func TestEnsureQuorumSafeToDisruptHost(t *testing.T) { + ctx := context.Background() + cr := chkWithHosts(3) + host := hostOnCR(cr) + host.Runtime.CurStatefulSet = &apps.StatefulSet{} + host.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + + t.Run("allows disrupt when no live quorum", func(t *testing.T) { + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, + } + require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, host)) + }) + + t.Run("allows disrupt when siblings keep quorum", func(t *testing.T) { + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 3 }, + } + require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, host)) + }) + + t.Run("refuses disrupt when remaining would be below quorum", func(t *testing.T) { + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 2 }, + } + err := w.ensureQuorumSafeToDisruptHost(ctx, host) + require.Error(t, err) + require.Contains(t, err.Error(), "would drop below Raft quorum") + }) +} + +func TestChkStatefulSetFallbackAborts(t *testing.T) { + f := newChkStatefulSetFallback() + require.Equal(t, common.ErrCRUDAbort, f.OnStatefulSetCreateFailed(nil, nil)) + require.Equal(t, common.ErrCRUDAbort, f.OnStatefulSetUpdateFailed(nil, nil, nil, nil)) +} + +func chkWithHosts(n int) *apiChk.ClickHouseKeeperInstallation { + cr := &apiChk.ClickHouseKeeperInstallation{} + cr.EnsureRuntime() + cluster := &apiChk.Cluster{Name: "c"} + cluster.Layout = apiChk.NewChkClusterLayout() + shard := &apiChk.ChkShard{Name: "s"} + for i := 0; i < n; i++ { + h := &api.Host{Name: "h"} + h.Runtime.Address.ClusterName = cluster.Name + h.Runtime.Address.ShardName = shard.Name + h.Runtime.Address.HostName = "h" + shard.Hosts = append(shard.Hosts, h) + } + cluster.Layout.Shards = []*apiChk.ChkShard{shard} + cluster.Runtime.CHK = cr + cluster.Reconcile = (&api.ClusterReconcile{}).Ensure() + cluster.Reconcile.Host.Wait.Probes = &api.ReconcileHostWaitProbes{} + cr.Spec.Configuration = &apiChk.Configuration{ + Clusters: []*apiChk.Cluster{cluster}, + } + return cr +} + +func hostOnCR(cr *apiChk.ClickHouseKeeperInstallation) *api.Host { + cluster := cr.Spec.Configuration.Clusters[0] + host := cluster.Layout.Shards[0].Hosts[0] + host.SetCR(cr) + host.Runtime.Address.ClusterName = cluster.Name + host.Runtime.Address.ShardName = cluster.Layout.Shards[0].Name + return host +} diff --git a/pkg/controller/chk/worker-reconciler-chk.go b/pkg/controller/chk/worker-reconciler-chk.go index 1bd95a4e7..3ed94e88f 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) @@ -131,23 +133,26 @@ func (w *worker) reconcileCR(ctx context.Context, old, new *apiChk.ClickHouseKee if errors.Is(err, common.ErrCRUDAbort) { metrics.CRReconcilesAborted(ctx, new) } - } else { - // Reconcile successful - // Post-process added items - if util.IsContextDone(ctx) { - log.V(1).Info("Reconcile is aborted. CR post-process: %s ", new.GetName()) - return nil - } + return err + } - w.clean(ctx, new) - w.addToMonitoring(new) - w.waitForIPAddresses(ctx, new) - w.finalizeReconcileAndMarkCompleted(ctx, new) + // Reconcile successful + // Post-process added items + if util.IsContextDone(ctx) { + log.V(1).Info("Reconcile is aborted. CR post-process: %s ", new.GetName()) + return nil + } - metrics.CRReconcilesCompleted(ctx, new) - metrics.CRReconcilesTimings(ctx, new, time.Since(startTime).Seconds()) + w.clean(ctx, new) + w.addToMonitoring(new) + w.waitForIPAddresses(ctx, new) + if err := w.finalizeReconcileAndMarkCompleted(ctx, new); err != nil { + return err } + metrics.CRReconcilesCompleted(ctx, new) + metrics.CRReconcilesTimings(ctx, new, time.Since(startTime).Seconds()) + return nil } @@ -281,19 +286,29 @@ 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 + // Same-size reconciles do not wait (see #2035 / #2059) — a fixed 10s sleep + // previously left healthy ensembles cycling and delayed Completed. + d := keeperMembershipSettleDelay(cr.HostsCount(), cr.GetAncestor().HostsCount()) + if d == 0 { + return nil } util.WaitContextDoneOrTimeout(ctx, d) return nil } +// keeperMembershipSettleDelay is a best-effort pause after publishing membership +// changes so Raft can settle. Same host count → no delay. +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() { @@ -415,6 +430,10 @@ func (w *worker) reconcileHostStatefulSet(ctx context.Context, host *api.Host, o // Start with force-restart host forcedRestart := false if w.shouldForceRestartHost(ctx, host) { + if err := w.ensureQuorumSafeToDisruptHost(ctx, host); err != nil { + w.a.V(1).M(host).F().Error("%v", err) + return err + } w.a.V(1).M(host).F().Info("Reconcile host STS force restart: %s", host.GetName()) _ = w.hostForceRestart(ctx, host, opts) forcedRestart = true @@ -432,7 +451,16 @@ func (w *worker) reconcileHostStatefulSet(ctx context.Context, host *api.Host, o w.a.V(1).M(host).F().Info("Override ObjectStatusSame after force restart to ensure scale-up: %s", host.GetName()) host.GetReconcileAttributes().SetStatus(types.ObjectStatusModified) } - opts = w.prepareStsReconcileOptsWaitSection(host, opts) + // Update / recreate / force-recreate can take a Ready host down. Refuse when that + // would leave the ensemble below Raft quorum (#2069). No-op when there is no live + // quorum (bootstrap, resume-from-stopped). + if (opts != nil && opts.ForceRecreate()) || !host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusSame) { + if err := w.ensureQuorumSafeToDisruptHost(ctx, host); err != nil { + w.a.V(1).M(host).F().Error("%v", err) + return err + } + } + opts = w.prepareStsReconcileOptsWaitSection(ctx, host, opts) // We are in place, where we can reconcile StatefulSet to desired configuration. w.a.V(1).M(host).F().Info("Reconcile host STS: %s. Reconcile StatefulSet", host.GetName()) @@ -802,36 +830,20 @@ func (w *worker) reconcileHostMain(ctx context.Context, host *api.Host) error { Warning("Reconcile Host Main - unable to reconcile PVC. Host: %s Err: %v", host.GetName(), err) } - // Finalize main reconcile with domain activities + // Finalize main reconcile with domain activities. + // Membership / ensemble gates must abort the host loop — continuing would + // recreate the next replica while this one never rejoined (#2069). if err := w.reconcileHostMainDomain(ctx, host); err != nil { + metrics.HostReconcilesErrors(ctx, host.GetCR()) w.a.V(1). M(host).F(). - Warning("Reconcile Host Main - unable to reconcile domain reconcile. Host: %s Err: %v", host.GetName(), err) + Warning("Reconcile Host Main - ensemble join gate failed. Host: %s Err: %v", host.GetName(), err) + return err } return nil } -func (w *worker) prepareStsReconcileOptsWaitSection(host *api.Host, opts *statefulset.ReconcileOptions) *statefulset.ReconcileOptions { - probes := host.GetCluster().GetReconcile().Host.Wait.Probes - // Startup is required for newly starting node - if probes.GetStartup().IsTrue() || !host.HasAncestor() { - opts = opts.SetWaitUntilStarted() - w.a.V(1). - M(host).F(). - Warning("Setting option SetWaitUntilStarted") - } - // Readiness requires Raft quorum. New hosts (no ancestor) cannot satisfy - // readiness until all siblings start and form quorum — skip to avoid deadlock. - if probes.GetReadiness().IsTrue() && host.HasAncestor() { - opts = opts.SetWaitUntilReady() - w.a.V(1). - M(host).F(). - Warning("Setting option SetWaitUntilReady") - } - return opts -} - func (w *worker) reconcileHostPVCs(ctx context.Context, host *api.Host) storage.ErrorDataPersistence { return storage.NewStorageReconciler( w.task, @@ -841,19 +853,20 @@ func (w *worker) reconcileHostPVCs(ctx context.Context, host *api.Host) storage. } func (w *worker) reconcileHostMainDomain(ctx context.Context, host *api.Host) error { - // Should we wait for host to startup - wait := false - - if host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) { - wait = true + if !host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) { + return nil } - // Wait for host to startup; respect ctx cancellation so the worker - // unblocks on controller shutdown instead of running out the timer. - if wait { + if !w.ensembleHasLiveQuorum(ctx, host.GetCR()) { + // Bootstrap / resume-from-stopped / recovery: peers start together; + // keep the legacy pacing wait (Ready wait was skipped). util.WaitContextDoneOrTimeout(ctx, 7*time.Second) + return nil } - return nil + + // Live quorum mode: STS Ready wait already ran. Hook for richer Raft + // membership confirmation (PR #2041) lives in waitHostRaftJoined. + return w.waitHostRaftJoined(ctx, host) } // reconcileHostIncludeIntoAllActivities includes specified ClickHouse host into all activities 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..7c10f1fe8 100644 --- a/pkg/controller/chk/worker.go +++ b/pkg/controller/chk/worker.go @@ -56,6 +56,9 @@ type worker struct { task *common.Task stsReconciler *statefulset.Reconciler + // countReadyEnsembleMembersFn overrides live Ready counting (tests only). + countReadyEnsembleMembersFn func(ctx context.Context, cr api.ICustomResource) int + start time.Time } @@ -119,7 +122,7 @@ func (w *worker) newTask(new, old *apiChk.ClickHouseKeeperInstallation) { labeler.New(new), storage.NewStorageReconciler(w.task, w.c.namer, w.c.kube.Storage()), w.c.kube, - statefulset.NewDefaultFallback(), + newChkStatefulSetFallback(), ) } @@ -197,42 +200,45 @@ func (w *worker) ensureFinalizer(ctx context.Context, chk *apiChk.ClickHouseKeep return true } -func (w *worker) finalizeCR( - ctx context.Context, - obj meta.Object, - updateStatusOpts types.UpdateStatusOptions, - f func(*apiChk.ClickHouseKeeperInstallation), -) error { - chi, err := w.buildCRFromObj(ctx, obj) +func (w *worker) finalizeCR(ctx context.Context, obj meta.Object) error { + 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,20 @@ 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 +268,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/regression.py b/tests/regression.py index 15fcd442f..89cef28ce 100755 --- a/tests/regression.py +++ b/tests/regression.py @@ -9,7 +9,7 @@ # test_operator.py "/regression/e2e.test_operator/test_010021*": [(Fail, "Storage test is flaky on github")], "/regression/e2e.test_operator/test_010082_1*": [(Fail, "Canary via CHIT injection does not work")], - "/regression/e2e.test_operator/test_020005*": [(Fail, "Keeper scale-up/scale-down is flaky")], + # "/regression/e2e.test_operator/test_020005*": [(Fail, "Keeper scale-up/scale-down is flaky")], } From 5cc0235da4e27d0dfd0bbfabf5042867087bbe36 Mon Sep 17 00:00:00 2001 From: alz Date: Sat, 22 Aug 2026 00:56:23 +0300 Subject: [PATCH 2/6] ensureQuorumSafeToDisruptHost now skips the barrier when there is a single host --- pkg/controller/chk/worker-raft-safety.go | 7 ++++++- pkg/controller/chk/worker-raft-safety_test.go | 10 ++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/pkg/controller/chk/worker-raft-safety.go b/pkg/controller/chk/worker-raft-safety.go index 80ba543d8..12bfff4c6 100644 --- a/pkg/controller/chk/worker-raft-safety.go +++ b/pkg/controller/chk/worker-raft-safety.go @@ -117,13 +117,18 @@ func hostContributesReady(host *api.Host) bool { // ensureQuorumSafeToDisruptHost refuses to take down a Ready host when the // remaining Ready members would fall below quorum. No-op when the ensemble -// already lacks live quorum (bootstrap / resume-from-stopped / recovery). +// already lacks live quorum (bootstrap / resume-from-stopped / recovery), or +// when there is only one host (restart is unavoidable — no sibling can hold +// quorum). func (w *worker) ensureQuorumSafeToDisruptHost(ctx context.Context, host *api.Host) error { cr := host.GetCR() if cr == nil { return nil } n := cr.HostsCount() + if n <= 1 { + return nil + } q := raftQuorumSize(n) ready := w.countReadyEnsembleMembers(ctx, cr) if ready < q { diff --git a/pkg/controller/chk/worker-raft-safety_test.go b/pkg/controller/chk/worker-raft-safety_test.go index 2134fff69..79c6b578d 100644 --- a/pkg/controller/chk/worker-raft-safety_test.go +++ b/pkg/controller/chk/worker-raft-safety_test.go @@ -117,6 +117,16 @@ func TestEnsureQuorumSafeToDisruptHost(t *testing.T) { require.Error(t, err) require.Contains(t, err.Error(), "would drop below Raft quorum") }) + + t.Run("allows disrupt of the sole host (n=1)", func(t *testing.T) { + solo := hostOnCR(chkWithHosts(1)) + solo.Runtime.CurStatefulSet = &apps.StatefulSet{} + solo.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 1 }, + } + require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, solo)) + }) } func TestChkStatefulSetFallbackAborts(t *testing.T) { From b5efadd2c75c0da6b069a1a145bd951c01645998 Mon Sep 17 00:00:00 2001 From: alz Date: Sat, 22 Aug 2026 03:08:13 +0300 Subject: [PATCH 3/6] Decide CHK Ready wait before host disrupt. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Capturing shouldWaitHostReady before force-restart avoids Started-only after ReadyReplicas drops to 0, which let 3→1 downscale complete while the survivor was still 0/1. Co-authored-by: Cursor --- pkg/controller/chk/worker-deleter.go | 6 +- pkg/controller/chk/worker-raft-safety.go | 44 ++++++++++--- pkg/controller/chk/worker-raft-safety_test.go | 61 +++++++++++++++---- pkg/controller/chk/worker-reconciler-chk.go | 29 +++++++-- .../chk/worker-reconciler-chk_test.go | 60 ++++++++---------- 5 files changed, 138 insertions(+), 62 deletions(-) diff --git a/pkg/controller/chk/worker-deleter.go b/pkg/controller/chk/worker-deleter.go index eef72d355..4fb76fc71 100644 --- a/pkg/controller/chk/worker-deleter.go +++ b/pkg/controller/chk/worker-deleter.go @@ -16,9 +16,10 @@ package chk import ( "context" + "time" + apiChk "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse-keeper.altinity.com/v1" "github.com/altinity/clickhouse-operator/pkg/controller" - "time" meta "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -51,6 +52,9 @@ func (w *worker) clean(ctx context.Context, cr api.ICustomResource) { objs.Subtract(need) w.a.V(1).M(cr).F().Info("List of non-reconciled objects:\n%s", objs) if w.purge(ctx, cr, objs, w.task.RegistryFailed()) > 0 { + // Give Raft time to notice removed peers before Completed. Survivors + // often flip /ready → 503 briefly after purge. + w.a.V(1).M(cr).F().Info("Purged non-reconciled objects; waiting 1m for membership to settle") util.WaitContextDoneOrTimeout(ctx, 1*time.Minute) } diff --git a/pkg/controller/chk/worker-raft-safety.go b/pkg/controller/chk/worker-raft-safety.go index 12bfff4c6..b7d81c2f1 100644 --- a/pkg/controller/chk/worker-raft-safety.go +++ b/pkg/controller/chk/worker-raft-safety.go @@ -147,28 +147,54 @@ func (w *worker) ensureQuorumSafeToDisruptHost(ctx context.Context, host *api.Ho return nil } +// shouldWaitHostReady decides whether STS reconcile must wait until Ready. +// +// Decision is taken BEFORE disrupting the host. After a force-restart, +// ReadyReplicas is 0 so ensembleHasLiveQuorum is false — re-checking then +// would wrongly fall into "bootstrap / Started-only" and complete while the +// pod is still 0/1 Running (seen after CHK 3→1 downscale). +// +// Rules: +// - n <= 1: always wait Ready (no siblings to deadlock on) +// - n > 1: wait Ready only when the ensemble already has live quorum +func (w *worker) shouldWaitHostReady(ctx context.Context, host *api.Host) bool { + if host == nil || host.GetCR() == nil { + return false + } + if host.GetCR().HostsCount() <= 1 { + return true + } + return w.ensembleHasLiveQuorum(ctx, host.GetCR()) +} + // prepareStsReconcileOptsWaitSection sets STS launch waits for Keeper. // -// Live quorum present: wait until Ready before moving on. -// No live quorum: wait Started only — Ready would deadlock until siblings exist -// (bootstrap, resume-from-stopped, or recovery). -func (w *worker) prepareStsReconcileOptsWaitSection(ctx context.Context, host *api.Host, opts *statefulset.ReconcileOptions) *statefulset.ReconcileOptions { +// waitReady must be computed via shouldWaitHostReady before any disruption. +// Live quorum / single-node: wait until Ready before moving on. +// No live quorum on multi-node: wait Started only — Ready would deadlock +// until siblings exist (bootstrap / resume-from-stopped / recovery). +func (w *worker) prepareStsReconcileOptsWaitSection( + ctx context.Context, + host *api.Host, + opts *statefulset.ReconcileOptions, + waitReady bool, +) *statefulset.ReconcileOptions { if opts == nil { opts = statefulset.NewReconcileStatefulSetOptions() } probes := host.GetCluster().GetReconcile().Host.Wait.Probes - liveQuorum := w.ensembleHasLiveQuorum(ctx, host.GetCR()) + _ = ctx - if probes.GetStartup().IsTrue() || !liveQuorum { + if probes.GetStartup().IsTrue() || !waitReady { opts = opts.SetWaitUntilStarted() w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilStarted") } switch { - case liveQuorum && !probes.GetReadiness().IsFalse(): + case waitReady && !probes.GetReadiness().IsFalse(): opts = opts.SetWaitUntilReady() - w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilReady (live Raft quorum)") - case !liveQuorum: + w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilReady (Keeper must become Ready)") + case !waitReady: w.a.V(1).M(host).F().Info("Skip WaitUntilReady — no live quorum (bootstrap / resume-from-stopped / recovery)") } diff --git a/pkg/controller/chk/worker-raft-safety_test.go b/pkg/controller/chk/worker-raft-safety_test.go index 79c6b578d..f2aade5fa 100644 --- a/pkg/controller/chk/worker-raft-safety_test.go +++ b/pkg/controller/chk/worker-raft-safety_test.go @@ -55,37 +55,72 @@ func TestEnsembleHasLiveQuorum(t *testing.T) { }) } -func TestPrepareStsReconcileOptsWaitSection(t *testing.T) { +func TestShouldWaitHostReady(t *testing.T) { ctx := context.Background() - t.Run("no live quorum skips Ready", func(t *testing.T) { + t.Run("single host always waits Ready even with 0 ReadyReplicas", func(t *testing.T) { w := &worker{ countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, } - host := hostOnCR(chkWithHosts(3)) - opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil) - require.True(t, opts.WaitUntilStarted()) - require.False(t, opts.WaitUntilReady()) + host := hostOnCR(chkWithHosts(1)) + require.True(t, w.shouldWaitHostReady(ctx, host)) }) - t.Run("live quorum waits Ready", func(t *testing.T) { + t.Run("multi-host without live quorum does not wait Ready", func(t *testing.T) { w := &worker{ - countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 3 }, + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, } host := hostOnCR(chkWithHosts(3)) - opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil) - require.True(t, opts.WaitUntilReady()) + require.False(t, w.shouldWaitHostReady(ctx, host)) }) - t.Run("live quorum can opt out of Ready", func(t *testing.T) { + t.Run("multi-host with live quorum waits Ready", func(t *testing.T) { w := &worker{ - countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 3 }, + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 2 }, } host := hostOnCR(chkWithHosts(3)) + require.True(t, w.shouldWaitHostReady(ctx, host)) + }) +} + +func TestPrepareStsReconcileOptsWaitSection(t *testing.T) { + ctx := context.Background() + + t.Run("no live quorum skips Ready", func(t *testing.T) { + w := &worker{} + host := hostOnCR(chkWithHosts(3)) + opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil, false) + require.True(t, opts.WaitUntilStarted()) + require.False(t, opts.WaitUntilReady()) + }) + + t.Run("waitReady waits Ready", func(t *testing.T) { + w := &worker{} + host := hostOnCR(chkWithHosts(3)) + opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil, true) + require.True(t, opts.WaitUntilReady()) + }) + + t.Run("waitReady can opt out of Ready probe", func(t *testing.T) { + w := &worker{} + host := hostOnCR(chkWithHosts(3)) host.GetCluster().GetReconcile().Host.Wait.Probes.Readiness = types.NewStringBool(false) - opts := w.prepareStsReconcileOptsWaitSection(ctx, host, statefulset.NewReconcileStatefulSetOptions()) + opts := w.prepareStsReconcileOptsWaitSection(ctx, host, statefulset.NewReconcileStatefulSetOptions(), true) require.False(t, opts.WaitUntilReady()) }) + + t.Run("single-host post-restart still waits Ready", func(t *testing.T) { + // Simulates ReadyReplicas=0 after force-restart: shouldWaitHostReady was + // true beforehand; prepareSts must honor that and not fall back to Started-only. + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, + } + host := hostOnCR(chkWithHosts(1)) + waitReady := w.shouldWaitHostReady(ctx, host) + require.True(t, waitReady) + opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil, waitReady) + require.True(t, opts.WaitUntilReady()) + }) } func TestEnsureQuorumSafeToDisruptHost(t *testing.T) { diff --git a/pkg/controller/chk/worker-reconciler-chk.go b/pkg/controller/chk/worker-reconciler-chk.go index 3ed94e88f..ee08a39bd 100644 --- a/pkg/controller/chk/worker-reconciler-chk.go +++ b/pkg/controller/chk/worker-reconciler-chk.go @@ -288,17 +288,31 @@ func (w *worker) reconcileCRAuxObjectsPreliminaryDomain(ctx context.Context, cr // best-effort pacing only; the function returns early on ctx cancellation. // Same-size reconciles do not wait (see #2035 / #2059) — a fixed 10s sleep // previously left healthy ensembles cycling and delayed Completed. - d := keeperMembershipSettleDelay(cr.HostsCount(), cr.GetAncestor().HostsCount()) + d := w.membershipSettleDelay(cr) if d == 0 { return nil } + w.a.V(1).M(cr).Info("Waiting %s for Keeper Raft membership to settle", d) util.WaitContextDoneOrTimeout(ctx, d) return nil } -// keeperMembershipSettleDelay is a best-effort pause after publishing membership -// changes so Raft can settle. Same host count → no delay. -func keeperMembershipSettleDelay(currentHosts, ancestorHosts int) time.Duration { +// membershipSettleDelay is a best-effort pause after publishing membership +// changes so Raft can settle: +// - same host count → no delay +// - upscale → 30s +// - downscale → 120s (survivors still need time after raft_configuration shrink; +// peer purge later adds another 1m in clean()) +func (w *worker) membershipSettleDelay(cr *apiChk.ClickHouseKeeperInstallation) time.Duration { + if cr == nil { + return 0 + } + ancestorHosts := 0 + if ancestor := cr.GetAncestor(); ancestor != nil { + ancestorHosts = ancestor.HostsCount() + } + currentHosts := cr.HostsCount() + switch { case currentHosts < ancestorHosts: return 120 * time.Second @@ -427,6 +441,11 @@ func (w *worker) reconcileHostStatefulSet(ctx context.Context, host *api.Host, o w.a.V(1).M(host).F().Info("Reconcile host STS: %s. App version: %s", host.GetName(), host.Runtime.Version.Render()) + // Decide Ready vs Started-only wait before any disruption. Force-restart + // drops ReadyReplicas to 0; re-evaluating live quorum afterward would skip + // WaitUntilReady and let reconcile complete while the pod is still 0/1. + waitReady := w.shouldWaitHostReady(ctx, host) + // Start with force-restart host forcedRestart := false if w.shouldForceRestartHost(ctx, host) { @@ -460,7 +479,7 @@ func (w *worker) reconcileHostStatefulSet(ctx context.Context, host *api.Host, o return err } } - opts = w.prepareStsReconcileOptsWaitSection(ctx, host, opts) + opts = w.prepareStsReconcileOptsWaitSection(ctx, host, opts, waitReady) // We are in place, where we can reconcile StatefulSet to desired configuration. w.a.V(1).M(host).F().Info("Reconcile host STS: %s. Reconcile StatefulSet", host.GetName()) diff --git a/pkg/controller/chk/worker-reconciler-chk_test.go b/pkg/controller/chk/worker-reconciler-chk_test.go index bddbc8818..73e491c27 100644 --- a/pkg/controller/chk/worker-reconciler-chk_test.go +++ b/pkg/controller/chk/worker-reconciler-chk_test.go @@ -62,40 +62,32 @@ func newStatusTestWorker(statusWriter interfaces.IKubeCR) *worker { } } -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 TestMembershipSettleDelay(t *testing.T) { + w := &worker{a: a.NewAnnouncer(nil, nil)} + + t.Run("same size does not wait", func(t *testing.T) { + cr := chkWithHosts(3) + cr.SetAncestor(chkWithHosts(3)) + if got := w.membershipSettleDelay(cr); got != 0 { + t.Fatalf("membershipSettleDelay() = %s, want 0", got) + } + }) + + t.Run("upscale waits for raft membership", func(t *testing.T) { + cr := chkWithHosts(3) + cr.SetAncestor(chkWithHosts(1)) + if got := w.membershipSettleDelay(cr); got != 30*time.Second { + t.Fatalf("membershipSettleDelay() = %s, want 30s", got) + } + }) + + t.Run("downscale always waits 120s", func(t *testing.T) { + cr := chkWithHosts(1) + cr.SetAncestor(chkWithHosts(3)) + if got := w.membershipSettleDelay(cr); got != 120*time.Second { + t.Fatalf("membershipSettleDelay() = %s, want 120s", got) + } + }) } func TestPersistReconcileCompleted(t *testing.T) { From 3a6c9a1e53a69c0419396d8007aa2a0cd26d8a3d Mon Sep 17 00:00:00 2001 From: alz Date: Sat, 22 Aug 2026 13:41:08 +0300 Subject: [PATCH 4/6] CHK: defer quorum-unsafe disrupts and align raft safety with CHI (#2069) Replace abort-on-quorum-refuse with a wait-then-defer flow, recovery-first host ordering, and a CHI-style single late disrupt gate with an early ensemble snapshot. Add test_020003_3 for interrupted Keeper rolls. Co-authored-by: Cursor --- .../v1/type_status.go | 1 + .../clickhouse.altinity.com/v1/type_status.go | 3 + pkg/controller/chk/worker-raft-safety.go | 244 ++++++++++------- pkg/controller/chk/worker-raft-safety_test.go | 256 ++++++++++++------ pkg/controller/chk/worker-reconciler-chk.go | 158 ++++++++--- .../chk/worker-reconciler-chk_test.go | 43 +++ .../chk/worker-reconciler-helper.go | 16 +- pkg/controller/chk/worker.go | 5 + .../e2e/manifests/chk/test-020003-3-chi.yaml | 16 ++ .../manifests/chk/test-020003-3-chk-1.yaml | 20 ++ .../manifests/chk/test-020003-3-chk-2.yaml | 20 ++ .../manifests/chk/test-020003-3-chk-3.yaml | 20 ++ tests/e2e/test_operator.py | 154 +++++++++++ 13 files changed, 743 insertions(+), 213 deletions(-) create mode 100644 tests/e2e/manifests/chk/test-020003-3-chi.yaml create mode 100644 tests/e2e/manifests/chk/test-020003-3-chk-1.yaml create mode 100644 tests/e2e/manifests/chk/test-020003-3-chk-2.yaml create mode 100644 tests/e2e/manifests/chk/test-020003-3-chk-3.yaml diff --git a/pkg/apis/clickhouse-keeper.altinity.com/v1/type_status.go b/pkg/apis/clickhouse-keeper.altinity.com/v1/type_status.go index 937128d88..f630410c8 100644 --- a/pkg/apis/clickhouse-keeper.altinity.com/v1/type_status.go +++ b/pkg/apis/clickhouse-keeper.altinity.com/v1/type_status.go @@ -48,6 +48,7 @@ const ( StatusReasonFIPSValidationFailed = chi.StatusReasonFIPSValidationFailed StatusReasonFIPSImagePolicyViolation = chi.StatusReasonFIPSImagePolicyViolation StatusReasonNoKeeperListener = chi.StatusReasonNoKeeperListener + StatusReasonRaftQuorumUnsafe = chi.StatusReasonRaftQuorumUnsafe ) // Status defines status section of the custom resource. diff --git a/pkg/apis/clickhouse.altinity.com/v1/type_status.go b/pkg/apis/clickhouse.altinity.com/v1/type_status.go index ef7d96cc0..776bbed75 100644 --- a/pkg/apis/clickhouse.altinity.com/v1/type_status.go +++ b/pkg/apis/clickhouse.altinity.com/v1/type_status.go @@ -69,6 +69,9 @@ const ( // the shard left serving. Recoverable without a spec edit - the peer coming back triggers // a retry - so it is deliberately absent from normalizeTimeAbortReasons. StatusReasonShardHasNoHealthyPeer = "ShardHasNoHealthyPeer" + // StatusReasonRaftQuorumUnsafe: a CHK host's reconcile would disrupt a Ready replica while + // the ensemble lacks Raft quorum headroom. Disruption is deferred until siblings recover. + StatusReasonRaftQuorumUnsafe = "RaftQuorumUnsafe" // StatusReasonRemovedSecretRefSyntax: a user settings field uses the `k8s_secret_` or // `k8s_secret_env_` prefix, removed in 0.27.4 because it accepted a namespace/name/key // triple and could therefore read a Secret from any namespace. Aborts rather than diff --git a/pkg/controller/chk/worker-raft-safety.go b/pkg/controller/chk/worker-raft-safety.go index b7d81c2f1..704ba1574 100644 --- a/pkg/controller/chk/worker-raft-safety.go +++ b/pkg/controller/chk/worker-raft-safety.go @@ -17,28 +17,42 @@ package chk import ( "context" "fmt" + "time" apps "k8s.io/api/apps/v1" api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1" + "github.com/altinity/clickhouse-operator/pkg/apis/common/types" "github.com/altinity/clickhouse-operator/pkg/controller/common" + a "github.com/altinity/clickhouse-operator/pkg/controller/common/announcer" "github.com/altinity/clickhouse-operator/pkg/controller/common/statefulset" "github.com/altinity/clickhouse-operator/pkg/interfaces" + "github.com/altinity/clickhouse-operator/pkg/util" ) // Raft / ensemble safety for CHK (#2069). // -// Live Ready count decides the mode — not CR ancestor / host inventory: +// snapshotHostEnsemble records rolling vs bootstrap once per host, before any +// disruption. Live Ready count drives rolling — not CR ancestor / host inventory: // -// - Live quorum (Ready members >= majority): rolling recreate of a healthy -// ensemble — wait Ready before the next host, and refuse to disrupt a host -// if doing so would drop below quorum. -// - No live quorum (fresh install, resume-from-stopped, or already broken): -// bootstrap / recovery — wait Started only so siblings can come up together. +// - Rolling (n<=1 or live quorum): wait Ready, refuse disrupt below quorum. +// - Bootstrap: wait Started only so siblings can come up together. // -// Hooks below stay thin so a fuller Raft-membership barrier (committed -// /keeper/config + mntr, as in PR #2041) can replace verifyHostEnsembleMembership -// without reshaping the reconcile loop. +// verifyHostEnsembleMembership is the extension point for a fuller Raft barrier +// (committed /keeper/config + mntr, as in PR #2041). + +const ( + defaultQuorumDisruptPollInterval = 5 * time.Second + defaultQuorumDisruptWaitTimeout = 2 * time.Minute +) + +// hostEnsembleSnapshot captures ensemble state before any host disruption. +// rolling must not be re-derived after force-restart — ReadyReplicas drops to 0. +type hostEnsembleSnapshot struct { + rolling bool + members int + readyCount int +} // chkStatefulSetFallback aborts the reconcile on STS create/update wait failure. // DefaultFallback returns ErrCRUDIgnore, which lets the host loop recreate the @@ -96,124 +110,170 @@ func (w *worker) countReadyEnsembleMembers(ctx context.Context, cr api.ICustomRe return ready } -// ensembleHasLiveQuorum reports whether enough hosts are Ready to form a Raft -// majority. Resume-from-stopped and fresh bootstrap both yield false (0 Ready) -// and therefore take the fast startup path. -func (w *worker) ensembleHasLiveQuorum(ctx context.Context, cr api.ICustomResource) bool { - if cr == nil { - return false +// snapshotHostEnsemble records rolling vs bootstrap before disrupting a host. +func (w *worker) snapshotHostEnsemble(ctx context.Context, host *api.Host) hostEnsembleSnapshot { + if host == nil || host.GetCR() == nil { + return hostEnsembleSnapshot{} } + cr := host.GetCR() n := cr.HostsCount() - return w.countReadyEnsembleMembers(ctx, cr) >= raftQuorumSize(n) + ready := w.countReadyEnsembleMembers(ctx, cr) + return hostEnsembleSnapshot{ + rolling: n <= 1 || ready >= raftQuorumSize(n), + members: n, + readyCount: ready, + } } -// hostContributesReady reports whether this host currently counts toward live quorum. -func hostContributesReady(host *api.Host) bool { - if host == nil || host.Runtime.CurStatefulSet == nil { - return false +// refreshQuorumSnapshotCounts updates live Ready counts for an in-flight wait. +// rolling is intentionally frozen — it was captured before any disruption. +func (w *worker) refreshQuorumSnapshotCounts(ctx context.Context, host *api.Host, snap *hostEnsembleSnapshot) { + if host == nil || snap == nil || !snap.rolling { + return + } + if w.c != nil { + host.Runtime.CurStatefulSet, _ = w.c.kube.STS().Get(ctx, host) + } + if cr := host.GetCR(); cr != nil { + snap.readyCount = w.countReadyEnsembleMembers(ctx, cr) } - return host.Runtime.CurStatefulSet.Status.ReadyReplicas > 0 } -// ensureQuorumSafeToDisruptHost refuses to take down a Ready host when the -// remaining Ready members would fall below quorum. No-op when the ensemble -// already lacks live quorum (bootstrap / resume-from-stopped / recovery), or -// when there is only one host (restart is unavoidable — no sibling can hold -// quorum). -func (w *worker) ensureQuorumSafeToDisruptHost(ctx context.Context, host *api.Host) error { - cr := host.GetCR() - if cr == nil { - return nil +func (w *worker) quorumDisruptPollInterval() time.Duration { + if w.quorumDisruptPollOverride > 0 { + return w.quorumDisruptPollOverride } - n := cr.HostsCount() - if n <= 1 { + return defaultQuorumDisruptPollInterval +} + +func (w *worker) quorumDisruptWaitTimeout() time.Duration { + if w.quorumDisruptWaitOverride > 0 { + return w.quorumDisruptWaitOverride + } + return defaultQuorumDisruptWaitTimeout +} + +// waitForQuorumSafeToDisruptHost polls until disrupting the host would not drop +// the ensemble below Raft quorum, or until the wait budget expires. +func (w *worker) waitForQuorumSafeToDisruptHost( + ctx context.Context, + host *api.Host, + opts *statefulset.ReconcileOptions, + snap *hostEnsembleSnapshot, +) error { + if snap == nil || !snap.rolling || snap.members <= 1 { return nil } - q := raftQuorumSize(n) - ready := w.countReadyEnsembleMembers(ctx, cr) - if ready < q { + if !w.hostDisruptionWouldBreakQuorum(ctx, host, opts, *snap) { return nil } - remaining := ready - if hostContributesReady(host) { - remaining-- + + w.a.V(1).M(host).F().Info( + "Waiting for Raft quorum headroom before disrupting host %s (ready=%d quorum=%d)", + host.GetName(), snap.readyCount, raftQuorumSize(snap.members), + ) + + deadline := time.Now().Add(w.quorumDisruptWaitTimeout()) + for time.Now().Before(deadline) { + if util.WaitContextDoneOrTimeout(ctx, w.quorumDisruptPollInterval()) { + return ctx.Err() + } + w.refreshQuorumSnapshotCounts(ctx, host, snap) + if !w.hostDisruptionWouldBreakQuorum(ctx, host, opts, *snap) { + w.a.V(1).M(host).F().Info( + "Raft quorum headroom available — proceeding with host %s disruption (ready=%d)", + host.GetName(), snap.readyCount, + ) + return nil + } } - if remaining < q { - return fmt.Errorf( - "refusing to disrupt host %s: ready=%d remaining=%d quorum=%d (would drop below Raft quorum)", - host.GetName(), ready, remaining, q, + + w.a.V(1).M(host).F(). + WithEvent(host.GetCR(), a.EventActionReconcile, a.EventReasonHostReconcileDeferredShardSafety). + Warning( + "Deferring host StatefulSet reconcile: disrupting %s would drop below Raft quorum (%s)", + host.GetName(), quorumDisruptDeferMessage(host, *snap), ) + return common.ErrCRUDDeferred +} + +// isHostHealthyForReconcile is true when the host counts as live for recovery-first +// ordering and quorum headroom. Stopped/troubleshoot hosts are intentionally +// unavailable and are ordered after recovery hosts (CHI #1704). +func (w *worker) isHostHealthyForReconcile(ctx context.Context, host *api.Host) bool { + if host == nil { + return false } - return nil + if host.IsStopped() || host.IsTroubleshoot() { + return true + } + sts := host.Runtime.CurStatefulSet + if sts == nil && w.c != nil { + sts, _ = w.c.kube.STS().Get(ctx, host) + } + return sts != nil && sts.Status.ReadyReplicas > 0 } -// shouldWaitHostReady decides whether STS reconcile must wait until Ready. -// -// Decision is taken BEFORE disrupting the host. After a force-restart, -// ReadyReplicas is 0 so ensembleHasLiveQuorum is false — re-checking then -// would wrongly fall into "bootstrap / Started-only" and complete while the -// pod is still 0/1 Running (seen after CHK 3→1 downscale). -// -// Rules: -// - n <= 1: always wait Ready (no siblings to deadlock on) -// - n > 1: wait Ready only when the ensemble already has live quorum -func (w *worker) shouldWaitHostReady(ctx context.Context, host *api.Host) bool { - if host == nil || host.GetCR() == nil { +// hostContributesReady reports whether this host currently counts toward live quorum. +func hostContributesReady(host *api.Host) bool { + if host == nil || host.Runtime.CurStatefulSet == nil { return false } - if host.GetCR().HostsCount() <= 1 { + return host.Runtime.CurStatefulSet.Status.ReadyReplicas > 0 +} + +// ensembleQuorumSafeAfterDisrupt reports whether remaining Ready members would still +// meet quorum if this host were disrupted. Pure — snap counts are frozen before disrupt. +func ensembleQuorumSafeAfterDisrupt(snap hostEnsembleSnapshot, host *api.Host) bool { + if !snap.rolling || snap.members <= 1 { return true } - return w.ensembleHasLiveQuorum(ctx, host.GetCR()) + remaining := snap.readyCount + if hostContributesReady(host) { + remaining-- + } + return remaining >= raftQuorumSize(snap.members) } -// prepareStsReconcileOptsWaitSection sets STS launch waits for Keeper. +// hostDisruptionWouldBreakQuorum is true when this pass would disrupt a Ready host and +// drop the ensemble below Raft quorum (#2069). // -// waitReady must be computed via shouldWaitHostReady before any disruption. -// Live quorum / single-node: wait until Ready before moving on. -// No live quorum on multi-node: wait Started only — Ready would deadlock -// until siblings exist (bootstrap / resume-from-stopped / recovery). -func (w *worker) prepareStsReconcileOptsWaitSection( +// Must be called after PrepareHostStatefulSetWithStatus — ObjectStatusSame is assigned only there. +func (w *worker) hostDisruptionWouldBreakQuorum( ctx context.Context, host *api.Host, opts *statefulset.ReconcileOptions, - waitReady bool, -) *statefulset.ReconcileOptions { - if opts == nil { - opts = statefulset.NewReconcileStatefulSetOptions() + snap hostEnsembleSnapshot, +) bool { + if host == nil || host.IsStopped() { + return false } - probes := host.GetCluster().GetReconcile().Host.Wait.Probes - _ = ctx - - if probes.GetStartup().IsTrue() || !waitReady { - opts = opts.SetWaitUntilStarted() - w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilStarted") + if host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) { + return false } - - switch { - case waitReady && !probes.GetReadiness().IsFalse(): - opts = opts.SetWaitUntilReady() - w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilReady (Keeper must become Ready)") - case !waitReady: - w.a.V(1).M(host).F().Info("Skip WaitUntilReady — no live quorum (bootstrap / resume-from-stopped / recovery)") + willDisrupt := !host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusSame) || + w.shouldForceRestartHost(ctx, host) || + (opts != nil && opts.ForceRecreate()) + if !willDisrupt { + return false } - - return opts + return hostContributesReady(host) && !ensembleQuorumSafeAfterDisrupt(snap, host) } -// waitHostRaftJoined is the post-host hook for confirming the replica is part of -// the live ensemble before the reconciler advances. -// -// Today a no-op beyond what STS Ready wait already enforced. Replace or extend -// verifyHostEnsembleMembership with committed Raft membership checks as in -// Altinity/clickhouse-operator#2041. -func (w *worker) waitHostRaftJoined(ctx context.Context, host *api.Host) error { - return w.verifyHostEnsembleMembership(ctx, host) +func quorumDisruptDeferMessage(host *api.Host, snap hostEnsembleSnapshot) string { + remaining := snap.readyCount + if hostContributesReady(host) { + remaining-- + } + return fmt.Sprintf( + "ready=%d remaining=%d quorum=%d", + snap.readyCount, remaining, raftQuorumSize(snap.members), + ) } // verifyHostEnsembleMembership is the extension point for Raft membership -// verification. Currently a no-op: STS Ready wait already ran when live quorum -// mode applied. Implement committed-config / leader sync barriers here when +// verification after a host joins in rolling mode. Currently a no-op: STS Ready +// wait already ran. Implement committed-config / leader sync barriers here when // adopting the fuller rescale design from PR #2041. func (w *worker) verifyHostEnsembleMembership(ctx context.Context, host *api.Host) error { _ = ctx diff --git a/pkg/controller/chk/worker-raft-safety_test.go b/pkg/controller/chk/worker-raft-safety_test.go index f2aade5fa..131daab27 100644 --- a/pkg/controller/chk/worker-raft-safety_test.go +++ b/pkg/controller/chk/worker-raft-safety_test.go @@ -16,7 +16,9 @@ package chk import ( "context" + "errors" "testing" + "time" "github.com/stretchr/testify/require" apps "k8s.io/api/apps/v1" @@ -35,139 +37,227 @@ func TestRaftQuorumSize(t *testing.T) { require.Equal(t, 3, raftQuorumSize(5)) } -func TestEnsembleHasLiveQuorum(t *testing.T) { - cr := chkWithHosts(3) - w := &worker{} - - t.Run("no ready members — bootstrap / resume-from-stopped", func(t *testing.T) { - w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 0 } - require.False(t, w.ensembleHasLiveQuorum(context.Background(), cr)) - }) - - t.Run("below quorum", func(t *testing.T) { - w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 1 } - require.False(t, w.ensembleHasLiveQuorum(context.Background(), cr)) - }) - - t.Run("at quorum", func(t *testing.T) { - w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 2 } - require.True(t, w.ensembleHasLiveQuorum(context.Background(), cr)) - }) -} - -func TestShouldWaitHostReady(t *testing.T) { +func TestSnapshotHostEnsemble(t *testing.T) { ctx := context.Background() - t.Run("single host always waits Ready even with 0 ReadyReplicas", func(t *testing.T) { + t.Run("single host is rolling even with 0 ReadyReplicas", func(t *testing.T) { w := &worker{ countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, } host := hostOnCR(chkWithHosts(1)) - require.True(t, w.shouldWaitHostReady(ctx, host)) + snap := w.snapshotHostEnsemble(ctx, host) + require.True(t, snap.rolling) + require.Equal(t, 1, snap.members) + require.Equal(t, 0, snap.readyCount) }) - t.Run("multi-host without live quorum does not wait Ready", func(t *testing.T) { + t.Run("multi-host without live quorum is bootstrap", func(t *testing.T) { w := &worker{ countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, } host := hostOnCR(chkWithHosts(3)) - require.False(t, w.shouldWaitHostReady(ctx, host)) + snap := w.snapshotHostEnsemble(ctx, host) + require.False(t, snap.rolling) + require.Equal(t, 3, snap.members) + require.Equal(t, 0, snap.readyCount) }) - t.Run("multi-host with live quorum waits Ready", func(t *testing.T) { + t.Run("multi-host below quorum is bootstrap", func(t *testing.T) { + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 1 }, + } + host := hostOnCR(chkWithHosts(3)) + snap := w.snapshotHostEnsemble(ctx, host) + require.False(t, snap.rolling) + }) + + t.Run("multi-host at quorum is rolling", func(t *testing.T) { w := &worker{ countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 2 }, } host := hostOnCR(chkWithHosts(3)) - require.True(t, w.shouldWaitHostReady(ctx, host)) + snap := w.snapshotHostEnsemble(ctx, host) + require.True(t, snap.rolling) + require.Equal(t, 2, snap.readyCount) }) } -func TestPrepareStsReconcileOptsWaitSection(t *testing.T) { +func TestEnsembleQuorumSafeAfterDisrupt(t *testing.T) { + host := hostOnCR(chkWithHosts(3)) + host.Runtime.CurStatefulSet = &apps.StatefulSet{} + host.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + + t.Run("bootstrap mode is always safe", func(t *testing.T) { + snap := hostEnsembleSnapshot{rolling: false, members: 3, readyCount: 0} + require.True(t, ensembleQuorumSafeAfterDisrupt(snap, host)) + }) + + t.Run("safe when siblings keep quorum", func(t *testing.T) { + snap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 3} + require.True(t, ensembleQuorumSafeAfterDisrupt(snap, host)) + }) + + t.Run("unsafe when remaining would be below quorum", func(t *testing.T) { + snap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 2} + require.False(t, ensembleQuorumSafeAfterDisrupt(snap, host)) + }) + + t.Run("sole host is always safe", func(t *testing.T) { + solo := hostOnCR(chkWithHosts(1)) + solo.Runtime.CurStatefulSet = &apps.StatefulSet{} + solo.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + snap := hostEnsembleSnapshot{rolling: true, members: 1, readyCount: 1} + require.True(t, ensembleQuorumSafeAfterDisrupt(snap, solo)) + }) +} + +func TestHostDisruptionWouldBreakQuorum(t *testing.T) { ctx := context.Background() + w := &worker{} + host := hostOnCR(chkWithHosts(3)) + host.Runtime.CurStatefulSet = &apps.StatefulSet{} + host.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + host.GetReconcileAttributes().SetStatus(types.ObjectStatusModified) + snap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 2} - t.Run("no live quorum skips Ready", func(t *testing.T) { - w := &worker{} - host := hostOnCR(chkWithHosts(3)) - opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil, false) - require.True(t, opts.WaitUntilStarted()) - require.False(t, opts.WaitUntilReady()) + t.Run("no-op for new host", func(t *testing.T) { + newHost := hostOnCR(chkWithHosts(3)) + newHost.GetReconcileAttributes().SetStatus(types.ObjectStatusRequested) + require.False(t, w.hostDisruptionWouldBreakQuorum(ctx, newHost, nil, snap)) }) - t.Run("waitReady waits Ready", func(t *testing.T) { - w := &worker{} - host := hostOnCR(chkWithHosts(3)) - opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil, true) - require.True(t, opts.WaitUntilReady()) + t.Run("no-op when STS is unchanged", func(t *testing.T) { + same := hostOnCR(chkWithHosts(3)) + same.Runtime.CurStatefulSet = &apps.StatefulSet{} + same.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + same.GetReconcileAttributes().SetStatus(types.ObjectStatusSame) + require.False(t, w.hostDisruptionWouldBreakQuorum(ctx, same, nil, snap)) }) - t.Run("waitReady can opt out of Ready probe", func(t *testing.T) { - w := &worker{} - host := hostOnCR(chkWithHosts(3)) - host.GetCluster().GetReconcile().Host.Wait.Probes.Readiness = types.NewStringBool(false) - opts := w.prepareStsReconcileOptsWaitSection(ctx, host, statefulset.NewReconcileStatefulSetOptions(), true) - require.False(t, opts.WaitUntilReady()) + t.Run("blocks disruptive roll without quorum headroom", func(t *testing.T) { + require.True(t, w.hostDisruptionWouldBreakQuorum(ctx, host, nil, snap)) }) - t.Run("single-host post-restart still waits Ready", func(t *testing.T) { - // Simulates ReadyReplicas=0 after force-restart: shouldWaitHostReady was - // true beforehand; prepareSts must honor that and not fall back to Started-only. - w := &worker{ - countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, - } - host := hostOnCR(chkWithHosts(1)) - waitReady := w.shouldWaitHostReady(ctx, host) - require.True(t, waitReady) - opts := w.prepareStsReconcileOptsWaitSection(ctx, host, nil, waitReady) - require.True(t, opts.WaitUntilReady()) + t.Run("allows disruptive roll when siblings keep quorum", func(t *testing.T) { + bigSnap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 3} + require.False(t, w.hostDisruptionWouldBreakQuorum(ctx, host, nil, bigSnap)) + }) + + t.Run("force recreate counts as disruptive", func(t *testing.T) { + same := hostOnCR(chkWithHosts(3)) + same.Runtime.CurStatefulSet = &apps.StatefulSet{} + same.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + same.GetReconcileAttributes().SetStatus(types.ObjectStatusSame) + opts := statefulset.NewReconcileStatefulSetOptions().SetForceRecreate() + require.True(t, w.hostDisruptionWouldBreakQuorum(ctx, same, opts, snap)) }) } -func TestEnsureQuorumSafeToDisruptHost(t *testing.T) { +func TestChkStatefulSetFallbackAborts(t *testing.T) { + f := newChkStatefulSetFallback() + require.Equal(t, common.ErrCRUDAbort, f.OnStatefulSetCreateFailed(nil, nil)) + require.Equal(t, common.ErrCRUDAbort, f.OnStatefulSetUpdateFailed(nil, nil, nil, nil)) +} + +func TestErrCRUDDeferredIsDistinctFromAbort(t *testing.T) { + require.False(t, errors.Is(common.ErrCRUDDeferred, common.ErrCRUDAbort)) + require.False(t, errors.Is(common.ErrCRUDAbort, common.ErrCRUDDeferred)) +} + +func TestWaitForQuorumSafeToDisruptHost(t *testing.T) { ctx := context.Background() - cr := chkWithHosts(3) - host := hostOnCR(cr) + host := hostOnCR(chkWithHosts(3)) host.Runtime.CurStatefulSet = &apps.StatefulSet{} host.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + host.GetReconcileAttributes().SetStatus(types.ObjectStatusModified) + snap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 2} - t.Run("allows disrupt when no live quorum", func(t *testing.T) { - w := &worker{ - countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, - } - require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, host)) + t.Run("returns immediately when already safe", func(t *testing.T) { + w := &worker{} + safeSnap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 3} + require.NoError(t, w.waitForQuorumSafeToDisruptHost(ctx, host, nil, &safeSnap)) }) - t.Run("allows disrupt when siblings keep quorum", func(t *testing.T) { + t.Run("waits until ready count increases", func(t *testing.T) { + ready := 2 w := &worker{ - countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 3 }, + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { + return ready + }, + quorumDisruptPollOverride: 5 * time.Millisecond, + quorumDisruptWaitOverride: 200 * time.Millisecond, } - require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, host)) + waitSnap := snap + done := make(chan struct{}) + go func() { + time.Sleep(20 * time.Millisecond) + ready = 3 + close(done) + }() + require.NoError(t, w.waitForQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap)) + <-done }) - t.Run("refuses disrupt when remaining would be below quorum", func(t *testing.T) { + t.Run("defers after wait budget expires", func(t *testing.T) { w := &worker{ countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 2 }, + quorumDisruptPollOverride: 5 * time.Millisecond, + quorumDisruptWaitOverride: 20 * time.Millisecond, } - err := w.ensureQuorumSafeToDisruptHost(ctx, host) - require.Error(t, err) - require.Contains(t, err.Error(), "would drop below Raft quorum") + waitSnap := snap + err := w.waitForQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap) + require.ErrorIs(t, err, common.ErrCRUDDeferred) }) +} - t.Run("allows disrupt of the sole host (n=1)", func(t *testing.T) { - solo := hostOnCR(chkWithHosts(1)) - solo.Runtime.CurStatefulSet = &apps.StatefulSet{} - solo.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 - w := &worker{ - countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 1 }, - } - require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, solo)) +func TestIsHostHealthyForReconcile(t *testing.T) { + ctx := context.Background() + w := &worker{} + + t.Run("nil host", func(t *testing.T) { + require.False(t, w.isHostHealthyForReconcile(ctx, nil)) + }) + + t.Run("stopped counts as healthy for ordering", func(t *testing.T) { + cr := chkWithHosts(1) + cr.Spec.Stop = types.NewStringBool(true) + host := hostOnCR(cr) + require.True(t, w.isHostHealthyForReconcile(ctx, host)) + }) + + t.Run("ready STS counts as healthy", func(t *testing.T) { + host := hostOnCR(chkWithHosts(1)) + host.Runtime.CurStatefulSet = &apps.StatefulSet{} + host.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + require.True(t, w.isHostHealthyForReconcile(ctx, host)) + }) + + t.Run("not ready STS is recovery", func(t *testing.T) { + host := hostOnCR(chkWithHosts(1)) + host.Runtime.CurStatefulSet = &apps.StatefulSet{} + require.False(t, w.isHostHealthyForReconcile(ctx, host)) }) } -func TestChkStatefulSetFallbackAborts(t *testing.T) { - f := newChkStatefulSetFallback() - require.Equal(t, common.ErrCRUDAbort, f.OnStatefulSetCreateFailed(nil, nil)) - require.Equal(t, common.ErrCRUDAbort, f.OnStatefulSetUpdateFailed(nil, nil, nil, nil)) +func TestShardHostsRecoveryFirst(t *testing.T) { + cr := chkWithHosts(2) + shard := cr.Spec.Configuration.Clusters[0].Layout.Shards[0] + h0 := shard.Hosts[0] + h1 := shard.Hosts[1] + h0.SetCR(cr) + h1.SetCR(cr) + + h0.Runtime.CurStatefulSet = &apps.StatefulSet{} + h0.Runtime.CurStatefulSet.Status.ReadyReplicas = 1 + h1.Runtime.CurStatefulSet = &apps.StatefulSet{} + + healthy := func(host *api.Host) bool { + return host.Runtime.CurStatefulSet != nil && host.Runtime.CurStatefulSet.Status.ReadyReplicas > 0 + } + ordered := shardHostsRecoveryFirst(shard, healthy) + require.Len(t, ordered, 2) + require.Same(t, h1, ordered[0], "not-ready host should reconcile first") + require.Same(t, h0, ordered[1], "ready host should reconcile second") } func chkWithHosts(n int) *apiChk.ClickHouseKeeperInstallation { diff --git a/pkg/controller/chk/worker-reconciler-chk.go b/pkg/controller/chk/worker-reconciler-chk.go index ee08a39bd..398ede8ae 100644 --- a/pkg/controller/chk/worker-reconciler-chk.go +++ b/pkg/controller/chk/worker-reconciler-chk.go @@ -123,6 +123,21 @@ func (w *worker) reconcileCR(ctx context.Context, old, new *apiChk.ClickHouseKee w.setHostStatusesPreliminary(ctx, new) if err := w.reconcile(ctx, new); err != nil { + if errors.Is(err, common.ErrCRUDDeferred) { + w.a.V(1).M(new).F().Info("Reconcile deferred — waiting for Raft quorum headroom") + new.EnsureStatus().PushError(fmt.Sprintf( + "[%s] deferred host disruption until Raft quorum headroom is available", + apiChk.StatusReasonRaftQuorumUnsafe, + )) + _ = w.c.updateCRObjectStatus(ctx, new, types.UpdateStatusOptions{ + CopyStatusOptions: types.CopyStatusOptions{ + CopyStatusFieldGroup: types.CopyStatusFieldGroup{ + FieldGroupMain: true, + }, + }, + }) + return err + } // Something went wrong w.a.WithEvent(new, a.EventActionReconcile, a.EventReasonReconcileFailed). WithError(new). @@ -435,30 +450,35 @@ func (w *worker) reconcileConfigMapHost(ctx context.Context, host *api.Host) err } // reconcileHostStatefulSet reconciles host's StatefulSet -func (w *worker) reconcileHostStatefulSet(ctx context.Context, host *api.Host, opts *statefulset.ReconcileOptions) error { +func (w *worker) reconcileHostStatefulSet( + ctx context.Context, + host *api.Host, + opts *statefulset.ReconcileOptions, + snap hostEnsembleSnapshot, +) error { log.V(1).M(host).F().S().Info("reconcile StatefulSet start") defer log.V(1).M(host).F().E().Info("reconcile StatefulSet end") w.a.V(1).M(host).F().Info("Reconcile host STS: %s. App version: %s", host.GetName(), host.Runtime.Version.Render()) - // Decide Ready vs Started-only wait before any disruption. Force-restart - // drops ReadyReplicas to 0; re-evaluating live quorum afterward would skip - // WaitUntilReady and let reconcile complete while the pod is still 0/1. - waitReady := w.shouldWaitHostReady(ctx, host) + w.stsReconciler.PrepareHostStatefulSetWithStatus(ctx, host, host.IsStopped()) + opts = w.prepareStsReconcileOptsWaitSection(host, opts, snap.rolling) - // Start with force-restart host - forcedRestart := false - if w.shouldForceRestartHost(ctx, host) { - if err := w.ensureQuorumSafeToDisruptHost(ctx, host); err != nil { - w.a.V(1).M(host).F().Error("%v", err) + // First point where "would this pass take the host down?" can be answered — same + // placement as CHI hostDisruptionWouldDegradeShard (#1704). + if w.hostDisruptionWouldBreakQuorum(ctx, host, opts, snap) { + if err := w.waitForQuorumSafeToDisruptHost(ctx, host, opts, &snap); err != nil { return err } + } + + forcedRestart := false + if w.shouldForceRestartHost(ctx, host) { w.a.V(1).M(host).F().Info("Reconcile host STS force restart: %s", host.GetName()) _ = w.hostForceRestart(ctx, host, opts) forcedRestart = true + w.stsReconciler.PrepareHostStatefulSetWithStatus(ctx, host, host.IsStopped()) } - - w.stsReconciler.PrepareHostStatefulSetWithStatus(ctx, host, host.IsStopped()) // After a force restart the STS was scaled down to 0 replicas. PrepareHostStatefulSetWithStatus // compares fingerprints of the desired STS (replicas=1) vs the current STS (replicas=0, set by // hostScaleDown). In the current codebase these fingerprints differ, so ObjectStatusSame is not @@ -470,16 +490,6 @@ func (w *worker) reconcileHostStatefulSet(ctx context.Context, host *api.Host, o w.a.V(1).M(host).F().Info("Override ObjectStatusSame after force restart to ensure scale-up: %s", host.GetName()) host.GetReconcileAttributes().SetStatus(types.ObjectStatusModified) } - // Update / recreate / force-recreate can take a Ready host down. Refuse when that - // would leave the ensemble below Raft quorum (#2069). No-op when there is no live - // quorum (bootstrap, resume-from-stopped). - if (opts != nil && opts.ForceRecreate()) || !host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusSame) { - if err := w.ensureQuorumSafeToDisruptHost(ctx, host); err != nil { - w.a.V(1).M(host).F().Error("%v", err) - return err - } - } - opts = w.prepareStsReconcileOptsWaitSection(ctx, host, opts, waitReady) // We are in place, where we can reconcile StatefulSet to desired configuration. w.a.V(1).M(host).F().Info("Reconcile host STS: %s. Reconcile StatefulSet", host.GetName()) @@ -647,6 +657,7 @@ func (w *worker) reconcileClusterShardsAndHosts(ctx context.Context, cluster *ap // Which shard to start concurrent processing with var startShard int + deferred := false if opts.FullFanOut { // For full fan-out scenarios we'll start shards processing from the very beginning startShard = 0 @@ -657,8 +668,11 @@ func (w *worker) reconcileClusterShardsAndHosts(ctx context.Context, cluster *ap // and for large clusters it is a small price to pay before performing concurrent fan-out. w.a.V(1).Info("starting first shard separately") if err := w.reconcileShardWithHosts(ctx, shards[0]); err != nil { - w.a.V(1).Warning("first shard failed, skipping rest of shards due to an error: %v", err) - return err + if !errors.Is(err, common.ErrCRUDDeferred) { + w.a.V(1).Warning("first shard failed, skipping rest of shards due to an error: %v", err) + return err + } + deferred = true } // Since shard with 0 index is already done, we'll proceed concurrently starting with the 1-st @@ -670,10 +684,16 @@ func (w *worker) reconcileClusterShardsAndHosts(ctx context.Context, cluster *ap workersNum := w.getReconcileShardsWorkersNum(cluster, opts) w.a.V(1).Info("Starting rest of shards on workers. Workers num: %d", workersNum) if err := w.runConcurrently(ctx, workersNum, startShard, shards[startShard:]); err != nil { - w.a.V(1).Info("Finished with ERROR rest of shards on workers: %d, err: %v", workersNum, err) - return err + if !errors.Is(err, common.ErrCRUDDeferred) { + w.a.V(1).Info("Finished with ERROR rest of shards on workers: %d, err: %v", workersNum, err) + return err + } + deferred = true } w.a.V(1).Info("Finished successfully rest of shards on workers: %d", workersNum) + if deferred { + return common.ErrCRUDDeferred + } return nil } @@ -681,9 +701,44 @@ func (w *worker) reconcileShardWithHosts(ctx context.Context, shard api.IShard) if err := w.reconcileShard(ctx, shard); err != nil { return err } - return shard.WalkHostsAbortOnError(func(host *api.Host) error { - return w.reconcileHost(ctx, host) + + // Recovery first, then rollout: bring a replica that is already down back up before + // touching its healthy peer, so an interrupted roll cannot take the ensemble below + // Raft quorum (#2069, n=2). Disruption is gated in reconcileHostStatefulSet. + deferred := false + for _, host := range shardHostsRecoveryFirst(shard, func(h *api.Host) bool { + return w.isHostHealthyForReconcile(ctx, h) + }) { + if err := w.reconcileHost(ctx, host); err != nil { + if errors.Is(err, common.ErrCRUDDeferred) { + deferred = true + continue + } + return err + } + } + if deferred { + return common.ErrCRUDDeferred + } + return nil +} + +// shardHostsRecoveryFirst returns shard hosts with not-ready replicas first, then +// ready ones — same ordering as CHI reconcileShardWithHosts (#1704). +func shardHostsRecoveryFirst(shard api.IShard, healthy func(*api.Host) bool) []*api.Host { + if shard == nil { + return nil + } + var recovery, rollout []*api.Host + shard.WalkHosts(func(host *api.Host) error { + if healthy(host) { + rollout = append(rollout, host) + } else { + recovery = append(recovery, host) + } + return nil }) + return append(recovery, rollout...) } // reconcileShard reconciles specified shard, excluding nested replicas @@ -833,8 +888,11 @@ func (w *worker) reconcileHostMain(ctx context.Context, host *api.Host) error { Warning("Reconcile Host Main - unable to reconcile Service. Host: %s Err: %v", host.GetName(), err) } + // Snapshot rolling vs bootstrap before any STS disruption on this host. + snap := w.snapshotHostEnsemble(ctx, host) + // Reconcile StatefulSet - if err := w.reconcileHostStatefulSet(ctx, host, stsReconcileOpts); err != nil { + if err := w.reconcileHostStatefulSet(ctx, host, stsReconcileOpts, snap); err != nil { metrics.HostReconcilesErrors(ctx, host.GetCR()) w.a.V(1). M(host).F(). @@ -852,7 +910,7 @@ func (w *worker) reconcileHostMain(ctx context.Context, host *api.Host) error { // Finalize main reconcile with domain activities. // Membership / ensemble gates must abort the host loop — continuing would // recreate the next replica while this one never rejoined (#2069). - if err := w.reconcileHostMainDomain(ctx, host); err != nil { + if err := w.reconcileHostMainDomain(ctx, host, snap); err != nil { metrics.HostReconcilesErrors(ctx, host.GetCR()) w.a.V(1). M(host).F(). @@ -871,21 +929,49 @@ func (w *worker) reconcileHostPVCs(ctx context.Context, host *api.Host) storage. ).ReconcilePVCs(ctx, host, api.DesiredStatefulSet) } -func (w *worker) reconcileHostMainDomain(ctx context.Context, host *api.Host) error { +func (w *worker) reconcileHostMainDomain(ctx context.Context, host *api.Host, snap hostEnsembleSnapshot) error { if !host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) { return nil } - if !w.ensembleHasLiveQuorum(ctx, host.GetCR()) { + if !snap.rolling { // Bootstrap / resume-from-stopped / recovery: peers start together; - // keep the legacy pacing wait (Ready wait was skipped). + // legacy pacing wait (Ready wait was skipped on STS). util.WaitContextDoneOrTimeout(ctx, 7*time.Second) return nil } - // Live quorum mode: STS Ready wait already ran. Hook for richer Raft - // membership confirmation (PR #2041) lives in waitHostRaftJoined. - return w.waitHostRaftJoined(ctx, host) + // Extension point for richer Raft membership confirmation (PR #2041). + return w.verifyHostEnsembleMembership(ctx, host) +} + +// prepareStsReconcileOptsWaitSection sets STS launch waits for Keeper. +// +// rolling comes from snapshotHostEnsemble before any disruption. +func (w *worker) prepareStsReconcileOptsWaitSection( + host *api.Host, + opts *statefulset.ReconcileOptions, + rolling bool, +) *statefulset.ReconcileOptions { + if opts == nil { + opts = statefulset.NewReconcileStatefulSetOptions() + } + probes := host.GetCluster().GetReconcile().Host.Wait.Probes + + if probes.GetStartup().IsTrue() || !rolling { + opts = opts.SetWaitUntilStarted() + w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilStarted") + } + + switch { + case rolling && !probes.GetReadiness().IsFalse(): + opts = opts.SetWaitUntilReady() + w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilReady (Keeper must become Ready)") + case !rolling: + w.a.V(1).M(host).F().Info("Skip WaitUntilReady — bootstrap / resume-from-stopped / recovery") + } + + return opts } // reconcileHostIncludeIntoAllActivities includes specified ClickHouse host into all activities diff --git a/pkg/controller/chk/worker-reconciler-chk_test.go b/pkg/controller/chk/worker-reconciler-chk_test.go index 73e491c27..c2df941ce 100644 --- a/pkg/controller/chk/worker-reconciler-chk_test.go +++ b/pkg/controller/chk/worker-reconciler-chk_test.go @@ -26,6 +26,7 @@ import ( 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/controller/common/statefulset" "github.com/altinity/clickhouse-operator/pkg/interfaces" ) @@ -154,3 +155,45 @@ func TestMarkReconcileStartReturnsStatusUpdateError(t *testing.T) { t.Fatalf("status = %q, want %q before persistence attempt", got, api.StatusInProgress) } } + +func TestPrepareStsReconcileOptsWaitSection(t *testing.T) { + w := &worker{a: a.NewAnnouncer(nil, nil)} + + t.Run("bootstrap skips Ready", func(t *testing.T) { + host := hostOnCR(chkWithHosts(3)) + opts := w.prepareStsReconcileOptsWaitSection(host, nil, false) + if !opts.WaitUntilStarted() || opts.WaitUntilReady() { + t.Fatal("bootstrap should wait Started only") + } + }) + + t.Run("rolling waits Ready", func(t *testing.T) { + host := hostOnCR(chkWithHosts(3)) + opts := w.prepareStsReconcileOptsWaitSection(host, nil, true) + if !opts.WaitUntilReady() { + t.Fatal("rolling should wait Ready") + } + }) + + t.Run("rolling can opt out of Ready probe", func(t *testing.T) { + host := hostOnCR(chkWithHosts(3)) + host.GetCluster().GetReconcile().Host.Wait.Probes.Readiness = types.NewStringBool(false) + opts := w.prepareStsReconcileOptsWaitSection(host, statefulset.NewReconcileStatefulSetOptions(), true) + if opts.WaitUntilReady() { + t.Fatal("readiness=false should skip Ready wait") + } + }) + + t.Run("single-host post-restart still waits Ready", func(t *testing.T) { + w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 0 } + host := hostOnCR(chkWithHosts(1)) + snap := w.snapshotHostEnsemble(context.Background(), host) + if !snap.rolling { + t.Fatal("single host should be rolling") + } + opts := w.prepareStsReconcileOptsWaitSection(host, nil, snap.rolling) + if !opts.WaitUntilReady() { + t.Fatal("rolling snapshot must drive Ready wait after force-restart") + } + }) +} diff --git a/pkg/controller/chk/worker-reconciler-helper.go b/pkg/controller/chk/worker-reconciler-helper.go index d3e289f5e..3bdcd9a9f 100644 --- a/pkg/controller/chk/worker-reconciler-helper.go +++ b/pkg/controller/chk/worker-reconciler-helper.go @@ -16,6 +16,7 @@ package chk import ( "context" + "errors" "sync" apiChk "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse-keeper.altinity.com/v1" @@ -80,6 +81,7 @@ func (w *worker) runConcurrently(ctx context.Context, workersNum int, startShard // Launch workers var err error + var deferred bool var errLock sync.Mutex for i := 0; i < workersNum; i++ { wg.Add(1) @@ -89,7 +91,11 @@ func (w *worker) runConcurrently(ctx context.Context, workersNum int, startShard w.a.V(1).Info("Starting shard index: %d on worker", rq.index) if e := w.reconcileShardWithHosts(ctx, rq.shard); e != nil { errLock.Lock() - err = e + if errors.Is(e, common.ErrCRUDDeferred) { + deferred = true + } else { + err = e + } errLock.Unlock() } } @@ -99,7 +105,13 @@ func (w *worker) runConcurrently(ctx context.Context, workersNum int, startShard w.a.V(1).Info("Starting to wait shards from index: %d on workers.", startShardIndex) wg.Wait() w.a.V(1).Info("Finished to wait shards from index: %d on workers.", startShardIndex) - return err + if err != nil { + return err + } + if deferred { + return common.ErrCRUDDeferred + } + return nil } func (w *worker) hostPVCsDataLossDetectedOptions(host *api.Host) *statefulset.ReconcileOptions { diff --git a/pkg/controller/chk/worker.go b/pkg/controller/chk/worker.go index 7c10f1fe8..f56c8148b 100644 --- a/pkg/controller/chk/worker.go +++ b/pkg/controller/chk/worker.go @@ -59,6 +59,11 @@ type worker struct { // countReadyEnsembleMembersFn overrides live Ready counting (tests only). countReadyEnsembleMembersFn func(ctx context.Context, cr api.ICustomResource) int + // quorumDisruptPollOverride / quorumDisruptWaitOverride override pacing in + // waitForQuorumSafeToDisruptHost (tests only). Zero means use defaults. + quorumDisruptPollOverride time.Duration + quorumDisruptWaitOverride time.Duration + start time.Time } diff --git a/tests/e2e/manifests/chk/test-020003-3-chi.yaml b/tests/e2e/manifests/chk/test-020003-3-chi.yaml new file mode 100644 index 000000000..e9f56fc5e --- /dev/null +++ b/tests/e2e/manifests/chk/test-020003-3-chi.yaml @@ -0,0 +1,16 @@ +apiVersion: "clickhouse.altinity.com/v1" +kind: "ClickHouseInstallation" +metadata: + name: test-020003-3-chi +spec: + useTemplates: + - name: clickhouse-version + configuration: + zookeeper: + nodes: + - host: keeper-test-020003-3-chk + port: 2181 + clusters: + - name: default + layout: + replicasCount: 2 diff --git a/tests/e2e/manifests/chk/test-020003-3-chk-1.yaml b/tests/e2e/manifests/chk/test-020003-3-chk-1.yaml new file mode 100644 index 000000000..7b497e43a --- /dev/null +++ b/tests/e2e/manifests/chk/test-020003-3-chk-1.yaml @@ -0,0 +1,20 @@ +apiVersion: "clickhouse-keeper.altinity.com/v1" +kind: "ClickHouseKeeperInstallation" +metadata: + name: test-020003-3-chk +spec: + defaults: + templates: + podTemplate: default + configuration: + clusters: + - name: keeper + layout: + replicasCount: 3 + templates: + podTemplates: + - name: default + spec: + containers: + - name: clickhouse-keeper + image: "clickhouse/clickhouse-keeper:25.8" diff --git a/tests/e2e/manifests/chk/test-020003-3-chk-2.yaml b/tests/e2e/manifests/chk/test-020003-3-chk-2.yaml new file mode 100644 index 000000000..8843174bf --- /dev/null +++ b/tests/e2e/manifests/chk/test-020003-3-chk-2.yaml @@ -0,0 +1,20 @@ +apiVersion: "clickhouse-keeper.altinity.com/v1" +kind: "ClickHouseKeeperInstallation" +metadata: + name: test-020003-3-chk +spec: + defaults: + templates: + podTemplate: default + configuration: + clusters: + - name: keeper + layout: + replicasCount: 3 + templates: + podTemplates: + - name: default + spec: + containers: + - name: clickhouse-keeper + image: "clickhouse/clickhouse-keeper:25.8-broken" diff --git a/tests/e2e/manifests/chk/test-020003-3-chk-3.yaml b/tests/e2e/manifests/chk/test-020003-3-chk-3.yaml new file mode 100644 index 000000000..34fbd0155 --- /dev/null +++ b/tests/e2e/manifests/chk/test-020003-3-chk-3.yaml @@ -0,0 +1,20 @@ +apiVersion: "clickhouse-keeper.altinity.com/v1" +kind: "ClickHouseKeeperInstallation" +metadata: + name: test-020003-3-chk +spec: + defaults: + templates: + podTemplate: default + configuration: + clusters: + - name: keeper + layout: + replicasCount: 3 + templates: + podTemplates: + - name: default + spec: + containers: + - name: clickhouse-keeper + image: "clickhouse/clickhouse-keeper:26.3" diff --git a/tests/e2e/test_operator.py b/tests/e2e/test_operator.py index 6e5bf0bbe..ad6618730 100644 --- a/tests/e2e/test_operator.py +++ b/tests/e2e/test_operator.py @@ -8243,6 +8243,157 @@ def test_020003_2(self): delete_test_namespace() +@TestScenario +@Tags("HEAVY") +@Name("test_020003_3. Interrupted Keeper roll must preserve Raft quorum (issue #2069)") +def test_020003_3(self): + """Companion to test_010083 for ClickHouse Keeper. + + Reproduce issue #2069: a broken keeper image roll must stop after the first + replica and must not disrupt healthy peers that still hold Raft quorum. + + Uses a broken image so the first replica stays permanently unhealthy + (ImagePullBackOff), giving a deterministic mid-roll window. + + Scenario: + 1. Start a 3-node CHK and a 2-replica CHI on a good keeper image + 2. Roll keeper to a broken image (operator updates replica 0 first) + 3. Wait until replica 0 is ImagePullBackOff and replicas 1-2 are still Ready + 4. Restart the operator and force reconcile + 5. Replicas 1-2 must stay Ready and must not be recreated (startTime unchanged) + 6. Roll the good image back and wait for all keeper replicas to recover + 7. ClickHouse replication must still work + """ + create_shell_namespace_clickhouse_template() + + chk = "test-020003-3-chk" + chi = "test-020003-3-chi" + cluster = "keeper" + down_pod = f"chk-{chk}-{cluster}-0-0-0" + healthy_pods = [ + f"chk-{chk}-{cluster}-0-1-0", + f"chk-{chk}-{cluster}-0-2-0", + ] + good_version = "clickhouse/clickhouse-keeper:25.8" + broken_version = "clickhouse/clickhouse-keeper:25.8-broken" + new_version = "clickhouse/clickhouse-keeper:26.3" + + with Given("CHK with 3 replicas on a good image"): + kubectl.create_and_check( + manifest="manifests/chk/test-020003-3-chk-1.yaml", + kind="chk", + check={ + "pod_count": 3, + "do_not_delete": 1, + }, + ) + + with And("CHI with 2 replicas connected to the keeper"): + kubectl.create_and_check( + manifest="manifests/chk/test-020003-3-chi.yaml", + check={ + "pod_count": 2, + "do_not_delete": 1, + }, + ) + + check_replication(chi, {0, 1}, 1) + + healthy_start_times = { + pod: kubectl.get_field("pod", pod, ".status.startTime") + for pod in healthy_pods + } + + with When("Rolling update to a broken keeper image is started"): + kubectl.create_and_check( + manifest="manifests/chk/test-020003-3-chk-2.yaml", + kind="chk", + check={ + "chk_status": "InProgress", + "do_not_delete": 1, + }, + ) + + with And("First replica is stuck on the broken image while the others stay Ready"): + kubectl.wait_field( + "pod", + down_pod, + ".status.containerStatuses[0].state.waiting.reason", + ["ErrImagePull", "ImagePullBackOff"], + ) + down_image = kubectl.get_field("pod", down_pod, ".spec.containers[0].image") + assert broken_version in down_image, error( + f"down replica {down_pod} must be on broken image {broken_version}, got {down_image}" + ) + for pod in healthy_pods: + assert kubectl.get_condition_status(pod, "Ready") == "True", error( + f"healthy replica {pod} must stay Ready while {down_pod} is pulling the broken image" + ) + cur_start = kubectl.get_field("pod", pod, ".status.startTime") + assert cur_start == healthy_start_times[pod], error( + f"healthy replica {pod} must not be restarted during the broken-image roll, " + f"but startTime changed from {healthy_start_times[pod]} to {cur_start}" + ) + + with And("ClickHouse still reaches Keeper while the first replica is down"): + for attempt in retries(timeout=60, delay=5): + out = clickhouse.query_with_error(chi, "select * from system.zookeeper_connection") + if "KEEPER_EXCEPTION" not in out and "Exception" not in out: + break + + with And("Operator is restarted while the first replica is still down"): + util.restart_operator() + + with And("Reconcile is forced while the first replica is still down"): + kubectl.force_chk_reconcile(chk, "force", "InProgress") + + with Then("Healthy replicas were never restarted while the broken replica is down"): + assert kubectl.get_condition_status(down_pod, "Ready") != "True", error( + f"broken-image replica {down_pod} unexpectedly became Ready" + ) + for pod in healthy_pods: + assert kubectl.get_condition_status(pod, "Ready") == "True", error( + f"healthy replica {pod} must stay Ready while {down_pod} is down " + f"(issue #2069 Raft quorum safety)" + ) + cur_start = kubectl.get_field("pod", pod, ".status.startTime") + assert cur_start == healthy_start_times[pod], error( + f"healthy replica {pod} must never be restarted while peer is down, " + f"but startTime changed from {healthy_start_times[pod]} to {cur_start}" + ) + + with When("New keeper image is applied"): + kubectl.create_and_check( + manifest="manifests/chk/test-020003-3-chk-3.yaml", + kind="chk", + check={ + "pod_count": 3, + "chk_status": "Completed", + "do_not_delete": 1, + }, + ) + + with Then("All keeper replicas recover on the new image"): + for pod in [down_pod] + healthy_pods: + kubectl.wait_field( + "pod", pod, ".status.containerStatuses[0].ready", "true", retries=30, + ) + image = kubectl.get_field("pod", pod, ".spec.containers[0].image") + assert new_version in image, error( + f"{pod} must run {new_version} after restore, but image={image}" + ) + + with And("ClickHouse replication works after keeper recovery"): + for attempt in retries(timeout=180, delay=5): + out = clickhouse.query_with_error(chi, "select * from system.zookeeper_connection") + if "KEEPER_EXCEPTION" not in out and "Exception" not in out: + break + check_replication(chi, {0, 1}, 2) + + with Finally("I clean up"): + delete_test_namespace() + + @TestScenario @Tags("HEAVY") @Name("test_020005. Clickhouse-keeper scale-up/scale-down") @@ -8304,6 +8455,9 @@ def test_020005(self): }, ) + with Then("Confirm CHK pod is ready"): + kubectl.wait_field('pod', 'chk-test-052-chk-keeper-0-0-0', '.status.containerStatuses[0].ready', 'true', retries=10) + check_replication(chi, {0, 1}, 5) with Finally("I clean up"): From 41cae72571b5f5b4f85ef3b7d5044ebf95d79d6d Mon Sep 17 00:00:00 2001 From: alz Date: Sat, 22 Aug 2026 13:47:05 +0300 Subject: [PATCH 5/6] Fix data race in TestWaitForQuorumSafeToDisruptHost Use atomic.Int32 for the ready-count stub shared between the wait loop and the goroutine that simulates a peer recovering. Co-authored-by: Cursor --- pkg/controller/chk/worker-raft-safety_test.go | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/pkg/controller/chk/worker-raft-safety_test.go b/pkg/controller/chk/worker-raft-safety_test.go index 131daab27..8b4c2d077 100644 --- a/pkg/controller/chk/worker-raft-safety_test.go +++ b/pkg/controller/chk/worker-raft-safety_test.go @@ -17,6 +17,7 @@ package chk import ( "context" "errors" + "sync/atomic" "testing" "time" @@ -179,23 +180,21 @@ func TestWaitForQuorumSafeToDisruptHost(t *testing.T) { }) t.Run("waits until ready count increases", func(t *testing.T) { - ready := 2 + var ready atomic.Int32 + ready.Store(2) w := &worker{ countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { - return ready + return int(ready.Load()) }, quorumDisruptPollOverride: 5 * time.Millisecond, - quorumDisruptWaitOverride: 200 * time.Millisecond, + quorumDisruptWaitOverride: 200 * time.Millisecond, } waitSnap := snap - done := make(chan struct{}) go func() { time.Sleep(20 * time.Millisecond) - ready = 3 - close(done) + ready.Store(3) }() require.NoError(t, w.waitForQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap)) - <-done }) t.Run("defers after wait budget expires", func(t *testing.T) { From a35be53b85013d21b3a67b76dd0e1e1b3a5147bb Mon Sep 17 00:00:00 2001 From: alz Date: Sun, 23 Aug 2026 13:50:18 +0300 Subject: [PATCH 6/6] Clarify CHK raft safety layout and soft-requeue deferred quorum waits. Move ensemble policy into worker-raft-safety, collapse the disrupt gate behind ensureQuorumSafeToDisruptHost, and requeue ErrCRUDDeferred after 5s instead of error backoff so Raft headroom waits stay intentional. Co-authored-by: Cursor --- pkg/controller/chk/controller.go | 15 +++ pkg/controller/chk/worker-raft-safety.go | 96 +++++++++++++++-- pkg/controller/chk/worker-raft-safety_test.go | 78 +++++++++++++- pkg/controller/chk/worker-reconciler-chk.go | 102 ++---------------- .../chk/worker-reconciler-chk_test.go | 72 ------------- .../chk/worker-reconciler-helper.go | 21 +++- pkg/controller/chk/worker.go | 2 +- 7 files changed, 204 insertions(+), 182 deletions(-) diff --git a/pkg/controller/chk/controller.go b/pkg/controller/chk/controller.go index 2942efdd3..d28337076 100644 --- a/pkg/controller/chk/controller.go +++ b/pkg/controller/chk/controller.go @@ -16,6 +16,7 @@ package chk import ( "context" + "errors" "time" apiExtensions "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" @@ -30,11 +31,16 @@ import ( api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1" "github.com/altinity/clickhouse-operator/pkg/chop" "github.com/altinity/clickhouse-operator/pkg/controller/chk/kube" + "github.com/altinity/clickhouse-operator/pkg/controller/common" "github.com/altinity/clickhouse-operator/pkg/interfaces" "github.com/altinity/clickhouse-operator/pkg/model/managers" "github.com/altinity/clickhouse-operator/pkg/util" ) +// raftQuorumDeferredRequeueAfter is the fixed retry interval when reconcile +// returns ErrCRUDDeferred (Raft quorum headroom not available yet). +const raftQuorumDeferredRequeueAfter = 5 * time.Second + // Controller reconciles a ClickHouseKeeper object type Controller struct { client.Client @@ -113,6 +119,15 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu } if err := w.reconcileCR(ctx, nil, new); err != nil { + if errors.Is(err, common.ErrCRUDDeferred) { + // Soft defer: quorum headroom not available yet. Status already + // records [RaftQuorumUnsafe]; retry on a fixed interval instead of + // error backoff. + log.V(1).M(new).F().Info( + "Raft quorum defer — requeue in %s", raftQuorumDeferredRequeueAfter, + ) + return ctrl.Result{RequeueAfter: raftQuorumDeferredRequeueAfter}, nil + } return ctrl.Result{}, err } diff --git a/pkg/controller/chk/worker-raft-safety.go b/pkg/controller/chk/worker-raft-safety.go index 704ba1574..e2beb5da0 100644 --- a/pkg/controller/chk/worker-raft-safety.go +++ b/pkg/controller/chk/worker-raft-safety.go @@ -21,6 +21,7 @@ import ( apps "k8s.io/api/apps/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" "github.com/altinity/clickhouse-operator/pkg/controller/common" @@ -30,14 +31,19 @@ import ( "github.com/altinity/clickhouse-operator/pkg/util" ) -// Raft / ensemble safety for CHK (#2069). +// Raft / ensemble safety policy for CHK (#2069). // -// snapshotHostEnsemble records rolling vs bootstrap once per host, before any -// disruption. Live Ready count drives rolling — not CR ancestor / host inventory: +// Owns rolling-vs-bootstrap classification, STS wait probes, quorum disrupt gate, +// recovery-first host ordering, and membership settle delays. The reconciler +// calls into these helpers; it should not restate the policy inline. // -// - Rolling (n<=1 or live quorum): wait Ready, refuse disrupt below quorum. -// - Bootstrap: wait Started only so siblings can come up together. +// Before disrupting a host STS: // +// 1. snapshotHostEnsemble — freeze rolling vs bootstrap (live Ready, not ancestor) +// 2. prepareStsReconcileOptsWaitSection — Ready wait iff rolling +// 3. ensureQuorumSafeToDisruptHost — wait/defer if disrupt would break Raft majority +// +// hostDisruptionWouldBreakQuorum is the tested predicate used inside the façade. // verifyHostEnsembleMembership is the extension point for a fuller Raft barrier // (committed /keeper/config + mntr, as in PR #2041). @@ -153,9 +159,11 @@ func (w *worker) quorumDisruptWaitTimeout() time.Duration { return defaultQuorumDisruptWaitTimeout } -// waitForQuorumSafeToDisruptHost polls until disrupting the host would not drop -// the ensemble below Raft quorum, or until the wait budget expires. -func (w *worker) waitForQuorumSafeToDisruptHost( +// ensureQuorumSafeToDisruptHost is the reconciler façade for the Raft disrupt gate. +// Call after PrepareHostStatefulSetWithStatus so ObjectStatusSame is assigned. +// Returns nil when safe (or not a rolling multi-member disrupt); ErrCRUDDeferred +// after waiting up to the budget without quorum headroom. +func (w *worker) ensureQuorumSafeToDisruptHost( ctx context.Context, host *api.Host, opts *statefulset.ReconcileOptions, @@ -280,3 +288,75 @@ func (w *worker) verifyHostEnsembleMembership(ctx context.Context, host *api.Hos _ = host return nil } + +// prepareStsReconcileOptsWaitSection sets STS launch waits for Keeper. +// rolling comes from snapshotHostEnsemble before any disruption. +func (w *worker) prepareStsReconcileOptsWaitSection( + host *api.Host, + opts *statefulset.ReconcileOptions, + rolling bool, +) *statefulset.ReconcileOptions { + if opts == nil { + opts = statefulset.NewReconcileStatefulSetOptions() + } + probes := host.GetCluster().GetReconcile().Host.Wait.Probes + + if probes.GetStartup().IsTrue() || !rolling { + opts = opts.SetWaitUntilStarted() + w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilStarted") + } + + switch { + case rolling && !probes.GetReadiness().IsFalse(): + opts = opts.SetWaitUntilReady() + w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilReady (Keeper must become Ready)") + case !rolling: + w.a.V(1).M(host).F().Info("Skip WaitUntilReady — bootstrap / resume-from-stopped / recovery") + } + + return opts +} + +// membershipSettleDelay is a best-effort pause after publishing membership +// changes so Raft can settle: +// - same host count → no delay +// - upscale → 30s +// - downscale → 120s (survivors still need time after raft_configuration shrink; +// peer purge later adds another 1m in clean()) +func (w *worker) membershipSettleDelay(cr *apiChk.ClickHouseKeeperInstallation) time.Duration { + if cr == nil { + return 0 + } + ancestorHosts := 0 + if ancestor := cr.GetAncestor(); ancestor != nil { + ancestorHosts = ancestor.HostsCount() + } + currentHosts := cr.HostsCount() + + switch { + case currentHosts < ancestorHosts: + return 120 * time.Second + case currentHosts > ancestorHosts: + return 30 * time.Second + default: + return 0 + } +} + +// shardHostsRecoveryFirst returns shard hosts with not-ready replicas first, then +// ready ones — same ordering as CHI reconcileShardWithHosts (#1704). +func shardHostsRecoveryFirst(shard api.IShard, healthy func(*api.Host) bool) []*api.Host { + if shard == nil { + return nil + } + var recovery, rollout []*api.Host + shard.WalkHosts(func(host *api.Host) error { + if healthy(host) { + rollout = append(rollout, host) + } else { + recovery = append(recovery, host) + } + return nil + }) + return append(recovery, rollout...) +} diff --git a/pkg/controller/chk/worker-raft-safety_test.go b/pkg/controller/chk/worker-raft-safety_test.go index 8b4c2d077..f3609889d 100644 --- a/pkg/controller/chk/worker-raft-safety_test.go +++ b/pkg/controller/chk/worker-raft-safety_test.go @@ -165,7 +165,7 @@ func TestErrCRUDDeferredIsDistinctFromAbort(t *testing.T) { require.False(t, errors.Is(common.ErrCRUDAbort, common.ErrCRUDDeferred)) } -func TestWaitForQuorumSafeToDisruptHost(t *testing.T) { +func TestEnsureQuorumSafeToDisruptHost(t *testing.T) { ctx := context.Background() host := hostOnCR(chkWithHosts(3)) host.Runtime.CurStatefulSet = &apps.StatefulSet{} @@ -176,7 +176,7 @@ func TestWaitForQuorumSafeToDisruptHost(t *testing.T) { t.Run("returns immediately when already safe", func(t *testing.T) { w := &worker{} safeSnap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 3} - require.NoError(t, w.waitForQuorumSafeToDisruptHost(ctx, host, nil, &safeSnap)) + require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, host, nil, &safeSnap)) }) t.Run("waits until ready count increases", func(t *testing.T) { @@ -194,7 +194,7 @@ func TestWaitForQuorumSafeToDisruptHost(t *testing.T) { time.Sleep(20 * time.Millisecond) ready.Store(3) }() - require.NoError(t, w.waitForQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap)) + require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap)) }) t.Run("defers after wait budget expires", func(t *testing.T) { @@ -204,7 +204,7 @@ func TestWaitForQuorumSafeToDisruptHost(t *testing.T) { quorumDisruptWaitOverride: 20 * time.Millisecond, } waitSnap := snap - err := w.waitForQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap) + err := w.ensureQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap) require.ErrorIs(t, err, common.ErrCRUDDeferred) }) } @@ -259,6 +259,76 @@ func TestShardHostsRecoveryFirst(t *testing.T) { require.Same(t, h0, ordered[1], "ready host should reconcile second") } +func TestMembershipSettleDelay(t *testing.T) { + w := &worker{} + + t.Run("same size does not wait", func(t *testing.T) { + cr := chkWithHosts(3) + cr.SetAncestor(chkWithHosts(3)) + if got := w.membershipSettleDelay(cr); got != 0 { + t.Fatalf("membershipSettleDelay() = %s, want 0", got) + } + }) + + t.Run("upscale waits for raft membership", func(t *testing.T) { + cr := chkWithHosts(3) + cr.SetAncestor(chkWithHosts(1)) + if got := w.membershipSettleDelay(cr); got != 30*time.Second { + t.Fatalf("membershipSettleDelay() = %s, want 30s", got) + } + }) + + t.Run("downscale always waits 120s", func(t *testing.T) { + cr := chkWithHosts(1) + cr.SetAncestor(chkWithHosts(3)) + if got := w.membershipSettleDelay(cr); got != 120*time.Second { + t.Fatalf("membershipSettleDelay() = %s, want 120s", got) + } + }) +} + +func TestPrepareStsReconcileOptsWaitSection(t *testing.T) { + w := &worker{} + + t.Run("bootstrap skips Ready", func(t *testing.T) { + host := hostOnCR(chkWithHosts(3)) + opts := w.prepareStsReconcileOptsWaitSection(host, nil, false) + if !opts.WaitUntilStarted() || opts.WaitUntilReady() { + t.Fatal("bootstrap should wait Started only") + } + }) + + t.Run("rolling waits Ready", func(t *testing.T) { + host := hostOnCR(chkWithHosts(3)) + opts := w.prepareStsReconcileOptsWaitSection(host, nil, true) + if !opts.WaitUntilReady() { + t.Fatal("rolling should wait Ready") + } + }) + + t.Run("rolling can opt out of Ready probe", func(t *testing.T) { + host := hostOnCR(chkWithHosts(3)) + host.GetCluster().GetReconcile().Host.Wait.Probes.Readiness = types.NewStringBool(false) + opts := w.prepareStsReconcileOptsWaitSection(host, statefulset.NewReconcileStatefulSetOptions(), true) + if opts.WaitUntilReady() { + t.Fatal("readiness=false should skip Ready wait") + } + }) + + t.Run("single-host post-restart still waits Ready", func(t *testing.T) { + w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 0 } + host := hostOnCR(chkWithHosts(1)) + snap := w.snapshotHostEnsemble(context.Background(), host) + if !snap.rolling { + t.Fatal("single host should be rolling") + } + opts := w.prepareStsReconcileOptsWaitSection(host, nil, snap.rolling) + if !opts.WaitUntilReady() { + t.Fatal("rolling snapshot must drive Ready wait after force-restart") + } + }) +} + func chkWithHosts(n int) *apiChk.ClickHouseKeeperInstallation { cr := &apiChk.ClickHouseKeeperInstallation{} cr.EnsureRuntime() diff --git a/pkg/controller/chk/worker-reconciler-chk.go b/pkg/controller/chk/worker-reconciler-chk.go index 398ede8ae..9b8b938bf 100644 --- a/pkg/controller/chk/worker-reconciler-chk.go +++ b/pkg/controller/chk/worker-reconciler-chk.go @@ -312,32 +312,6 @@ func (w *worker) reconcileCRAuxObjectsPreliminaryDomain(ctx context.Context, cr return nil } -// membershipSettleDelay is a best-effort pause after publishing membership -// changes so Raft can settle: -// - same host count → no delay -// - upscale → 30s -// - downscale → 120s (survivors still need time after raft_configuration shrink; -// peer purge later adds another 1m in clean()) -func (w *worker) membershipSettleDelay(cr *apiChk.ClickHouseKeeperInstallation) time.Duration { - if cr == nil { - return 0 - } - ancestorHosts := 0 - if ancestor := cr.GetAncestor(); ancestor != nil { - ancestorHosts = ancestor.HostsCount() - } - currentHosts := cr.HostsCount() - - 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() { @@ -463,13 +437,8 @@ func (w *worker) reconcileHostStatefulSet( w.stsReconciler.PrepareHostStatefulSetWithStatus(ctx, host, host.IsStopped()) opts = w.prepareStsReconcileOptsWaitSection(host, opts, snap.rolling) - - // First point where "would this pass take the host down?" can be answered — same - // placement as CHI hostDisruptionWouldDegradeShard (#1704). - if w.hostDisruptionWouldBreakQuorum(ctx, host, opts, snap) { - if err := w.waitForQuorumSafeToDisruptHost(ctx, host, opts, &snap); err != nil { - return err - } + if err := w.ensureQuorumSafeToDisruptHost(ctx, host, opts, &snap); err != nil { + return err } forcedRestart := false @@ -668,11 +637,10 @@ func (w *worker) reconcileClusterShardsAndHosts(ctx context.Context, cluster *ap // and for large clusters it is a small price to pay before performing concurrent fan-out. w.a.V(1).Info("starting first shard separately") if err := w.reconcileShardWithHosts(ctx, shards[0]); err != nil { - if !errors.Is(err, common.ErrCRUDDeferred) { - w.a.V(1).Warning("first shard failed, skipping rest of shards due to an error: %v", err) - return err + if hard := noteCRUDResult(err, &deferred); hard != nil { + w.a.V(1).Warning("first shard failed, skipping rest of shards due to an error: %v", hard) + return hard } - deferred = true } // Since shard with 0 index is already done, we'll proceed concurrently starting with the 1-st @@ -684,11 +652,10 @@ func (w *worker) reconcileClusterShardsAndHosts(ctx context.Context, cluster *ap workersNum := w.getReconcileShardsWorkersNum(cluster, opts) w.a.V(1).Info("Starting rest of shards on workers. Workers num: %d", workersNum) if err := w.runConcurrently(ctx, workersNum, startShard, shards[startShard:]); err != nil { - if !errors.Is(err, common.ErrCRUDDeferred) { - w.a.V(1).Info("Finished with ERROR rest of shards on workers: %d, err: %v", workersNum, err) - return err + if hard := noteCRUDResult(err, &deferred); hard != nil { + w.a.V(1).Info("Finished with ERROR rest of shards on workers: %d, err: %v", workersNum, hard) + return hard } - deferred = true } w.a.V(1).Info("Finished successfully rest of shards on workers: %d", workersNum) if deferred { @@ -709,11 +676,7 @@ func (w *worker) reconcileShardWithHosts(ctx context.Context, shard api.IShard) for _, host := range shardHostsRecoveryFirst(shard, func(h *api.Host) bool { return w.isHostHealthyForReconcile(ctx, h) }) { - if err := w.reconcileHost(ctx, host); err != nil { - if errors.Is(err, common.ErrCRUDDeferred) { - deferred = true - continue - } + if err := noteCRUDResult(w.reconcileHost(ctx, host), &deferred); err != nil { return err } } @@ -723,24 +686,6 @@ func (w *worker) reconcileShardWithHosts(ctx context.Context, shard api.IShard) return nil } -// shardHostsRecoveryFirst returns shard hosts with not-ready replicas first, then -// ready ones — same ordering as CHI reconcileShardWithHosts (#1704). -func shardHostsRecoveryFirst(shard api.IShard, healthy func(*api.Host) bool) []*api.Host { - if shard == nil { - return nil - } - var recovery, rollout []*api.Host - shard.WalkHosts(func(host *api.Host) error { - if healthy(host) { - rollout = append(rollout, host) - } else { - recovery = append(recovery, host) - } - return nil - }) - return append(recovery, rollout...) -} - // reconcileShard reconciles specified shard, excluding nested replicas func (w *worker) reconcileShard(ctx context.Context, shard api.IShard) error { if util.IsContextDone(ctx) { @@ -945,35 +890,6 @@ func (w *worker) reconcileHostMainDomain(ctx context.Context, host *api.Host, sn return w.verifyHostEnsembleMembership(ctx, host) } -// prepareStsReconcileOptsWaitSection sets STS launch waits for Keeper. -// -// rolling comes from snapshotHostEnsemble before any disruption. -func (w *worker) prepareStsReconcileOptsWaitSection( - host *api.Host, - opts *statefulset.ReconcileOptions, - rolling bool, -) *statefulset.ReconcileOptions { - if opts == nil { - opts = statefulset.NewReconcileStatefulSetOptions() - } - probes := host.GetCluster().GetReconcile().Host.Wait.Probes - - if probes.GetStartup().IsTrue() || !rolling { - opts = opts.SetWaitUntilStarted() - w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilStarted") - } - - switch { - case rolling && !probes.GetReadiness().IsFalse(): - opts = opts.SetWaitUntilReady() - w.a.V(1).M(host).F().Warning("Setting option SetWaitUntilReady (Keeper must become Ready)") - case !rolling: - w.a.V(1).M(host).F().Info("Skip WaitUntilReady — bootstrap / resume-from-stopped / recovery") - } - - return opts -} - // reconcileHostIncludeIntoAllActivities includes specified ClickHouse host into all activities func (w *worker) reconcileHostIncludeIntoAllActivities(ctx context.Context, host *api.Host) error { if !w.shouldIncludeHost(host) { diff --git a/pkg/controller/chk/worker-reconciler-chk_test.go b/pkg/controller/chk/worker-reconciler-chk_test.go index c2df941ce..e11e35c90 100644 --- a/pkg/controller/chk/worker-reconciler-chk_test.go +++ b/pkg/controller/chk/worker-reconciler-chk_test.go @@ -18,7 +18,6 @@ import ( "context" "errors" "testing" - "time" meta "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -26,7 +25,6 @@ import ( 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/controller/common/statefulset" "github.com/altinity/clickhouse-operator/pkg/interfaces" ) @@ -63,34 +61,6 @@ func newStatusTestWorker(statusWriter interfaces.IKubeCR) *worker { } } -func TestMembershipSettleDelay(t *testing.T) { - w := &worker{a: a.NewAnnouncer(nil, nil)} - - t.Run("same size does not wait", func(t *testing.T) { - cr := chkWithHosts(3) - cr.SetAncestor(chkWithHosts(3)) - if got := w.membershipSettleDelay(cr); got != 0 { - t.Fatalf("membershipSettleDelay() = %s, want 0", got) - } - }) - - t.Run("upscale waits for raft membership", func(t *testing.T) { - cr := chkWithHosts(3) - cr.SetAncestor(chkWithHosts(1)) - if got := w.membershipSettleDelay(cr); got != 30*time.Second { - t.Fatalf("membershipSettleDelay() = %s, want 30s", got) - } - }) - - t.Run("downscale always waits 120s", func(t *testing.T) { - cr := chkWithHosts(1) - cr.SetAncestor(chkWithHosts(3)) - if got := w.membershipSettleDelay(cr); got != 120*time.Second { - t.Fatalf("membershipSettleDelay() = %s, want 120s", got) - } - }) -} - func TestPersistReconcileCompleted(t *testing.T) { target := &apiChk.ClickHouseKeeperInstallation{ ObjectMeta: meta.ObjectMeta{Namespace: "test", Name: "keeper"}, @@ -155,45 +125,3 @@ func TestMarkReconcileStartReturnsStatusUpdateError(t *testing.T) { t.Fatalf("status = %q, want %q before persistence attempt", got, api.StatusInProgress) } } - -func TestPrepareStsReconcileOptsWaitSection(t *testing.T) { - w := &worker{a: a.NewAnnouncer(nil, nil)} - - t.Run("bootstrap skips Ready", func(t *testing.T) { - host := hostOnCR(chkWithHosts(3)) - opts := w.prepareStsReconcileOptsWaitSection(host, nil, false) - if !opts.WaitUntilStarted() || opts.WaitUntilReady() { - t.Fatal("bootstrap should wait Started only") - } - }) - - t.Run("rolling waits Ready", func(t *testing.T) { - host := hostOnCR(chkWithHosts(3)) - opts := w.prepareStsReconcileOptsWaitSection(host, nil, true) - if !opts.WaitUntilReady() { - t.Fatal("rolling should wait Ready") - } - }) - - t.Run("rolling can opt out of Ready probe", func(t *testing.T) { - host := hostOnCR(chkWithHosts(3)) - host.GetCluster().GetReconcile().Host.Wait.Probes.Readiness = types.NewStringBool(false) - opts := w.prepareStsReconcileOptsWaitSection(host, statefulset.NewReconcileStatefulSetOptions(), true) - if opts.WaitUntilReady() { - t.Fatal("readiness=false should skip Ready wait") - } - }) - - t.Run("single-host post-restart still waits Ready", func(t *testing.T) { - w.countReadyEnsembleMembersFn = func(context.Context, api.ICustomResource) int { return 0 } - host := hostOnCR(chkWithHosts(1)) - snap := w.snapshotHostEnsemble(context.Background(), host) - if !snap.rolling { - t.Fatal("single host should be rolling") - } - opts := w.prepareStsReconcileOptsWaitSection(host, nil, snap.rolling) - if !opts.WaitUntilReady() { - t.Fatal("rolling snapshot must drive Ready wait after force-restart") - } - }) -} diff --git a/pkg/controller/chk/worker-reconciler-helper.go b/pkg/controller/chk/worker-reconciler-helper.go index 3bdcd9a9f..7d3314a4b 100644 --- a/pkg/controller/chk/worker-reconciler-helper.go +++ b/pkg/controller/chk/worker-reconciler-helper.go @@ -53,6 +53,21 @@ func (w *worker) reconcileShardsAndHostsFetchOpts(ctx context.Context) *common.R } } +// noteCRUDResult records ErrCRUDDeferred on deferred and returns hard errors only. +// Soft deferred results return nil so the caller can continue other hosts/shards. +func noteCRUDResult(err error, deferred *bool) error { + if err == nil { + return nil + } + if errors.Is(err, common.ErrCRUDDeferred) { + if deferred != nil { + *deferred = true + } + return nil + } + return err +} + func (w *worker) runConcurrently(ctx context.Context, workersNum int, startShardIndex int, shards []*apiChk.ChkShard) error { if len(shards) == 0 { return nil @@ -91,10 +106,8 @@ func (w *worker) runConcurrently(ctx context.Context, workersNum int, startShard w.a.V(1).Info("Starting shard index: %d on worker", rq.index) if e := w.reconcileShardWithHosts(ctx, rq.shard); e != nil { errLock.Lock() - if errors.Is(e, common.ErrCRUDDeferred) { - deferred = true - } else { - err = e + if hard := noteCRUDResult(e, &deferred); hard != nil { + err = hard } errLock.Unlock() } diff --git a/pkg/controller/chk/worker.go b/pkg/controller/chk/worker.go index f56c8148b..6ef8aad89 100644 --- a/pkg/controller/chk/worker.go +++ b/pkg/controller/chk/worker.go @@ -60,7 +60,7 @@ type worker struct { countReadyEnsembleMembersFn func(ctx context.Context, cr api.ICustomResource) int // quorumDisruptPollOverride / quorumDisruptWaitOverride override pacing in - // waitForQuorumSafeToDisruptHost (tests only). Zero means use defaults. + // ensureQuorumSafeToDisruptHost (tests only). Zero means use defaults. quorumDisruptPollOverride time.Duration quorumDisruptWaitOverride time.Duration