From 0aac99c359641c5fc3163a9b64bca58c4ec83bcf Mon Sep 17 00:00:00 2001 From: guoyonghao <1711322114@qq.com> Date: Wed, 29 Jul 2026 18:24:36 +0800 Subject: [PATCH] feat: register MySQL instances in CloudDM --- server/internal/service/delivery.go | 99 ++++++++++++++++++++++-- server/internal/service/delivery_test.go | 33 ++++++++ 2 files changed, 126 insertions(+), 6 deletions(-) diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index cf9a06a..2d51770 100644 --- a/server/internal/service/delivery.go +++ b/server/internal/service/delivery.go @@ -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 diff --git a/server/internal/service/delivery_test.go b/server/internal/service/delivery_test.go index 52f8049..c66af85 100644 --- a/server/internal/service/delivery_test.go +++ b/server/internal/service/delivery_test.go @@ -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"}}