Skip to content

Commit 48a6f06

Browse files
dentinydentiny
committed
Propagate error for table migration
Signed-off-by: dentinyhao <dentinyhao@gmail.com>
1 parent 4ca7416 commit 48a6f06

6 files changed

Lines changed: 96 additions & 70 deletions

File tree

‎pkg/controller/chi/worker-migrator.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,7 @@ func (w *worker) migrateTables(ctx context.Context, host *api.Host, opts *migrat
105105
M(host).F().
106106
Error("ERROR add tables failed on shard/host:%d/%d cluster:%s err:%v",
107107
host.Runtime.Address.ShardIndex, host.Runtime.Address.ReplicaIndex, host.Runtime.Address.ClusterName, err)
108+
return err
108109
}
109110

110111
w.a.V(1).

‎pkg/controller/chi/worker-reconciler-chi.go‎

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -857,9 +857,11 @@ func (w *worker) reconcileHost(ctx context.Context, host *api.Host) error {
857857
return err
858858
}
859859
if err := w.reconcileHostMain(ctx, host); err != nil {
860+
// Domain/schema failure: keep status as Requested (excluded from remote_servers)
861+
// and do not include the host into cluster activities.
860862
return err
861863
}
862-
// Host is now added and functional
864+
// Host STS is up and schema migration (if required) succeeded — safe to include.
863865

864866
if host.GetReconcileAttributes().GetStatus().Is(types.ObjectStatusRequested) {
865867
host.GetReconcileAttributes().SetStatus(types.ObjectStatusCreated)
@@ -984,18 +986,18 @@ func (w *worker) reconcileHostMain(ctx context.Context, host *api.Host) error {
984986
Warning("Reconcile Host Main - unable to reconcile PVC. Host: %s Err: %v", host.GetName(), err)
985987
}
986988

987-
// Finalize main reconcile with domain activities
989+
// Finalize main reconcile with domain activities (schema migration, FIPS checks).
990+
// Schema/domain failures must fail this host's reconcile so we do not:
991+
// 1. flip status Requested → Created (which re-includes the host in remote_servers),
992+
// 2. call includeHost / mark the host ready for Distributed traffic,
993+
// while local tables are still missing. Callers of reconcileHost only include
994+
// the host after reconcileHostMain returns nil.
988995
if err := w.reconcileHostMainDomain(ctx, host, migrateTableOpts); err != nil {
996+
metrics.HostReconcilesErrors(ctx, host.GetCR())
989997
w.a.V(1).
990998
M(host).F().
991999
Warning("Reconcile Host Main - unable to reconcile domain reconcile. Host: %s Err: %v", host.GetName(), err)
992-
// Propagate abort signals (e.g. runtime FIPS image-policy violation) so
993-
// reconcile() routes them through markReconcileCompletedUnsuccessfully,
994-
// which persists StatusAborted via updateCRObjectStatus. Other errors
995-
// stay logged but non-fatal — preserves pre-existing tolerance.
996-
if errors.Is(err, common.ErrCRUDAbort) {
997-
return err
998-
}
1000+
return err
9991001
}
10001002

10011003
return nil

‎pkg/model/chi/schemer/cluster.go‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -53,20 +53,24 @@ func (c *Cluster) SetClusterConnectionParams(clusterConnectionParams *clickhouse
5353
func (c *Cluster) queryUnzipColumns(ctx context.Context, hosts []string, sql string, columns ...*[]string) error {
5454
if util.IsContextDone(ctx) {
5555
log.V(1).Info("ctx is done")
56-
return nil
56+
return ctx.Err()
5757
}
5858

5959
if len(hosts) == 0 {
6060
// Nowhere to fetch data from
6161
return nil
6262
}
6363

64-
// Fetch data from any of specified hosts
64+
// Fetch data from any of specified hosts. Propagate failures so schema migration
65+
// cannot silently treat "could not discover objects" as "nothing to create".
6566
query, err := c.SetHosts(hosts).QueryAny(ctx, sql)
6667
if err != nil {
67-
return nil
68+
return err
6869
}
6970
if query == nil {
71+
if util.IsContextDone(ctx) {
72+
return ctx.Err()
73+
}
7074
return nil
7175
}
7276

‎pkg/model/chi/schemer/distributed.go‎

Lines changed: 27 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -56,27 +56,36 @@ func (s *ClusterSchemer) getDistributedObjectsSQLs(ctx context.Context, host *ap
5656
// Exclude self — see the matching comment in getReplicatedObjectsSQLs.
5757
peers := s.Names(interfaces.NameFQDNs, host, api.Cluster{}, true)
5858

59-
databaseNames, createDatabaseSQLs := debugCreateSQLs(
60-
s.QueryUnzip2Columns(
61-
ctx,
62-
peers,
63-
s.sqlCreateDatabaseDistributed(host.Runtime.Address.ClusterName),
64-
),
59+
databaseNames, createDatabaseSQLs, err := s.QueryUnzip2Columns(
60+
ctx,
61+
peers,
62+
s.sqlCreateDatabaseDistributed(host.Runtime.Address.ClusterName),
6563
)
66-
tableNames, createTableSQLs := debugCreateSQLs(
67-
s.QueryUnzipAndApplyUUIDs(
68-
ctx,
69-
peers,
70-
s.sqlCreateTableDistributed(host.Runtime.Address.ClusterName),
71-
),
64+
databaseNames, createDatabaseSQLs = debugCreateSQLs(databaseNames, createDatabaseSQLs, err)
65+
if err != nil {
66+
return nil, nil, err
67+
}
68+
69+
tableNames, createTableSQLs, err := s.QueryUnzipAndApplyUUIDs(
70+
ctx,
71+
peers,
72+
s.sqlCreateTableDistributed(host.Runtime.Address.ClusterName),
7273
)
73-
functionNames, createFunctionSQLs := debugCreateSQLs(
74-
s.QueryUnzip2Columns(
75-
ctx,
76-
peers,
77-
s.sqlCreateFunction(host.Runtime.Address.ClusterName),
78-
),
74+
tableNames, createTableSQLs = debugCreateSQLs(tableNames, createTableSQLs, err)
75+
if err != nil {
76+
return nil, nil, err
77+
}
78+
79+
functionNames, createFunctionSQLs, err := s.QueryUnzip2Columns(
80+
ctx,
81+
peers,
82+
s.sqlCreateFunction(host.Runtime.Address.ClusterName),
7983
)
84+
functionNames, createFunctionSQLs = debugCreateSQLs(functionNames, createFunctionSQLs, err)
85+
if err != nil {
86+
return nil, nil, err
87+
}
88+
8089
return util.ConcatSlices([][]string{databaseNames, tableNames, functionNames}),
8190
util.ConcatSlices([][]string{createDatabaseSQLs, createTableSQLs, createFunctionSQLs}),
8291
nil

‎pkg/model/chi/schemer/replicated.go‎

Lines changed: 27 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -71,27 +71,36 @@ func (s *ClusterSchemer) getReplicatedObjectsSQLs(ctx context.Context, host *api
7171
// schemer silently skips CREATE. Peers hold the truth.
7272
peers := s.Names(interfaces.NameFQDNs, host, api.Cluster{}, true)
7373

74-
databaseNames, createDatabaseSQLs := debugCreateSQLs(
75-
s.QueryUnzip2Columns(
76-
ctx,
77-
peers,
78-
s.sqlCreateDatabaseReplicated(host.Runtime.Address.ClusterName),
79-
),
74+
databaseNames, createDatabaseSQLs, err := s.QueryUnzip2Columns(
75+
ctx,
76+
peers,
77+
s.sqlCreateDatabaseReplicated(host.Runtime.Address.ClusterName),
8078
)
81-
tableNames, createTableSQLs := debugCreateSQLs(
82-
s.QueryUnzipAndApplyUUIDs(
83-
ctx,
84-
peers,
85-
s.sqlCreateTableReplicated(host.Runtime.Address.ClusterName),
86-
),
79+
databaseNames, createDatabaseSQLs = debugCreateSQLs(databaseNames, createDatabaseSQLs, err)
80+
if err != nil {
81+
return nil, nil, err
82+
}
83+
84+
tableNames, createTableSQLs, err := s.QueryUnzipAndApplyUUIDs(
85+
ctx,
86+
peers,
87+
s.sqlCreateTableReplicated(host.Runtime.Address.ClusterName),
8788
)
88-
functionNames, createFunctionSQLs := debugCreateSQLs(
89-
s.QueryUnzip2Columns(
90-
ctx,
91-
peers,
92-
s.sqlCreateFunction(host.Runtime.Address.ClusterName),
93-
),
89+
tableNames, createTableSQLs = debugCreateSQLs(tableNames, createTableSQLs, err)
90+
if err != nil {
91+
return nil, nil, err
92+
}
93+
94+
functionNames, createFunctionSQLs, err := s.QueryUnzip2Columns(
95+
ctx,
96+
peers,
97+
s.sqlCreateFunction(host.Runtime.Address.ClusterName),
9498
)
99+
functionNames, createFunctionSQLs = debugCreateSQLs(functionNames, createFunctionSQLs, err)
100+
if err != nil {
101+
return nil, nil, err
102+
}
103+
95104
return util.ConcatSlices([][]string{databaseNames, tableNames, functionNames}),
96105
util.ConcatSlices([][]string{createDatabaseSQLs, createTableSQLs, createFunctionSQLs}),
97106
nil

‎pkg/model/chi/schemer/schemer.go‎

Lines changed: 23 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -86,23 +86,24 @@ func (s *ClusterSchemer) createTablesSQLs(
8686
replicatedCreateSQLs []string,
8787
distributedObjectNames []string,
8888
distributedCreateSQLs []string,
89+
err error,
8990
) {
90-
if names, sql, err := s.getReplicatedObjectsSQLs(ctx, host); err == nil {
91-
replicatedObjectNames = names
92-
replicatedCreateSQLs = sql
91+
replicatedObjectNames, replicatedCreateSQLs, err = s.getReplicatedObjectsSQLs(ctx, host)
92+
if err != nil {
93+
return nil, nil, nil, nil, err
9394
}
94-
if names, sql, err := s.getDistributedObjectsSQLs(ctx, host); err == nil {
95-
distributedObjectNames = names
96-
distributedCreateSQLs = sql
95+
distributedObjectNames, distributedCreateSQLs, err = s.getDistributedObjectsSQLs(ctx, host)
96+
if err != nil {
97+
return nil, nil, nil, nil, err
9798
}
98-
return
99+
return replicatedObjectNames, replicatedCreateSQLs, distributedObjectNames, distributedCreateSQLs, nil
99100
}
100101

101102
// HostCreateTables creates tables on a new host
102103
func (s *ClusterSchemer) HostCreateTables(ctx context.Context, host *api.Host) error {
103104
if util.IsContextDone(ctx) {
104105
log.V(1).Info("ctx is done")
105-
return nil
106+
return ctx.Err()
106107
}
107108

108109
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
111112
replicatedObjectNames,
112113
replicatedCreateSQLs,
113114
distributedObjectNames,
114-
distributedCreateSQLs := s.createTablesSQLs(ctx, host)
115+
distributedCreateSQLs,
116+
err := s.createTablesSQLs(ctx, host)
117+
if err != nil {
118+
log.V(1).M(host).F().Error("Failed to discover schema objects for host %s: %v", host.Runtime.Address.HostName, err)
119+
return err
120+
}
115121

116-
var err1 error
117122
if len(replicatedCreateSQLs) > 0 {
118123
log.V(1).M(host).F().Info("Creating replicated objects at %s: %v", host.Runtime.Address.HostName, replicatedObjectNames)
119124
log.V(2).M(host).F().Info("\n%v", replicatedCreateSQLs)
120-
err1 = s.ExecHost(ctx, host, replicatedCreateSQLs,
121-
clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true))
125+
if err := s.ExecHost(ctx, host, replicatedCreateSQLs,
126+
clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true)); err != nil {
127+
return err
128+
}
122129
}
123130

124-
var err2 error
125131
if len(distributedCreateSQLs) > 0 {
126132
log.V(1).M(host).F().Info("Creating distributed objects at %s: %v", host.Runtime.Address.HostName, distributedObjectNames)
127133
log.V(2).M(host).F().Info("\n%v", distributedCreateSQLs)
128-
err2 = s.ExecHost(ctx, host, distributedCreateSQLs,
129-
clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true))
130-
}
131-
132-
if err2 != nil {
133-
return err2
134-
}
135-
if err1 != nil {
136-
return err1
134+
if err := s.ExecHost(ctx, host, distributedCreateSQLs,
135+
clickhouse.NewQueryOptions().SetRetry(true).SetLogQueries(true)); err != nil {
136+
return err
137+
}
137138
}
138139

139140
return nil

0 commit comments

Comments
 (0)