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") +}