Skip to content

Commit b029c97

Browse files
mingjia1mingjia1
committed
fix: not finish all mha cluster deploy
1 parent d9965ac commit b029c97

20 files changed

Lines changed: 829 additions & 735 deletions

‎agent/internal/executor/backup_scan_test.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,18 @@ func TestParseGTIDFromGzipDump(t *testing.T) {
9595
assert.Equal(t, "uuid:1-100", gtid)
9696
}
9797

98+
func TestParseGTIDFromDumpReturnsMissingGTIDError(t *testing.T) {
99+
root := t.TempDir()
100+
path := filepath.Join(root, "full.sql")
101+
require.NoError(t, os.WriteFile(path, []byte("-- dump without GTID_PURGED\nCREATE DATABASE db1;\n"), 0o644))
102+
103+
gtid, err := parseGTIDFromDump(path)
104+
105+
require.Error(t, err)
106+
assert.Empty(t, gtid)
107+
assert.Contains(t, err.Error(), "GTID_PURGED not found")
108+
}
109+
98110
func TestExecuteColdDatadirBackupCreatesArchive(t *testing.T) {
99111
t.Setenv("DBOPS_ALLOW_TMP_DECOMMISSION_TEST", "1")
100112
root := t.TempDir()

‎agent/internal/executor/task_executor.go‎

Lines changed: 40 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -525,6 +525,7 @@ func (e *TaskExecutor) deploySingleInstance(ctx context.Context, req DeployTaskR
525525
osUser, _ := req.Config["os_user"].(string)
526526
mysqlUser, _ := req.Config["mysql_user"].(string)
527527
mysqlPass, _ := req.Config["mysql_pass"].(string)
528+
serverID := configInt(req.Config, "server_id")
528529

529530
// MGR配置参数
530531
installType, _ := req.Config["install_type"].(string)
@@ -542,6 +543,9 @@ func (e *TaskExecutor) deploySingleInstance(ctx context.Context, req DeployTaskR
542543
if port == 0 {
543544
port = 3306
544545
}
546+
if serverID == 0 {
547+
serverID = port
548+
}
545549
if dataDir == "" {
546550
dataDir = fmt.Sprintf("/data/mysql/%d", port)
547551
}
@@ -653,7 +657,7 @@ func (e *TaskExecutor) deploySingleInstance(ctx context.Context, req DeployTaskR
653657
"--daemonize",
654658
"--datadir=" + dataDir,
655659
"--port=" + fmt.Sprintf("%d", port),
656-
"--server-id=" + fmt.Sprintf("%d", port),
660+
"--server-id=" + fmt.Sprintf("%d", serverID),
657661
"--log-bin=mysql-bin",
658662
"--binlog-format=ROW",
659663
"--gtid-mode=ON",
@@ -1030,7 +1034,7 @@ func (e *TaskExecutor) configureMaster(ctx context.Context, config MasterSlaveCo
10301034
_ = mysqlExecCommand(ctx, config.MasterHost, config.MasterPort, config.MySQLUser, config.MySQLPass,
10311035
fmt.Sprintf("ALTER USER '%s'@'%%' IDENTIFIED WITH mysql_native_password BY '%s';", escapeSQL(config.ReplicateUser), escapeSQL(config.ReplicatePass))).Run()
10321036

1033-
if out, err := setServerID(ctx, config.MasterHost, config.MasterPort, config.MySQLUser, config.MySQLPass, 1); err != nil {
1037+
if out, err := setServerID(ctx, config.MasterHost, config.MasterPort, config.MySQLUser, config.MySQLPass, config.ServerID); err != nil {
10341038
return &TaskResult{
10351039
Status: "failed",
10361040
Progress: 30,
@@ -1048,7 +1052,7 @@ func (e *TaskExecutor) configureMaster(ctx context.Context, config MasterSlaveCo
10481052
}
10491053

10501054
func (e *TaskExecutor) configureSlave(ctx context.Context, config MasterSlaveConfig) *TaskResult {
1051-
if out, err := setServerID(ctx, config.SlaveHost, config.SlavePort, config.MySQLUser, config.MySQLPass, config.ServerID+2); err != nil {
1055+
if out, err := setServerID(ctx, config.SlaveHost, config.SlavePort, config.MySQLUser, config.MySQLPass, config.ServerID); err != nil {
10521056
return &TaskResult{
10531057
Status: "failed",
10541058
Progress: 50,
@@ -1565,13 +1569,18 @@ func (e *TaskExecutor) executeGTIDIncrementalBackup(ctx context.Context, config
15651569
}, nil
15661570
}
15671571
baseGTID, err := parseGTIDFromDump(config.BaseBackupPath)
1572+
gtidWarning := ""
15681573
if err != nil {
1569-
return &TaskResult{
1570-
Status: "failed",
1571-
Progress: 0,
1572-
Message: fmt.Sprintf("parse GTID from base mysqldump failed: %v", err),
1573-
Timestamp: time.Now(),
1574-
}, nil
1574+
if strings.Contains(err.Error(), "GTID_PURGED not found") {
1575+
gtidWarning = "base mysqldump has no GTID_PURGED; mysqlbinlog will not exclude transactions already included in the base dump"
1576+
} else {
1577+
return &TaskResult{
1578+
Status: "failed",
1579+
Progress: 0,
1580+
Message: fmt.Sprintf("parse GTID from base mysqldump failed: %v", err),
1581+
Timestamp: time.Now(),
1582+
}, nil
1583+
}
15751584
}
15761585
binlogs, err := fetchBinaryLogFiles(ctx, config)
15771586
if err != nil {
@@ -1608,9 +1617,11 @@ func (e *TaskExecutor) executeGTIDIncrementalBackup(ctx context.Context, config
16081617
"--port=" + fmt.Sprintf("%d", config.MySQLPort),
16091618
"--user=" + defaultString(config.MySQLUser, "root"),
16101619
"--password=" + config.MySQLPass,
1611-
"--exclude-gtids=" + baseGTID,
16121620
"--result-file=" + backupFile,
16131621
}
1622+
if strings.TrimSpace(baseGTID) != "" {
1623+
args = append(args, "--exclude-gtids="+baseGTID)
1624+
}
16141625
args = append(args, binlogs...)
16151626
cmd := exec.CommandContext(ctx, "mysqlbinlog", args...)
16161627
if out, err := cmd.CombinedOutput(); err != nil {
@@ -1647,21 +1658,29 @@ func (e *TaskExecutor) executeGTIDIncrementalBackup(ctx context.Context, config
16471658
Timestamp: time.Now(),
16481659
}, nil
16491660
}
1661+
message := fmt.Sprintf("Incremental backup completed successfully. Path: %s, Checksum: %s", backupFile, checksum)
1662+
if gtidWarning != "" {
1663+
message += ". " + gtidWarning
1664+
}
1665+
data := map[string]any{
1666+
"backup_path": backupFile,
1667+
"backup_type": config.BackupType,
1668+
"backup_method": "mysqlbinlog",
1669+
"base_backup_path": config.BaseBackupPath,
1670+
"base_backup_label": config.BaseBackupLabel,
1671+
"base_backup_gtid": baseGTID,
1672+
"file_size": sizeBytes,
1673+
"checksum": checksum,
1674+
}
1675+
if gtidWarning != "" {
1676+
data["gtid_warning"] = gtidWarning
1677+
}
16501678
return &TaskResult{
16511679
Status: "completed",
16521680
Progress: 100,
1653-
Message: fmt.Sprintf("Incremental backup completed successfully. Path: %s, Checksum: %s", backupFile, checksum),
1681+
Message: message,
16541682
Timestamp: time.Now(),
1655-
Data: map[string]any{
1656-
"backup_path": backupFile,
1657-
"backup_type": config.BackupType,
1658-
"backup_method": "mysqlbinlog",
1659-
"base_backup_path": config.BaseBackupPath,
1660-
"base_backup_label": config.BaseBackupLabel,
1661-
"base_backup_gtid": baseGTID,
1662-
"file_size": sizeBytes,
1663-
"checksum": checksum,
1664-
},
1683+
Data: data,
16651684
}, nil
16661685
}
16671686

‎platform-backend/internal/models/instance.go‎

Lines changed: 26 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,12 @@ type Instance struct {
1111
HostID *string `json:"host_id"`
1212
CreatedAt time.Time `json:"created_at"`
1313
UpdatedAt time.Time `json:"updated_at"`
14-
15-
Connection InstanceConnection `json:"connection"`
16-
Version InstanceVersion `json:"version"`
17-
Config InstanceConfig `json:"config"`
18-
Status InstanceStatus `json:"status"`
19-
Topology InstanceTopology `json:"topology"`
14+
15+
Connection InstanceConnection `json:"connection"`
16+
Version InstanceVersion `json:"version"`
17+
Config InstanceConfig `json:"config"`
18+
Status InstanceStatus `json:"status"`
19+
Topology InstanceTopology `json:"topology"`
2020
}
2121

2222
type InstanceConnection struct {
@@ -29,33 +29,33 @@ type InstanceConnection struct {
2929
SSLEnabled bool `json:"ssl_enabled"`
3030
// Install / upgrade paths. These let the platform install or upgrade to
3131
// ANY version from the catalog (not hard-coded to 5.7/8.0).
32-
Basedir string `json:"basedir" `
33-
Datadir string `json:"datadir" `
34-
OSUser string `json:"os_user" `
32+
Basedir string `json:"basedir"`
33+
Datadir string `json:"datadir"`
34+
OSUser string `json:"os_user"`
3535
PackageURL string `json:"package_url"`
36-
VersionID string `json:"version_id" ` // FK to version catalog id e.g. "mysql-8.0.36"
36+
VersionID string `json:"version_id"` // FK to version catalog id e.g. "mysql-8.0.36"
3737
}
3838

3939
type InstanceVersion struct {
40-
ID string `gorm:"primaryKey;type:varchar(64)"`
41-
InstanceID string `json:"instance_id"`
42-
Flavor string `json:"flavor"`
43-
Version string `json:"version"`
44-
FullVersion string `json:"full_version"`
45-
ReleaseDate time.Time `json:"release_date"`
46-
EOLDate time.Time `json:"eol_date"`
47-
IsLTS bool `json:"is_lts"`
48-
Features string `json:"features"`
49-
Engines string `json:"engines"`
40+
ID string `gorm:"primaryKey;type:varchar(64)"`
41+
InstanceID string `json:"instance_id"`
42+
Flavor string `json:"flavor"`
43+
Version string `json:"version"`
44+
FullVersion string `json:"full_version"`
45+
ReleaseDate time.Time `json:"release_date"`
46+
EOLDate time.Time `json:"eol_date"`
47+
IsLTS bool `json:"is_lts"`
48+
Features string `json:"features"`
49+
Engines string `json:"engines"`
5050
}
5151

5252
type InstanceConfig struct {
53-
ID string `gorm:"primaryKey;type:varchar(64)"`
54-
InstanceID string `json:"instance_id"`
53+
ID string `gorm:"primaryKey;type:varchar(64)"`
54+
InstanceID string `json:"instance_id"`
5555
ParameterTemplateID string `json:"parameter_template_id"`
56-
Parameters string `json:"parameters"`
57-
Charset string `json:"charset"`
58-
Collation string `json:"collation"`
56+
Parameters string `json:"parameters"`
57+
Charset string `json:"charset"`
58+
Collation string `json:"collation"`
5959
}
6060

6161
type InstanceStatus struct {
@@ -76,4 +76,4 @@ type InstanceTopology struct {
7676
MasterID string `json:"master_id"`
7777
SlaveIDs string `json:"slave_ids"`
7878
ReplicationMode string `json:"replication_mode"`
79-
}
79+
}

‎platform-backend/internal/repositories/alert_rule_repository.go‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,16 @@ func (r *AlertRuleRepository) CreateAlertRule(ctx context.Context, rule *models.
3636
rule.DurationSeconds, rule.Severity, rule.NotificationChannels,
3737
rule.Expression, rule.CreatedAt, rule.UpdatedAt)
3838

39+
if err != nil {
40+
if strings.Contains(strings.ToLower(err.Error()), "expression") {
41+
if addErr := r.ensureAlertRuleExpressionColumn(ctx); addErr == nil {
42+
_, err = r.db.Pool.ExecContext(ctx, query,
43+
rule.ID, rule.Name, rule.Metric, rule.Condition, rule.Threshold,
44+
rule.DurationSeconds, rule.Severity, rule.NotificationChannels,
45+
rule.Expression, rule.CreatedAt, rule.UpdatedAt)
46+
}
47+
}
48+
}
3949
if err != nil {
4050
return fmt.Errorf("failed to create alert rule: %w", err)
4151
}
@@ -61,6 +71,16 @@ func (r *AlertRuleRepository) UpdateAlertRule(ctx context.Context, rule *models.
6171
rule.DurationSeconds, rule.Severity, rule.NotificationChannels,
6272
rule.Expression, rule.UpdatedAt, rule.ID)
6373

74+
if err != nil {
75+
if strings.Contains(strings.ToLower(err.Error()), "expression") {
76+
if addErr := r.ensureAlertRuleExpressionColumn(ctx); addErr == nil {
77+
_, err = r.db.Pool.ExecContext(ctx, query,
78+
rule.Name, rule.Metric, rule.Condition, rule.Threshold,
79+
rule.DurationSeconds, rule.Severity, rule.NotificationChannels,
80+
rule.Expression, rule.UpdatedAt, rule.ID)
81+
}
82+
}
83+
}
6484
if err != nil {
6585
return fmt.Errorf("failed to update alert rule: %w", err)
6686
}
@@ -118,6 +138,13 @@ func (r *AlertRuleRepository) ListAlertRules(ctx context.Context, limit, offset
118138
`
119139

120140
rows, err := r.db.Pool.QueryContext(ctx, query, limit, offset)
141+
if err != nil {
142+
if strings.Contains(strings.ToLower(err.Error()), "expression") {
143+
if addErr := r.ensureAlertRuleExpressionColumn(ctx); addErr == nil {
144+
rows, err = r.db.Pool.QueryContext(ctx, query, limit, offset)
145+
}
146+
}
147+
}
121148
if err != nil {
122149
return nil, fmt.Errorf("failed to list alert rules: %w", err)
123150
}
@@ -138,6 +165,17 @@ func (r *AlertRuleRepository) ListAlertRules(ctx context.Context, limit, offset
138165
return rules, nil
139166
}
140167

168+
func (r *AlertRuleRepository) ensureAlertRuleExpressionColumn(ctx context.Context) error {
169+
if r.db == nil || r.db.Pool == nil {
170+
return nil
171+
}
172+
_, err := r.db.Pool.ExecContext(ctx, `ALTER TABLE alert_rules ADD COLUMN expression TEXT DEFAULT ''`)
173+
if err != nil && !isAlreadyExistsError(err) {
174+
return err
175+
}
176+
return nil
177+
}
178+
141179
func (r *AlertRuleRepository) GetActiveAlertRules(ctx context.Context) ([]models.AlertRule, error) {
142180
if r.db.Pool == nil {
143181
return []models.AlertRule{}, nil
@@ -149,6 +187,13 @@ func (r *AlertRuleRepository) GetActiveAlertRules(ctx context.Context) ([]models
149187
`
150188

151189
rows, err := r.db.Pool.QueryContext(ctx, query)
190+
if err != nil {
191+
if strings.Contains(strings.ToLower(err.Error()), "expression") {
192+
if addErr := r.ensureAlertRuleExpressionColumn(ctx); addErr == nil {
193+
rows, err = r.db.Pool.QueryContext(ctx, query)
194+
}
195+
}
196+
}
152197
if err != nil {
153198
return nil, fmt.Errorf("failed to get active alert rules: %w", err)
154199
}

‎platform-backend/internal/services/cluster_deploy_service.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -376,6 +376,7 @@ func (s *ClusterDeployService) ExecuteClusterDeployPlan(ctx context.Context, pla
376376
Name: name,
377377
Status: "partial",
378378
Message: errMsg,
379+
ErrorMessage: errMsg,
379380
StartedAt: dep.StartedAt,
380381
FinishedAt: &finish,
381382
CreatedAt: dep.CreatedAt,
@@ -479,6 +480,7 @@ func (s *ClusterDeployService) buildPartialResponse(ctx context.Context, cluster
479480
Name: name,
480481
Status: "failed",
481482
Message: errMsg,
483+
ErrorMessage: errMsg,
482484
StartedAt: dep.StartedAt,
483485
FinishedAt: finishedAt,
484486
CreatedAt: dep.CreatedAt,
@@ -1068,6 +1070,7 @@ type DeployResponse struct {
10681070
Stage string `json:"stage,omitempty"`
10691071
Progress int `json:"progress"`
10701072
Message string `json:"message"`
1073+
ErrorMessage string `json:"error_message,omitempty"`
10711074
StartedAt *time.Time `json:"started_at,omitempty"`
10721075
FinishedAt *time.Time `json:"finished_at,omitempty"`
10731076
CreatedAt time.Time `json:"created_at"`

0 commit comments

Comments
 (0)