From 48a6f060246ed46a47b6c44bbb04192afb67ec88 Mon Sep 17 00:00:00 2001 From: dentinyhao Date: Tue, 30 Jun 2026 13:53:55 -0700 Subject: [PATCH] Propagate error for table migration Signed-off-by: dentinyhao --- pkg/controller/chi/worker-migrator.go | 1 + pkg/controller/chi/worker-reconciler-chi.go | 20 ++++----- pkg/model/chi/schemer/cluster.go | 10 +++-- pkg/model/chi/schemer/distributed.go | 45 ++++++++++++--------- pkg/model/chi/schemer/replicated.go | 45 ++++++++++++--------- pkg/model/chi/schemer/schemer.go | 45 +++++++++++---------- 6 files changed, 96 insertions(+), 70 deletions(-) diff --git a/pkg/controller/chi/worker-migrator.go b/pkg/controller/chi/worker-migrator.go index 1e1a8d8f4..41048516c 100644 --- a/pkg/controller/chi/worker-migrator.go +++ b/pkg/controller/chi/worker-migrator.go @@ -105,6 +105,7 @@ func (w *worker) migrateTables(ctx context.Context, host *api.Host, opts *migrat M(host).F(). Error("ERROR add tables failed on shard/host:%d/%d cluster:%s err:%v", host.Runtime.Address.ShardIndex, host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ClusterName, err) + return err } w.a.V(1). diff --git a/pkg/controller/chi/worker-reconciler-chi.go b/pkg/controller/chi/worker-reconciler-chi.go index 34123eaa5..a3adc6d01 100644 --- a/pkg/controller/chi/worker-reconciler-chi.go +++ b/pkg/controller/chi/worker-reconciler-chi.go @@ -857,9 +857,11 @@ func (w *worker) reconcileHost(ctx context.Context, host *api.Host) error { return err } if err := w.reconcileHostMain(ctx, host); err != nil { + // Domain/schema failure: keep status as Requested (excluded from remote_servers) + // and do not include the host into cluster activities. return err } - // Host is now added and functional + // Host STS is up and schema migration (if required) succeeded — safe to include. if host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) { host.GetReconcileAttributes().SetStatus(types.ObjectStatusCreated) @@ -984,18 +986,18 @@ func (w *worker) reconcileHostMain(ctx context.Context, host *api.Host) error { Warning("Reconcile Host Main - unable to reconcile PVC. Host: %s Err: %v", host.GetName(), err) } - // Finalize main reconcile with domain activities + // Finalize main reconcile with domain activities (schema migration, FIPS checks). + // Schema/domain failures must fail this host's reconcile so we do not: + // 1. flip status Requested → Created (which re-includes the host in remote_servers), + // 2. call includeHost / mark the host ready for Distributed traffic, + // while local tables are still missing. Callers of reconcileHost only include + // the host after reconcileHostMain returns nil. if err := w.reconcileHostMainDomain(ctx, host, migrateTableOpts); 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) - // Propagate abort signals (e.g. runtime FIPS image-policy violation) so - // reconcile() routes them through markReconcileCompletedUnsuccessfully, - // which persists StatusAborted via updateCRObjectStatus. Other errors - // stay logged but non-fatal — preserves pre-existing tolerance. - if errors.Is(err, common.ErrCRUDAbort) { - return err - } + return err } return nil diff --git a/pkg/model/chi/schemer/cluster.go b/pkg/model/chi/schemer/cluster.go index 21510d42a..0bae0b984 100644 --- a/pkg/model/chi/schemer/cluster.go +++ b/pkg/model/chi/schemer/cluster.go @@ -53,7 +53,7 @@ func (c *Cluster) SetClusterConnectionParams(clusterConnectionParams *clickhouse func (c *Cluster) queryUnzipColumns(ctx context.Context, hosts []string, sql string, columns ...*[]string) error { if util.IsContextDone(ctx) { log.V(1).Info("ctx is done") - return nil + return ctx.Err() } if len(hosts) == 0 { @@ -61,12 +61,16 @@ func (c *Cluster) queryUnzipColumns(ctx context.Context, hosts []string, sql str return nil } - // Fetch data from any of specified hosts + // Fetch data from any of specified hosts. Propagate failures so schema migration + // cannot silently treat "could not discover objects" as "nothing to create". query, err := c.SetHosts(hosts).QueryAny(ctx, sql) if err != nil { - return nil + return err } if query == nil { + if util.IsContextDone(ctx) { + return ctx.Err() + } return nil } diff --git a/pkg/model/chi/schemer/distributed.go b/pkg/model/chi/schemer/distributed.go index fe06870bd..7595f41fb 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( - s.QueryUnzip2Columns( - ctx, - peers, - s.sqlCreateDatabaseDistributed(host.Runtime.Address.ClusterName), - ), + databaseNames, createDatabaseSQLs, err := s.QueryUnzip2Columns( + ctx, + peers, + s.sqlCreateDatabaseDistributed(host.Runtime.Address.ClusterName), ) - tableNames, createTableSQLs := debugCreateSQLs( - s.QueryUnzipAndApplyUUIDs( - ctx, - peers, - s.sqlCreateTableDistributed(host.Runtime.Address.ClusterName), - ), + databaseNames, createDatabaseSQLs = debugCreateSQLs(databaseNames, createDatabaseSQLs, err) + if err != nil { + return nil, nil, err + } + + tableNames, createTableSQLs, err := s.QueryUnzipAndApplyUUIDs( + ctx, + peers, + s.sqlCreateTableDistributed(host.Runtime.Address.ClusterName), ) - functionNames, createFunctionSQLs := debugCreateSQLs( - s.QueryUnzip2Columns( - ctx, - peers, - s.sqlCreateFunction(host.Runtime.Address.ClusterName), - ), + tableNames, createTableSQLs = debugCreateSQLs(tableNames, createTableSQLs, err) + if err != nil { + return nil, nil, err + } + + functionNames, createFunctionSQLs, err := s.QueryUnzip2Columns( + ctx, + peers, + s.sqlCreateFunction(host.Runtime.Address.ClusterName), ) + functionNames, createFunctionSQLs = debugCreateSQLs(functionNames, createFunctionSQLs, err) + 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..63ef4b4f5 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( - s.QueryUnzip2Columns( - ctx, - peers, - s.sqlCreateDatabaseReplicated(host.Runtime.Address.ClusterName), - ), + databaseNames, createDatabaseSQLs, err := s.QueryUnzip2Columns( + ctx, + peers, + s.sqlCreateDatabaseReplicated(host.Runtime.Address.ClusterName), ) - tableNames, createTableSQLs := debugCreateSQLs( - s.QueryUnzipAndApplyUUIDs( - ctx, - peers, - s.sqlCreateTableReplicated(host.Runtime.Address.ClusterName), - ), + databaseNames, createDatabaseSQLs = debugCreateSQLs(databaseNames, createDatabaseSQLs, err) + if err != nil { + return nil, nil, err + } + + tableNames, createTableSQLs, err := s.QueryUnzipAndApplyUUIDs( + ctx, + peers, + s.sqlCreateTableReplicated(host.Runtime.Address.ClusterName), ) - functionNames, createFunctionSQLs := debugCreateSQLs( - s.QueryUnzip2Columns( - ctx, - peers, - s.sqlCreateFunction(host.Runtime.Address.ClusterName), - ), + tableNames, createTableSQLs = debugCreateSQLs(tableNames, createTableSQLs, err) + if err != nil { + return nil, nil, err + } + + functionNames, createFunctionSQLs, err := s.QueryUnzip2Columns( + ctx, + peers, + s.sqlCreateFunction(host.Runtime.Address.ClusterName), ) + functionNames, createFunctionSQLs = debugCreateSQLs(functionNames, createFunctionSQLs, err) + 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 6b503bedd..c1ded836f 100644 --- a/pkg/model/chi/schemer/schemer.go +++ b/pkg/model/chi/schemer/schemer.go @@ -86,23 +86,24 @@ 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 + replicatedObjectNames, replicatedCreateSQLs, err = s.getReplicatedObjectsSQLs(ctx, host) + if err != nil { + return nil, nil, nil, nil, err } - if names, sql, err := s.getDistributedObjectsSQLs(ctx, host); err == nil { - distributedObjectNames = names - distributedCreateSQLs = sql + distributedObjectNames, distributedCreateSQLs, err = s.getDistributedObjectsSQLs(ctx, host) + if err != nil { + return nil, nil, nil, nil, err } - return + return replicatedObjectNames, replicatedCreateSQLs, distributedObjectNames, distributedCreateSQLs, nil } // HostCreateTables creates tables on a new host func (s *ClusterSchemer) HostCreateTables(ctx context.Context, host *api.Host) error { if util.IsContextDone(ctx) { log.V(1).Info("ctx is done") - return nil + return ctx.Err() } log.V(1).M(host).F().S().Info("Migrating schema objects to host %s", host.Runtime.Address.HostName) @@ -111,29 +112,29 @@ 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 { + log.V(1).M(host).F().Error("Failed to discover schema objects for host %s: %v", host.Runtime.Address.HostName, err) + return err + } - var err1 error if len(replicatedCreateSQLs) > 0 { log.V(1).M(host).F().Info("Creating replicated objects at %s: %v", host.Runtime.Address.HostName, replicatedObjectNames) log.V(2).M(host).F().Info("\n%v", replicatedCreateSQLs) - err1 = s.ExecHost(ctx, host, replicatedCreateSQLs, - clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true)) + if err := s.ExecHost(ctx, host, replicatedCreateSQLs, + clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true)); err != nil { + return err + } } - var err2 error if len(distributedCreateSQLs) > 0 { log.V(1).M(host).F().Info("Creating distributed objects at %s: %v", host.Runtime.Address.HostName, distributedObjectNames) log.V(2).M(host).F().Info("\n%v", distributedCreateSQLs) - err2 = s.ExecHost(ctx, host, distributedCreateSQLs, - clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true)) - } - - if err2 != nil { - return err2 - } - if err1 != nil { - return err1 + if err := s.ExecHost(ctx, host, distributedCreateSQLs, + clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true)); err != nil { + return err + } } return nil