From 9a9e32716d2753baed207a359da8e53f18c4ec73 Mon Sep 17 00:00:00 2001 From: Parth Bhardwaj <196071556+bhardwajparth51@users.noreply.github.com> Date: Fri, 24 Jul 2026 21:03:09 +0530 Subject: [PATCH] fix(controller): wait for replica inactivity and retry SYSTEM DROP REPLICA on Code 305 (#1943) Fixes issue #1943 where scaling down replicas causes SYSTEM DROP REPLICA to fail with Code: 305, Can't drop replica ... because it's active when executed before ZooKeeper session timeout expires. - Extracts ISchemer interface for dependency injection and testability. - Adds waitHostReplicaInactive polling until ZooKeeper session expires. - Adds retryDropReplica with linear backoff on Code 305 errors. - Adds getCoordinatorHost fallback helper for single-replica shard drops. - Adds comprehensive unit test suite in worker-deleter_test.go. Signed-off-by: Parth Bhardwaj <196071556+bhardwajparth51@users.noreply.github.com> --- pkg/controller/chi/worker-deleter.go | 155 +++++++++- pkg/controller/chi/worker-deleter_test.go | 340 ++++++++++++++++++++++ pkg/controller/chi/worker-migrator.go | 5 +- pkg/controller/chi/worker.go | 2 +- pkg/model/chi/schemer/schemer.go | 17 ++ 5 files changed, 510 insertions(+), 9 deletions(-) create mode 100644 pkg/controller/chi/worker-deleter_test.go diff --git a/pkg/controller/chi/worker-deleter.go b/pkg/controller/chi/worker-deleter.go index 9b1e38179..7bcb1d0df 100644 --- a/pkg/controller/chi/worker-deleter.go +++ b/pkg/controller/chi/worker-deleter.go @@ -16,6 +16,7 @@ package chi import ( "context" + "strings" "sync" "time" @@ -450,21 +451,125 @@ func (a dropReplicaOptionsArr) First() *dropReplicaOptions { return nil } +const ( + // maxDropReplicaRetries defines maximum attempts for SYSTEM DROP REPLICA on active replica errors (Code 305) + maxDropReplicaRetries = 6 + + // defaultRetryDelay specifies base backoff delay between retry attempts + defaultRetryDelay = 5 * time.Second + + // defaultReplicaWaitTimeout is used when ZooKeeper session timeout is unconfigured + defaultReplicaWaitTimeout = 120 * time.Second + + // defaultReplicaPollInterval is the frequency to check replica activity in ClickHouse + defaultReplicaPollInterval = 5 * time.Second + + // replicaWaitGracePeriod gives ZooKeeper extra time after session expiration before giving up + replicaWaitGracePeriod = 30 * time.Second +) + +// isReplicaActiveError targets ClickHouse Code 305 ("replica active"), caused by lingering ZooKeeper sessions. +func isReplicaActiveError(err error) bool { + if err == nil { + return false + } + s := err.Error() + return strings.Contains(s, "Code: 305") || + strings.Contains(s, "because it's active") || + strings.Contains(s, "Replica is active") +} + +// getCoordinatorHost finds an operational host to execute SYSTEM DROP REPLICA queries against. +func (w *worker) getCoordinatorHost(hostToDrop *api.Host) *api.Host { + if hostToDrop == nil || !hostToDrop.HasCR() { + return nil + } + if shard := hostToDrop.GetShard(); shard != nil { + if h := shard.FirstHost(); h != nil { + return h + } + } + // Fallback: any alive host in the cluster if the target host's shard is unavailable + if cluster := hostToDrop.GetCluster(); cluster != nil { + var candidate *api.Host + cluster.WalkHosts(func(h *api.Host) error { + if candidate == nil { + candidate = h + } + return nil + }) + return candidate + } + return nil +} + +// waitHostReplicaInactive polls until ZooKeeper session expires and ClickHouse reports host inactive. +func (w *worker) waitHostReplicaInactive(ctx context.Context, hostToRunOn, hostToDrop *api.Host, pollInterval, waitTimeout time.Duration) error { + if util.IsContextDone(ctx) { + return ctx.Err() + } + + if hostToDrop == nil { + return nil + } + + timeout := defaultReplicaWaitTimeout + if hostToDrop.HasCR() && hostToDrop.GetCluster() != nil { + if zk := hostToDrop.GetZookeeper(); zk != nil && zk.SessionTimeoutMs > 0 { + timeout = time.Duration(zk.SessionTimeoutMs)*time.Millisecond + replicaWaitGracePeriod + } + } + if waitTimeout > 0 { + timeout = waitTimeout + } + + if pollInterval <= 0 { + pollInterval = defaultReplicaPollInterval + } + + w.a.V(2).M(hostToRunOn).F().Info("Wait for host %s to become inactive", hostToDrop.GetName()) + + timeoutCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + schemer := w.ensureClusterSchemer(hostToRunOn) + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + + for { + if util.IsContextDone(timeoutCtx) { + return timeoutCtx.Err() + } + + if !schemer.IsHostActiveReplica(timeoutCtx, hostToRunOn, hostToDrop) { + w.a.V(2).M(hostToRunOn).F().Info("Host %s is inactive", hostToDrop.GetName()) + return nil + } + + select { + case <-timeoutCtx.Done(): + return timeoutCtx.Err() + case <-ticker.C: + } + } +} + // dropZKReplica drops replica's info from Zookeeper func (w *worker) dropZKReplica(ctx context.Context, hostToDrop *api.Host, opts *dropReplicaOptions) error { if hostToDrop == nil { - w.a.V(1).F().Error("FAILED to drop replica. Need to have host to drop. hostToDrop: %s", hostToDrop.GetName()) + w.a.V(1).F().Error("FAILED to drop replica. Need host to drop.") return nil } - // Sometimes host to drop is already unavailable, so let's run SQL statement of the first replica in the shard - var hostToRunOn *api.Host - if shard := hostToDrop.GetShard(); shard != nil { - hostToRunOn = shard.FirstHost() + if !hostToDrop.HasCR() { + w.a.V(1).F().Error("FAILED to drop replica. Host has no CR, hostToDrop: %s", hostToDrop.GetName()) + return nil } + // Sometimes host to drop is already unavailable, so let's run SQL statement of the first replica in the shard or cluster candidate + hostToRunOn := w.getCoordinatorHost(hostToDrop) if hostToRunOn == nil { - w.a.V(1).F().Error("FAILED to drop replica. hostToRunOn: %s, hostToDrop: %s", hostToRunOn.GetName(), hostToDrop.GetName()) + w.a.V(1).F().Error("FAILED to drop replica. hostToRunOn is nil, hostToDrop: %s", hostToDrop.GetName()) return nil } @@ -473,7 +578,11 @@ func (w *worker) dropZKReplica(ctx context.Context, hostToDrop *api.Host, opts * return nil } - err := w.ensureClusterSchemer(hostToRunOn).HostDropReplica(ctx, hostToRunOn, hostToDrop) + if err := w.waitHostReplicaInactive(ctx, hostToRunOn, hostToDrop, 0, 0); err != nil { + w.a.V(2).M(hostToRunOn).F().Warning("Replica %s still active after wait: %v", hostToDrop.GetName(), err) + } + + err := w.retryDropReplica(ctx, hostToRunOn, hostToDrop, 0) if err == nil { w.a.V(1). @@ -491,6 +600,38 @@ func (w *worker) dropZKReplica(ctx context.Context, hostToDrop *api.Host, opts * return err } +func (w *worker) retryDropReplica(ctx context.Context, hostToRunOn, hostToDrop *api.Host, baseDelay time.Duration) error { + schemer := w.ensureClusterSchemer(hostToRunOn) + var err error + + if baseDelay <= 0 { + baseDelay = defaultRetryDelay + } + + for try := 1; try <= maxDropReplicaRetries; try++ { + if util.IsContextDone(ctx) { + err = ctx.Err() + break + } + + err = schemer.HostDropReplica(ctx, hostToRunOn, hostToDrop) + if err == nil { + break + } + + // Retry if replica is still active + if isReplicaActiveError(err) && try < maxDropReplicaRetries { + delay := time.Duration(try) * baseDelay + w.a.V(1).M(hostToRunOn).F().Warning("Retrying SYSTEM DROP REPLICA for host %s (attempt %d/%d), backoff %v: %v", hostToDrop.GetName(), try, maxDropReplicaRetries, delay, err) + util.WaitContextDoneOrTimeout(ctx, delay) + } else { + break + } + } + + return err +} + // deleteTables func (w *worker) deleteTables(ctx context.Context, host *api.Host) error { if util.IsContextDone(ctx) { diff --git a/pkg/controller/chi/worker-deleter_test.go b/pkg/controller/chi/worker-deleter_test.go new file mode 100644 index 000000000..18235bc73 --- /dev/null +++ b/pkg/controller/chi/worker-deleter_test.go @@ -0,0 +1,340 @@ +// 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 chi + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + + api "github.com/altinity/clickhouse-operator/pkg/apis/clickhouse.altinity.com/v1" + "github.com/altinity/clickhouse-operator/pkg/model/clickhouse" +) + +type fakeSchemer struct { + isHostActiveReplicaFunc func(ctx context.Context, hostToRunOn, hostToCheck *api.Host) bool + hostDropReplicaFunc func(ctx context.Context, hostToRunOn, hostToDrop *api.Host, calls int) error + + activeCalls int + dropCalls int +} + +func (f *fakeSchemer) IsHostActiveReplica(ctx context.Context, hostToRunOn, hostToCheck *api.Host) bool { + f.activeCalls++ + if f.isHostActiveReplicaFunc != nil { + return f.isHostActiveReplicaFunc(ctx, hostToRunOn, hostToCheck) + } + return false +} + +func (f *fakeSchemer) HostDropReplica(ctx context.Context, hostToRunOn, hostToDrop *api.Host) error { + f.dropCalls++ + if f.hostDropReplicaFunc != nil { + return f.hostDropReplicaFunc(ctx, hostToRunOn, hostToDrop, f.dropCalls) + } + return nil +} + +func (f *fakeSchemer) HostSyncTables(ctx context.Context, host *api.Host) error { return nil } +func (f *fakeSchemer) HostCreateTables(ctx context.Context, host *api.Host) error { return nil } +func (f *fakeSchemer) HostDropTables(ctx context.Context, host *api.Host) error { return nil } +func (f *fakeSchemer) IsHostInCluster(ctx context.Context, host *api.Host) bool { return true } +func (f *fakeSchemer) HostActiveQueriesNum(ctx context.Context, host *api.Host) (int, error) { return 0, nil } +func (f *fakeSchemer) HostClickHouseVersion(ctx context.Context, host *api.Host) (string, error) { return "", nil } +func (f *fakeSchemer) HostMaxReplicaDelay(ctx context.Context, host *api.Host) (int, error) { return 0, nil } +func (f *fakeSchemer) HostShutdown(ctx context.Context, host *api.Host) error { return nil } +func (f *fakeSchemer) ExecHost(ctx context.Context, host *api.Host, SQLs []string, _opts ...*clickhouse.QueryOptions) error { return nil } +func (f *fakeSchemer) ExecCluster(ctx context.Context, cluster *api.Cluster, SQLs []string, _opts ...*clickhouse.QueryOptions) error { return nil } +func (f *fakeSchemer) HostClusterDoesNotExistErrorCount(ctx context.Context, host *api.Host) (int, error) { return 0, nil } + +func TestWaitHostReplicaInactive_NilHost(t *testing.T) { + w := &worker{} + err := w.waitHostReplicaInactive(context.Background(), &api.Host{}, nil, 0, 0) + assert.NoError(t, err) +} + +func TestWaitHostReplicaInactive_ContextDone(t *testing.T) { + w := &worker{} + hostToRunOn := &api.Host{} + hostToDrop := &api.Host{} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel context immediately + + err := w.waitHostReplicaInactive(ctx, hostToRunOn, hostToDrop, 1*time.Millisecond, 0) + assert.Equal(t, context.Canceled, err) +} + +func TestIsReplicaActiveError(t *testing.T) { + err305 := errors.New("Code: 305, Can't drop replica chi-dev-0-1, because it's active") + errTextVariant1 := errors.New("Can't drop replica chi-dev-0-1, because it's active") + errTextVariant2 := errors.New("DB::Exception: Replica is active") + errOther := errors.New("Code: 192, Unknown table") + + assert.True(t, isReplicaActiveError(err305)) + assert.True(t, isReplicaActiveError(errTextVariant1)) + assert.True(t, isReplicaActiveError(errTextVariant2)) + assert.False(t, isReplicaActiveError(errOther)) + assert.False(t, isReplicaActiveError(nil)) +} + +func TestGetCoordinatorHost(t *testing.T) { + w := &worker{} + assert.Nil(t, w.getCoordinatorHost(nil)) + + hostNoShard := &api.Host{} + assert.Nil(t, w.getCoordinatorHost(hostNoShard)) +} + +func TestWaitHostReplicaInactive_ImmediateSuccess(t *testing.T) { + fake := &fakeSchemer{ + isHostActiveReplicaFunc: func(ctx context.Context, hostToRunOn, hostToCheck *api.Host) bool { + return false + }, + } + w := &worker{schemer: fake} + err := w.waitHostReplicaInactive(context.Background(), &api.Host{}, &api.Host{}, 1*time.Millisecond, 0) + assert.NoError(t, err) + assert.Equal(t, 1, fake.activeCalls) +} + +func TestWaitHostReplicaInactive_PollingTransitions(t *testing.T) { + calls := 0 + fake := &fakeSchemer{ + isHostActiveReplicaFunc: func(ctx context.Context, hostToRunOn, hostToCheck *api.Host) bool { + calls++ + return calls < 2 // Active on call 1, inactive on call 2 + }, + } + + w := &worker{schemer: fake} + hostToRunOn := &api.Host{} + hostToDrop := &api.Host{} + + err := w.waitHostReplicaInactive(context.Background(), hostToRunOn, hostToDrop, 1*time.Millisecond, 0) + assert.NoError(t, err) + assert.Equal(t, 2, fake.activeCalls) +} + +func TestWaitHostReplicaInactive_ParentContextExpires(t *testing.T) { + fake := &fakeSchemer{ + isHostActiveReplicaFunc: func(ctx context.Context, hostToRunOn, hostToCheck *api.Host) bool { + return true + }, + } + w := &worker{schemer: fake} + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) + defer cancel() + + err := w.waitHostReplicaInactive(ctx, &api.Host{}, &api.Host{}, 1*time.Millisecond, 0) + assert.Error(t, err) + assert.True(t, errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled)) +} + +func TestWaitHostReplicaInactive_InternalTimeout(t *testing.T) { + fake := &fakeSchemer{ + isHostActiveReplicaFunc: func(ctx context.Context, hostToRunOn, hostToCheck *api.Host) bool { + return true + }, + } + w := &worker{schemer: fake} + + // Test waitHostReplicaInactive internal timeout branch directly by specifying 10ms waitTimeout + err := w.waitHostReplicaInactive(context.Background(), &api.Host{}, &api.Host{}, 1*time.Millisecond, 10*time.Millisecond) + assert.Error(t, err) + assert.True(t, errors.Is(err, context.DeadlineExceeded)) +} + +func TestWaitTimeoutProceedsToRetryAndSucceeds(t *testing.T) { + fake := &fakeSchemer{ + isHostActiveReplicaFunc: func(ctx context.Context, hostToRunOn, hostToCheck *api.Host) bool { + return true // Always active during wait phase -> waitHostReplicaInactive times out + }, + hostDropReplicaFunc: func(ctx context.Context, hostToRunOn, hostToDrop *api.Host, calls int) error { + if calls == 1 { + return errors.New("Code: 305, Can't drop replica, because it's active") + } + return nil // Succeeds on attempt 2 + }, + } + + w := &worker{schemer: fake} + hostToRunOn := &api.Host{} + hostToDrop := &api.Host{} + + // Step 1: waitHostReplicaInactive times out after 10ms + errWait := w.waitHostReplicaInactive(context.Background(), hostToRunOn, hostToDrop, 1*time.Millisecond, 10*time.Millisecond) + assert.Error(t, errWait) + assert.True(t, errors.Is(errWait, context.DeadlineExceeded)) + + // Step 2: retryDropReplica proceeds despite wait timeout and succeeds on attempt 2 + errRetry := w.retryDropReplica(context.Background(), hostToRunOn, hostToDrop, 1*time.Millisecond) + assert.NoError(t, errRetry) + assert.Equal(t, 2, fake.dropCalls) +} + +func TestDropZKReplica_NilGuards(t *testing.T) { + w := &worker{} + assert.Nil(t, w.dropZKReplica(context.Background(), nil, NewDropReplicaOptions())) + + hostToDropNoShard := &api.Host{} + assert.Nil(t, w.dropZKReplica(context.Background(), hostToDropNoShard, NewDropReplicaOptions())) +} + +func TestDropReplicaOptions(t *testing.T) { + opts := NewDropReplicaOptions() + assert.False(t, opts.RegularDrop()) + assert.False(t, opts.ForceDropUponStorageLoss()) + + opts.SetRegularDrop() + assert.True(t, opts.RegularDrop()) + + opts.SetForceDropUponStorageLoss() + assert.True(t, opts.ForceDropUponStorageLoss()) + + var nilOpts *dropReplicaOptions + assert.False(t, nilOpts.RegularDrop()) + assert.False(t, nilOpts.ForceDropUponStorageLoss()) + assert.Nil(t, nilOpts.SetRegularDrop()) + assert.Nil(t, nilOpts.SetForceDropUponStorageLoss()) + + arr := NewDropReplicaOptionsArr(opts) + assert.Equal(t, opts, arr.First()) + + emptyArr := NewDropReplicaOptionsArr() + assert.Nil(t, emptyArr.First()) +} + +func TestRetryDropReplica_SuccessOnFirstTry(t *testing.T) { + fake := &fakeSchemer{ + hostDropReplicaFunc: func(ctx context.Context, hostToRunOn, hostToDrop *api.Host, calls int) error { + return nil + }, + } + w := &worker{schemer: fake} + err := w.retryDropReplica(context.Background(), &api.Host{}, &api.Host{}, 1*time.Millisecond) + assert.NoError(t, err) + assert.Equal(t, 1, fake.dropCalls) +} + +func TestRetryDropReplica_RetriesOn305ThenSucceeds(t *testing.T) { + fake := &fakeSchemer{ + hostDropReplicaFunc: func(ctx context.Context, hostToRunOn, hostToDrop *api.Host, calls int) error { + if calls == 1 { + return errors.New("Code: 305, Can't drop replica, because it's active") + } + return nil + }, + } + + w := &worker{schemer: fake} + hostToRunOn := &api.Host{} + hostToDrop := &api.Host{} + ctx := context.Background() + + err := w.retryDropReplica(ctx, hostToRunOn, hostToDrop, 1*time.Millisecond) + + assert.NoError(t, err) + assert.Equal(t, 2, fake.dropCalls) +} + +func TestRetryDropReplica_ExhaustsRetriesReturnsLastError(t *testing.T) { + fake := &fakeSchemer{ + hostDropReplicaFunc: func(ctx context.Context, hostToRunOn, hostToDrop *api.Host, calls int) error { + return errors.New("Code: 305, Can't drop replica, because it's active") + }, + } + + w := &worker{schemer: fake} + hostToRunOn := &api.Host{} + hostToDrop := &api.Host{} + ctx := context.Background() + + err := w.retryDropReplica(ctx, hostToRunOn, hostToDrop, 1*time.Millisecond) + + assert.Error(t, err) + assert.True(t, isReplicaActiveError(err)) + assert.Equal(t, 6, fake.dropCalls) +} + +func TestRetryDropReplica_NonRetryableErrorFailsFast(t *testing.T) { + fake := &fakeSchemer{ + hostDropReplicaFunc: func(ctx context.Context, hostToRunOn, hostToDrop *api.Host, calls int) error { + return errors.New("Code: 192, Unknown table") + }, + } + + w := &worker{schemer: fake} + hostToRunOn := &api.Host{} + hostToDrop := &api.Host{} + ctx := context.Background() + + err := w.retryDropReplica(ctx, hostToRunOn, hostToDrop, 1*time.Millisecond) + + assert.Error(t, err) + assert.False(t, isReplicaActiveError(err)) + assert.Equal(t, 1, fake.dropCalls) +} + +func TestRetryDropReplica_ContextCancelledMidRetry_ReturnsErrNotNil(t *testing.T) { + fake := &fakeSchemer{ + hostDropReplicaFunc: func(ctx context.Context, hostToRunOn, hostToDrop *api.Host, calls int) error { + return errors.New("Code: 305, Can't drop replica, because it's active") + }, + } + + w := &worker{schemer: fake} + hostToRunOn := &api.Host{} + hostToDrop := &api.Host{} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel before loop + + err := w.retryDropReplica(ctx, hostToRunOn, hostToDrop, 1*time.Millisecond) + + assert.Error(t, err) + assert.Equal(t, context.Canceled, err) + assert.Equal(t, 0, fake.dropCalls) +} + +func TestRetryDropReplica_ContextCancelledDuringRetries(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + fake := &fakeSchemer{ + hostDropReplicaFunc: func(c context.Context, hostToRunOn, hostToDrop *api.Host, calls int) error { + cancel() // Cancel context during first retry attempt + return errors.New("Code: 305, Can't drop replica, because it's active") + }, + } + w := &worker{schemer: fake} + err := w.retryDropReplica(ctx, &api.Host{}, &api.Host{}, 1*time.Millisecond) + assert.Error(t, err) + assert.Equal(t, context.Canceled, err) + assert.Equal(t, 1, fake.dropCalls) +} + +func TestDefaultFallbackParameters(t *testing.T) { + w := &worker{schemer: &fakeSchemer{}} + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel immediately + + err1 := w.waitHostReplicaInactive(ctx, &api.Host{}, &api.Host{}, 0, 0) + assert.Error(t, err1) + + err2 := w.retryDropReplica(ctx, &api.Host{}, &api.Host{}, 0) + assert.Error(t, err2) +} diff --git a/pkg/controller/chi/worker-migrator.go b/pkg/controller/chi/worker-migrator.go index 1e1a8d8f4..4b2b38507 100644 --- a/pkg/controller/chi/worker-migrator.go +++ b/pkg/controller/chi/worker-migrator.go @@ -154,10 +154,13 @@ func (w *worker) shouldMigrateTables(host *api.Host, opts ...*migrateTableOption return true } -func (w *worker) ensureClusterSchemer(host *api.Host) *schemer.ClusterSchemer { +func (w *worker) ensureClusterSchemer(host *api.Host) schemer.ISchemer { if w == nil { return nil } + if w.schemer != nil { + return w.schemer + } // Make base cluster connection params from CHOP-config defaults, then // overlay the per-cluster security.clickhouse.tls fields populated by the // normalizer (3-level inheritance: CHOP-config → CHI → cluster). Without diff --git a/pkg/controller/chi/worker.go b/pkg/controller/chi/worker.go index 77e1eed29..f4c08708e 100644 --- a/pkg/controller/chi/worker.go +++ b/pkg/controller/chi/worker.go @@ -59,7 +59,7 @@ type worker struct { //queue workqueue.RateLimitingInterface queue queue.PriorityQueue - schemer *schemer.ClusterSchemer + schemer schemer.ISchemer normalizer *normalizer.Normalizer task *common.Task diff --git a/pkg/model/chi/schemer/schemer.go b/pkg/model/chi/schemer/schemer.go index 8e5273714..4e23bf1ed 100644 --- a/pkg/model/chi/schemer/schemer.go +++ b/pkg/model/chi/schemer/schemer.go @@ -27,6 +27,23 @@ import ( "github.com/altinity/clickhouse-operator/pkg/util" ) +// ISchemer defines cluster schema manager interface +type ISchemer interface { + HostSyncTables(ctx context.Context, host *api.Host) error + IsHostActiveReplica(ctx context.Context, hostToRunOn, hostToCheck *api.Host) bool + HostDropReplica(ctx context.Context, hostToRunOn, hostToDrop *api.Host) error + HostCreateTables(ctx context.Context, host *api.Host) error + HostDropTables(ctx context.Context, host *api.Host) error + IsHostInCluster(ctx context.Context, host *api.Host) bool + HostActiveQueriesNum(ctx context.Context, host *api.Host) (int, error) + HostClickHouseVersion(ctx context.Context, host *api.Host) (string, error) + HostMaxReplicaDelay(ctx context.Context, host *api.Host) (int, error) + HostShutdown(ctx context.Context, host *api.Host) error + ExecHost(ctx context.Context, host *api.Host, SQLs []string, _opts ...*clickhouse.QueryOptions) error + ExecCluster(ctx context.Context, cluster *api.Cluster, SQLs []string, _opts ...*clickhouse.QueryOptions) error + HostClusterDoesNotExistErrorCount(ctx context.Context, host *api.Host) (int, error) +} + // ClusterSchemer specifies cluster schema manager type ClusterSchemer struct { *Cluster