wonder/xinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: register MySQL instances in CloudDM

Browse Source
This commit is contained in:
guoyonghao
2026-07-29 18:24:36 +08:00
parent 33449b60b2
commit 0aac99c359
2 changed files with 126 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/service/delivery.go
+93 -6
View File
@@ -37,6 +37,7 @@ type MySQLDeliveryInput struct {
TargetID uint64 `json:"target_id" binding:"required"`
Namespace string `json:"namespace" binding:"required"`
InstanceName string `json:"instance_name" binding:"required"`
InstanceDesc string `json:"instance_desc,omitempty"`
MySQLVersion string `json:"mysql_version"`
Topology string `json:"topology"`
MySQLPort int `json:"mysql_port"`
@@ -815,6 +816,7 @@ func validateDeliveryInput(input MySQLDeliveryInput, _ []string) error {
func normalizeMySQLDeliveryInput(input *MySQLDeliveryInput) {
input.Namespace = normalizeDNSLabel(input.Namespace)
input.InstanceName = normalizeDNSLabel(input.InstanceName)
input.InstanceDesc = strings.TrimSpace(input.InstanceDesc)
input.TargetHost = strings.TrimSpace(input.TargetHost)
input.DataDisk = strings.TrimSpace(input.DataDisk)
if input.MySQLVersion == "" {
@@ -1971,24 +1973,109 @@ func (s *DeliveryService) RetryCloudDMRegistration(ctx context.Context, taskID s
return s.transition(ctx, &task, model.TaskFinished, "CloudDM registration completed", "")
}
type cloudDMRegisterRequest struct {
SourceSystem string `json:"sourceSystem"`
ResourceType string `json:"resourceType"`
ExternalResourceID string `json:"externalResourceId"`
DataSource cloudDMDataSource `json:"dataSource"`
}
type cloudDMDataSource struct {
InstanceName string `json:"instanceName"`
InstanceDesc string `json:"instanceDesc"`
DSType string `json:"dsType"`
Host string `json:"host"`
SecurityType string `json:"securityType"`
UserName string `json:"userName"`
Password string `json:"password"`
DefaultSchema any `json:"defaultSchema"`
ClientTimeZone string `json:"clientTimeZone"`
ConnectTimeoutMs int `json:"connectTimeoutMs"`
SocketTimeoutSecs int `json:"socketTimeoutSeconds"`
ConnectionCharset string `json:"connectionCharset"`
}
func cloudDMExternalResourceID(instanceID uint64) string {
return "mysql-instance:" + strconv.FormatUint(instanceID, 10)
}
func cloudDMClientTimeZone(value string) string {
value = strings.TrimSpace(value)
if value == "" {
return "Asia/Shanghai"
}
return value
}
func buildCloudDMRegisterRequest(instance model.DeploymentResult, payload deliveryPayload, password string) cloudDMRegisterRequest {
description := strings.TrimSpace(payload.InstanceDesc)
if description == "" {
description = instance.InstanceName
}
return cloudDMRegisterRequest{
SourceSystem: "xinfra",
ResourceType: "MYSQL_INSTANCE",
ExternalResourceID: cloudDMExternalResourceID(instance.ID),
DataSource: cloudDMDataSource{
InstanceName: instance.InstanceName,
InstanceDesc: description,
DSType: "MySQL",
Host: net.JoinHostPort(instance.Host, strconv.Itoa(instance.Port)),
SecurityType: "USER_PASSWD",
UserName: "root",
Password: password,
DefaultSchema: nil,
ClientTimeZone: cloudDMClientTimeZone(payload.Timezone),
ConnectTimeoutMs: 5000,
SocketTimeoutSecs: 10,
ConnectionCharset: "utf8",
},
}
}
func (s *DeliveryService) RegisterCloudDM(ctx context.Context, taskID string) error {
if s.cfg.CloudDMRegisterURL == "" {
if strings.TrimSpace(s.cfg.CloudDMRegisterURL) == "" {
return nil
}
if strings.TrimSpace(s.cfg.CloudDMAPIToken) == "" {
return fmt.Errorf("CloudDM registration is enabled but CLOUDDM_API_TOKEN is not configured")
}
var task model.DeliveryTask
if err := s.db.WithContext(ctx).First(&task, "id = ?", taskID).Error; err != nil {
return err
}
var payload deliveryPayload
if err := json.Unmarshal([]byte(task.ImmutablePayload), &payload); err != nil {
return fmt.Errorf("decode delivery payload: %w", err)
}
var instance model.DeploymentResult
if err := s.db.WithContext(ctx).Where("task_id = ? AND component = ? AND service_type = ?", taskID, "mysql", "database").First(&instance).Error; err != nil {
return err
}
body := map[string]any{"name": instance.InstanceName, "host": instance.Host, "port": instance.Port, "username": "root", "database_type": "mysql"}
raw, _ := json.Marshal(body)
var credential model.DeploymentCredential
if err := s.db.WithContext(ctx).
Where("task_id = ? AND username = ? AND account_host = ? AND status IN ?", taskID, "root", "localhost", []string{"pending", "available"}).
First(&credential).Error; err != nil {
return fmt.Errorf("CloudDM registration requires the MySQL root credential: %w", err)
}
password, err := decryptCredential(s.cfg.DeliveryCredentialSecret, credential.Ciphertext, credential.Nonce)
if err != nil {
return fmt.Errorf("decrypt MySQL root credential for CloudDM registration: %w", err)
}
body := buildCloudDMRegisterRequest(instance, payload, password)
raw, err := json.Marshal(body)
if err != nil {
return fmt.Errorf("encode CloudDM registration request: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.CloudDMRegisterURL, bytes.NewReader(raw))
if err != nil {
return err
}
// X-Request-ID is a trace identifier, so generate a fresh UUID for every
// HTTP call. externalResourceId above is the stable business idempotency key.
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(s.cfg.CloudDMAPIToken))
req.Header.Set("Content-Type", "application/json")
if s.cfg.CloudDMAPIToken != "" {
req.Header.Set("Authorization", "Bearer "+s.cfg.CloudDMAPIToken)
}
req.Header.Set("X-Request-ID", randomUUID())
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
server/internal/service/delivery_test.go
+33
View File
@@ -1,6 +1,7 @@
package service
import (
"encoding/json"
"testing"
"time"
@@ -131,6 +132,38 @@ func TestAllocatePort(t *testing.T) {
}
}
func TestBuildCloudDMRegisterRequest(t *testing.T) {
req := buildCloudDMRegisterRequest(
model.DeploymentResult{ID: 42, InstanceName: "mysql-payment-prod", Host: "10.0.0.10", Port: 3306},
deliveryPayload{MySQLDeliveryInput: MySQLDeliveryInput{InstanceDesc: "支付生产 MySQL", Timezone: "Asia/Shanghai"}},
"secret",
)
if req.SourceSystem != "xinfra" || req.ResourceType != "MYSQL_INSTANCE" {
t.Fatalf("unexpected request metadata: %#v", req)
}
if req.ExternalResourceID != "mysql-instance:42" {
t.Fatalf("unexpected externalResourceId: %q", req.ExternalResourceID)
}
if req.DataSource.Host != "10.0.0.10:3306" || req.DataSource.Password != "secret" {
t.Fatalf("unexpected data source fields: %#v", req.DataSource)
}
if req.DataSource.ClientTimeZone != "Asia/Shanghai" || req.DataSource.ConnectionCharset != "utf8" {
t.Fatalf("unexpected time zone or charset: %#v", req.DataSource)
}
raw, err := json.Marshal(req)
if err != nil {
t.Fatalf("marshal CloudDM request: %v", err)
}
var decoded map[string]any
if err := json.Unmarshal(raw, &decoded); err != nil {
t.Fatalf("unmarshal CloudDM request: %v", err)
}
dataSource := decoded["dataSource"].(map[string]any)
if value, ok := dataSource["defaultSchema"]; !ok || value != nil {
t.Fatalf("defaultSchema must be present as null, got %#v", dataSource["defaultSchema"])
}
}
func TestRollbackExtraVarsTargetsOnlyTheAllocatedInstance(t *testing.T) {
task := &model.DeliveryTask{ID: "task-1", TargetHost: "db-01"}
payload := deliveryPayload{MySQLDeliveryInput: MySQLDeliveryInput{InstanceName: "mysql-a", DataDisk: "/disk1"}}