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/controller.go b/pkg/controller/chk/controller.go index bef2cd13f..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 @@ -112,7 +118,18 @@ 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 { + 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 + } return ctrl.Result{}, nil } 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 new file mode 100644 index 000000000..e2beb5da0 --- /dev/null +++ b/pkg/controller/chk/worker-raft-safety.go @@ -0,0 +1,362 @@ +// 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" + "time" + + 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" + 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 policy for CHK (#2069). +// +// 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. +// +// 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). + +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 +// 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 +} + +// 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() + ready := w.countReadyEnsembleMembers(ctx, cr) + return hostEnsembleSnapshot{ + rolling: n <= 1 || ready >= raftQuorumSize(n), + members: n, + readyCount: ready, + } +} + +// 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) + } +} + +func (w *worker) quorumDisruptPollInterval() time.Duration { + if w.quorumDisruptPollOverride > 0 { + return w.quorumDisruptPollOverride + } + return defaultQuorumDisruptPollInterval +} + +func (w *worker) quorumDisruptWaitTimeout() time.Duration { + if w.quorumDisruptWaitOverride > 0 { + return w.quorumDisruptWaitOverride + } + return defaultQuorumDisruptWaitTimeout +} + +// 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, + snap *hostEnsembleSnapshot, +) error { + if snap == nil || !snap.rolling || snap.members <= 1 { + return nil + } + if !w.hostDisruptionWouldBreakQuorum(ctx, host, opts, *snap) { + return nil + } + + 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 + } + } + + 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 + } + 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 +} + +// 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 +} + +// 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 + } + remaining := snap.readyCount + if hostContributesReady(host) { + remaining-- + } + return remaining >= raftQuorumSize(snap.members) +} + +// hostDisruptionWouldBreakQuorum is true when this pass would disrupt a Ready host and +// drop the ensemble below Raft quorum (#2069). +// +// Must be called after PrepareHostStatefulSetWithStatus — ObjectStatusSame is assigned only there. +func (w *worker) hostDisruptionWouldBreakQuorum( + ctx context.Context, + host *api.Host, + opts *statefulset.ReconcileOptions, + snap hostEnsembleSnapshot, +) bool { + if host == nil || host.IsStopped() { + return false + } + if host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) { + return false + } + willDisrupt := !host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusSame) || + w.shouldForceRestartHost(ctx, host) || + (opts != nil && opts.ForceRecreate()) + if !willDisrupt { + return false + } + return hostContributesReady(host) && !ensembleQuorumSafeAfterDisrupt(snap, 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 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 + _ = 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 new file mode 100644 index 000000000..f3609889d --- /dev/null +++ b/pkg/controller/chk/worker-raft-safety_test.go @@ -0,0 +1,362 @@ +// 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" + "sync/atomic" + "testing" + "time" + + "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 TestSnapshotHostEnsemble(t *testing.T) { + ctx := context.Background() + + 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)) + 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 is bootstrap", func(t *testing.T) { + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { return 0 }, + } + host := hostOnCR(chkWithHosts(3)) + 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 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)) + snap := w.snapshotHostEnsemble(ctx, host) + require.True(t, snap.rolling) + require.Equal(t, 2, snap.readyCount) + }) +} + +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-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("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("blocks disruptive roll without quorum headroom", func(t *testing.T) { + require.True(t, w.hostDisruptionWouldBreakQuorum(ctx, host, nil, snap)) + }) + + 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 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 TestEnsureQuorumSafeToDisruptHost(t *testing.T) { + ctx := context.Background() + 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("returns immediately when already safe", func(t *testing.T) { + w := &worker{} + safeSnap := hostEnsembleSnapshot{rolling: true, members: 3, readyCount: 3} + require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, host, nil, &safeSnap)) + }) + + t.Run("waits until ready count increases", func(t *testing.T) { + var ready atomic.Int32 + ready.Store(2) + w := &worker{ + countReadyEnsembleMembersFn: func(context.Context, api.ICustomResource) int { + return int(ready.Load()) + }, + quorumDisruptPollOverride: 5 * time.Millisecond, + quorumDisruptWaitOverride: 200 * time.Millisecond, + } + waitSnap := snap + go func() { + time.Sleep(20 * time.Millisecond) + ready.Store(3) + }() + require.NoError(t, w.ensureQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap)) + }) + + 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, + } + waitSnap := snap + err := w.ensureQuorumSafeToDisruptHost(ctx, host, nil, &waitSnap) + require.ErrorIs(t, err, common.ErrCRUDDeferred) + }) +} + +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 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 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() + 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..9b8b938bf 100644 --- a/pkg/controller/chk/worker-reconciler-chk.go +++ b/pkg/controller/chk/worker-reconciler-chk.go @@ -116,11 +116,28 @@ 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) 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). @@ -131,23 +148,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,15 +301,13 @@ 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 := 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 } @@ -406,21 +424,30 @@ 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()) - // Start with force-restart host + w.stsReconciler.PrepareHostStatefulSetWithStatus(ctx, host, host.IsStopped()) + opts = w.prepareStsReconcileOptsWaitSection(host, opts, snap.rolling) + if err := w.ensureQuorumSafeToDisruptHost(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 @@ -432,7 +459,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) } - opts = w.prepareStsReconcileOptsWaitSection(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()) @@ -600,6 +626,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 @@ -610,8 +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 { - 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 + } } // Since shard with 0 index is already done, we'll proceed concurrently starting with the 1-st @@ -623,10 +652,15 @@ 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 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 + } } w.a.V(1).Info("Finished successfully rest of shards on workers: %d", workersNum) + if deferred { + return common.ErrCRUDDeferred + } return nil } @@ -634,9 +668,22 @@ 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 := noteCRUDResult(w.reconcileHost(ctx, host), &deferred); err != nil { + return err + } + } + if deferred { + return common.ErrCRUDDeferred + } + return nil } // reconcileShard reconciles specified shard, excluding nested replicas @@ -786,8 +833,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(). @@ -802,36 +852,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 - if err := w.reconcileHostMainDomain(ctx, host); err != nil { + // 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, snap); 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, @@ -840,20 +874,20 @@ 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 { - // Should we wait for host to startup - wait := false - - if host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) { - wait = true +func (w *worker) reconcileHostMainDomain(ctx context.Context, host *api.Host, snap hostEnsembleSnapshot) error { + 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 !snap.rolling { + // Bootstrap / resume-from-stopped / recovery: peers start together; + // legacy pacing wait (Ready wait was skipped on STS). util.WaitContextDoneOrTimeout(ctx, 7*time.Second) + return nil } - return nil + + // Extension point for richer Raft membership confirmation (PR #2041). + return w.verifyHostEnsembleMembership(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..e11e35c90 --- /dev/null +++ b/pkg/controller/chk/worker-reconciler-chk_test.go @@ -0,0 +1,127 @@ +// 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" + + 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 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-reconciler-helper.go b/pkg/controller/chk/worker-reconciler-helper.go index d3e289f5e..7d3314a4b 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" @@ -52,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 @@ -80,6 +96,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 +106,9 @@ 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 hard := noteCRUDResult(e, &deferred); hard != nil { + err = hard + } errLock.Unlock() } } @@ -99,7 +118,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 61fa1b613..6ef8aad89 100644 --- a/pkg/controller/chk/worker.go +++ b/pkg/controller/chk/worker.go @@ -56,6 +56,14 @@ 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 + + // quorumDisruptPollOverride / quorumDisruptWaitOverride override pacing in + // ensureQuorumSafeToDisruptHost (tests only). Zero means use defaults. + quorumDisruptPollOverride time.Duration + quorumDisruptWaitOverride time.Duration + start time.Time } @@ -119,7 +127,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 +205,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 +252,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 +273,7 @@ func (w *worker) finalizeReconcileAndMarkCompleted(ctx context.Context, _cr *api WithActions(_cr). M(_cr).F(). Info("reconcile completed successfully, task id: %s", _cr.GetSpecT().GetTaskID()) + return nil } func (w *worker) markReconcileCompletedUnsuccessfully(ctx context.Context, cr *apiChk.ClickHouseKeeperInstallation, err error) { diff --git a/tests/e2e/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"): 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")], }