From a6d2104833140f711dd680afe842837d83130c14 Mon Sep 17 00:00:00 2001 From: theluckystrike <51033404+theluckystrike@users.noreply.github.com> Date: Sun, 8 Nov 2026 03:04:15 +0100 Subject: [PATCH] fix(schemer): fail table migration when the peer schema cannot be read queryUnzipColumns returned nil when QueryAny failed on every peer, debugCreateSQLs dropped the error and createTablesSQLs ignored it, so HostCreateTables ran no DDL and returned nil. migrateTables then recorded the host in hostsWithTablesCreated and never migrated it again. Propagate the error up to HostCreateTables so migrateTables defers the host and retries the migration. Fixes #2129 Signed-off-by: theluckystrike <51033404+theluckystrike@users.noreply.github.com> --- pkg/model/chi/schemer/cluster.go | 2 +- pkg/model/chi/schemer/distributed.go | 15 +++- pkg/model/chi/schemer/replicated.go | 15 +++- pkg/model/chi/schemer/schemer.go | 24 ++++-- pkg/model/chi/schemer/schemer_test.go | 116 ++++++++++++++++++++++++++ 5 files changed, 156 insertions(+), 16 deletions(-) create mode 100644 pkg/model/chi/schemer/schemer_test.go diff --git a/pkg/model/chi/schemer/cluster.go b/pkg/model/chi/schemer/cluster.go index 21510d42a..259cde7d7 100644 --- a/pkg/model/chi/schemer/cluster.go +++ b/pkg/model/chi/schemer/cluster.go @@ -64,7 +64,7 @@ func (c *Cluster) queryUnzipColumns(ctx context.Context, hosts []string, sql str // Fetch data from any of specified hosts query, err := c.SetHosts(hosts).QueryAny(ctx, sql) if err != nil { - return nil + return err } if query == nil { return nil diff --git a/pkg/model/chi/schemer/distributed.go b/pkg/model/chi/schemer/distributed.go index fe06870bd..411b7c90b 100644 --- a/pkg/model/chi/schemer/distributed.go +++ b/pkg/model/chi/schemer/distributed.go @@ -56,27 +56,36 @@ func (s *ClusterSchemer) getDistributedObjectsSQLs(ctx context.Context, host *ap // Exclude self — see the matching comment in getReplicatedObjectsSQLs. peers := s.Names(interfaces.NameFQDNs, host, api.Cluster{}, true) - databaseNames, createDatabaseSQLs := debugCreateSQLs( + databaseNames, createDatabaseSQLs, err := debugCreateSQLs( s.QueryUnzip2Columns( ctx, peers, s.sqlCreateDatabaseDistributed(host.Runtime.Address.ClusterName), ), ) - tableNames, createTableSQLs := debugCreateSQLs( + if err != nil { + return nil, nil, err + } + tableNames, createTableSQLs, err := debugCreateSQLs( s.QueryUnzipAndApplyUUIDs( ctx, peers, s.sqlCreateTableDistributed(host.Runtime.Address.ClusterName), ), ) - functionNames, createFunctionSQLs := debugCreateSQLs( + if err != nil { + return nil, nil, err + } + functionNames, createFunctionSQLs, err := debugCreateSQLs( s.QueryUnzip2Columns( ctx, peers, s.sqlCreateFunction(host.Runtime.Address.ClusterName), ), ) + if err != nil { + return nil, nil, err + } return util.ConcatSlices([][]string{databaseNames, tableNames, functionNames}), util.ConcatSlices([][]string{createDatabaseSQLs, createTableSQLs, createFunctionSQLs}), nil diff --git a/pkg/model/chi/schemer/replicated.go b/pkg/model/chi/schemer/replicated.go index 1a630eafc..883f502d7 100644 --- a/pkg/model/chi/schemer/replicated.go +++ b/pkg/model/chi/schemer/replicated.go @@ -71,27 +71,36 @@ func (s *ClusterSchemer) getReplicatedObjectsSQLs(ctx context.Context, host *api // schemer silently skips CREATE. Peers hold the truth. peers := s.Names(interfaces.NameFQDNs, host, api.Cluster{}, true) - databaseNames, createDatabaseSQLs := debugCreateSQLs( + databaseNames, createDatabaseSQLs, err := debugCreateSQLs( s.QueryUnzip2Columns( ctx, peers, s.sqlCreateDatabaseReplicated(host.Runtime.Address.ClusterName), ), ) - tableNames, createTableSQLs := debugCreateSQLs( + if err != nil { + return nil, nil, err + } + tableNames, createTableSQLs, err := debugCreateSQLs( s.QueryUnzipAndApplyUUIDs( ctx, peers, s.sqlCreateTableReplicated(host.Runtime.Address.ClusterName), ), ) - functionNames, createFunctionSQLs := debugCreateSQLs( + if err != nil { + return nil, nil, err + } + functionNames, createFunctionSQLs, err := debugCreateSQLs( s.QueryUnzip2Columns( ctx, peers, s.sqlCreateFunction(host.Runtime.Address.ClusterName), ), ) + if err != nil { + return nil, nil, err + } return util.ConcatSlices([][]string{databaseNames, tableNames, functionNames}), util.ConcatSlices([][]string{createDatabaseSQLs, createTableSQLs, createFunctionSQLs}), nil diff --git a/pkg/model/chi/schemer/schemer.go b/pkg/model/chi/schemer/schemer.go index eaaad74e6..fa2bf1f52 100644 --- a/pkg/model/chi/schemer/schemer.go +++ b/pkg/model/chi/schemer/schemer.go @@ -17,6 +17,7 @@ package schemer import ( "context" "errors" + "fmt" "time" log "github.com/altinity/clickhouse-operator/pkg/announcer" @@ -87,14 +88,13 @@ func (s *ClusterSchemer) createTablesSQLs( replicatedCreateSQLs []string, distributedObjectNames []string, distributedCreateSQLs []string, + err error, ) { - if names, sql, err := s.getReplicatedObjectsSQLs(ctx, host); err == nil { - replicatedObjectNames = names - replicatedCreateSQLs = sql + if replicatedObjectNames, replicatedCreateSQLs, err = s.getReplicatedObjectsSQLs(ctx, host); err != nil { + return nil, nil, nil, nil, err } - if names, sql, err := s.getDistributedObjectsSQLs(ctx, host); err == nil { - distributedObjectNames = names - distributedCreateSQLs = sql + if distributedObjectNames, distributedCreateSQLs, err = s.getDistributedObjectsSQLs(ctx, host); err != nil { + return nil, nil, nil, nil, err } return } @@ -112,7 +112,13 @@ func (s *ClusterSchemer) HostCreateTables(ctx context.Context, host *api.Host) e replicatedObjectNames, replicatedCreateSQLs, distributedObjectNames, - distributedCreateSQLs := s.createTablesSQLs(ctx, host) + distributedCreateSQLs, + err := s.createTablesSQLs(ctx, host) + if err != nil { + // Without the peer schema there is nothing to create. Report it, so the host is not + // recorded as migrated and the migration is retried. + return fmt.Errorf("unable to read schema from peers of host %s: %w", host.Runtime.Address.HostName, err) + } var err1 error if len(replicatedCreateSQLs) > 0 { @@ -216,7 +222,7 @@ func (s *ClusterSchemer) HostShutdown(ctx context.Context, host *api.Host) error clickhouse.NewQueryOptions().SetRetry(false).SetLogQueries(true)) } -func debugCreateSQLs(names, sqls []string, err error) ([]string, []string) { +func debugCreateSQLs(names, sqls []string, err error) ([]string, []string, error) { if err != nil { log.V(1).Warning("got error: %v", err) } @@ -228,5 +234,5 @@ func debugCreateSQLs(names, sqls []string, err error) ([]string, []string) { for _, v := range sqls { log.V(2).Info("sql: %s", v) } - return names, sqls + return names, sqls, err } diff --git a/pkg/model/chi/schemer/schemer_test.go b/pkg/model/chi/schemer/schemer_test.go new file mode 100644 index 000000000..a310cc458 --- /dev/null +++ b/pkg/model/chi/schemer/schemer_test.go @@ -0,0 +1,116 @@ +// 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. + +// The test lives outside package schemer because the normalizer imports schemer. +package schemer_test + +import ( + "context" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/require" + + 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/apis/swversion" + "github.com/altinity/clickhouse-operator/pkg/chop" + "github.com/altinity/clickhouse-operator/pkg/model/chi/normalizer" + "github.com/altinity/clickhouse-operator/pkg/model/chi/schemer" + "github.com/altinity/clickhouse-operator/pkg/model/clickhouse" + commonNormalizer "github.com/altinity/clickhouse-operator/pkg/model/common/normalizer" +) + +// The normalizer and the namer resolve configuration through the global chop singleton. +func TestMain(m *testing.M) { + os.Unsetenv("WATCH_NAMESPACE") + os.Unsetenv("WATCH_NAMESPACES") + chop.New(nil, nil, "") + os.Exit(m.Run()) +} + +func testConnectionParams(port int) *clickhouse.ClusterConnectionParams { + params := clickhouse.NewClusterConnectionParams("http", "clickhouse_operator", "password", "", port) + params.SetTimeouts(clickhouse.NewTimeouts(2*time.Second, 2*time.Second)) + return params +} + +// closedPort returns a local port that nothing listens on. +func closedPort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + port := l.Addr().(*net.TCPAddr).Port + require.NoError(t, l.Close()) + return port +} + +// A peer that rejects the operator user must surface as an error. Reporting it as an empty +// result makes the schemer think the cluster has no schema to copy. +func TestQueryUnzipReportsRejectedPeer(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusForbidden) + _, _ = w.Write([]byte("Code: 516. DB::Exception: clickhouse_operator: Authentication failed. (AUTHENTICATION_FAILED)\n")) + })) + defer srv.Close() + + u, err := url.Parse(srv.URL) + require.NoError(t, err) + port, err := strconv.Atoi(u.Port()) + require.NoError(t, err) + + cluster := schemer.NewCluster().SetClusterConnectionParams(testConnectionParams(port)) + peers := []string{u.Hostname()} + + names, sqls, err := cluster.QueryUnzip2Columns(context.Background(), peers, "SELECT name, create_query FROM system.tables") + require.Error(t, err) + require.Empty(t, names) + require.Empty(t, sqls) + + names, sqls, err = cluster.QueryUnzipAndApplyUUIDs(context.Background(), peers, "SELECT name, create_query, uuid, inner_uuid FROM system.tables") + require.Error(t, err) + require.Empty(t, names) + require.Empty(t, sqls) +} + +// HostCreateTables must fail when it cannot read the schema from any peer, so that migrateTables +// does not record the host in hostsWithTablesCreated. Before the fix it ran no DDL and returned nil. +func TestHostCreateTablesFailsWhenPeersCannotBeRead(t *testing.T) { + chi := &api.ClickHouseInstallation{} + chi.Name = "schema" + chi.Namespace = "ns" + // .invalid never resolves (RFC 6761), so every peer is unreachable without any network setup. + chi.Spec.NamespaceDomainPattern = types.NewString("%s.invalid") + chi.Spec.Configuration = &api.Configuration{ + Clusters: []*api.Cluster{{ + Name: "c1", + Layout: &api.ChiClusterLayout{ShardsCount: 1, ReplicasCount: 2}, + }}, + } + normalized, err := normalizer.New(nil).CreateTemplated(chi, commonNormalizer.NewOptions[api.ClickHouseInstallation]()) + require.NoError(t, err) + + host := normalized.FindHost("c1", 0, 1) + require.NotNil(t, host) + + s := schemer.NewClusterSchemer(testConnectionParams(closedPort(t)), swversion.NewSoftWareVersion("26.10.3.7")) + require.Error(t, s.HostCreateTables(context.Background(), host), + "a schema read failure must not be reported as a successful migration") +}