From f8e71251a7791538633d21ade6e395452e2b5e26 Mon Sep 17 00:00:00 2001 From: actiontech-zihan Date: Wed, 22 Jul 2026 16:31:20 +0800 Subject: [PATCH] feat: support redis connection mode in dms services Issue: https://github.com/actiontech/dms-ee/issues/937 --- api/dms/service/v1/db_service.go | 6 +- api/dms/service/v2/db_service.go | 4 +- internal/dms/biz/db_service.go | 14 +- internal/dms/biz/redis_connection_mode.go | 67 ++++++++ internal/dms/service/db_service.go | 73 +++++++-- internal/dms/service/redis_connection_mode.go | 151 ++++++++++++++++++ 6 files changed, 296 insertions(+), 19 deletions(-) create mode 100644 internal/dms/biz/redis_connection_mode.go create mode 100644 internal/dms/service/redis_connection_mode.go diff --git a/api/dms/service/v1/db_service.go b/api/dms/service/v1/db_service.go index d57deb5db..2064029bc 100644 --- a/api/dms/service/v1/db_service.go +++ b/api/dms/service/v1/db_service.go @@ -26,7 +26,7 @@ type DBService struct { Port string `json:"port" validate:"required"` // DB Service admin user // Required: true - User string `json:"user" validate:"required"` + User string `json:"user"` // DB Service admin password // Required: true Password string `json:"password" validate:"required"` @@ -182,7 +182,7 @@ type UpdateDBService struct { Port string `json:"port" validate:"required"` // DB Service admin user // Required: true - User string `json:"user" validate:"required"` + User string `json:"user"` // DB Service admin password Password *string `json:"password"` // DB Service business name @@ -413,7 +413,7 @@ type ListGlobalDBService struct { // TODO This parameter is deprecated and will be removed soon. // the db service business name // Deprecated: the business field is replaced with the environmentTag of the v2 interface. - Business string `json:"business"` + Business string `json:"business"` // DB Service maintenance time MaintenanceTimes []*dmsCommonV1.MaintenanceTime `json:"maintenance_times"` // DB desc diff --git a/api/dms/service/v2/db_service.go b/api/dms/service/v2/db_service.go index f4d4d3442..1bc4788b5 100644 --- a/api/dms/service/v2/db_service.go +++ b/api/dms/service/v2/db_service.go @@ -106,7 +106,7 @@ type DBService struct { Port string `json:"port" validate:"required"` // DB Service admin user // Required: true - User string `json:"user" validate:"required"` + User string `json:"user"` // DB Service admin password // Required: true Password string `json:"password" validate:"required"` @@ -161,7 +161,7 @@ type UpdateDBService struct { Port string `json:"port" validate:"required"` // DB Service admin user // Required: true - User string `json:"user" validate:"required"` + User string `json:"user"` // DB Service admin password Password *string `json:"password"` // DB Service environment tag diff --git a/internal/dms/biz/db_service.go b/internal/dms/biz/db_service.go index bdc4d5d8c..c0fc4ee7b 100644 --- a/internal/dms/biz/db_service.go +++ b/internal/dms/biz/db_service.go @@ -640,6 +640,12 @@ func (d *DBServiceUsecase) TestDbServiceConnection(ctx context.Context, dbServic Value: item.Value, }) } + additionParams, err := appendRedisConnectionModeIfMissing(dbService.DBType, additionParams) + if err != nil { + connectionResult.ConnectionStatus = LastConnectionStatusFailed + connectionResult.ConnectErrorMessage = err.Error() + return connectionResult, err + } checkDbConnectableParams := dmsCommonV1.CheckDbConnectable{ DBType: dbService.DBType, @@ -719,9 +725,11 @@ func (d *DBServiceUsecase) UpdateDBServiceByArgs(ctx context.Context, dbServiceU return fmt.Errorf("update db service db type is unsupported") } - if updateDBService.Host == "" || updateDBService.Port == "" || - updateDBService.User == "" || updateDBService.EnvironmentTagUID == "" { - return fmt.Errorf("db service's host,port,user,environment can't be empty") + if updateDBService.Host == "" || updateDBService.Port == "" || updateDBService.EnvironmentTagUID == "" { + return fmt.Errorf("db service's host,port,environment can't be empty") + } + if updateDBService.User == "" && !isRedisClusterDBServiceArgs(updateDBService) { + return fmt.Errorf("db service user can't be empty") } _, err := d.environmentTagUsecase.GetEnvironmentTagByUID(ctx, updateDBService.EnvironmentTagUID) if err != nil { diff --git a/internal/dms/biz/redis_connection_mode.go b/internal/dms/biz/redis_connection_mode.go new file mode 100644 index 000000000..7ac81ae36 --- /dev/null +++ b/internal/dms/biz/redis_connection_mode.go @@ -0,0 +1,67 @@ +package biz + +import ( + "fmt" + "strings" + + pkgConst "github.com/actiontech/dms/internal/dms/pkg/constant" + dmsCommonV1 "github.com/actiontech/dms/pkg/dms-common/api/dms/v1" + pkgParams "github.com/actiontech/dms/pkg/params" +) + +const ( + redisConnectionModeParam = "connection_mode" + redisConnectionModeStandalone = "standalone" + redisConnectionModeCluster = "cluster" +) + +func isRedisDBType(dbType string) bool { + return strings.EqualFold(dbType, string(pkgConst.DBTypeRedis)) +} + +func normalizeRedisConnectionModeValue(value string) (string, error) { + switch strings.TrimSpace(value) { + case "", redisConnectionModeStandalone: + return redisConnectionModeStandalone, nil + case redisConnectionModeCluster: + return redisConnectionModeCluster, nil + default: + return "", fmt.Errorf("invalid redis connection_mode: %s", value) + } +} + +func isRedisClusterDBServiceArgs(args *BizDBServiceArgs) bool { + if args == nil || !isRedisDBType(args.DBType) { + return false + } + mode, err := normalizeRedisConnectionModeValue(args.AdditionalParams.GetParam(redisConnectionModeParam).String()) + return err == nil && mode == redisConnectionModeCluster +} + +func appendRedisConnectionModeIfMissing(dbType string, additionalParams []*dmsCommonV1.AdditionalParam) ([]*dmsCommonV1.AdditionalParam, error) { + if !isRedisDBType(dbType) { + return additionalParams, nil + } + mode := "" + for _, item := range additionalParams { + if item != nil && item.Name == redisConnectionModeParam { + mode = item.Value + break + } + } + normalizedMode, err := normalizeRedisConnectionModeValue(mode) + if err != nil { + return nil, err + } + for _, item := range additionalParams { + if item != nil && item.Name == redisConnectionModeParam { + item.Value = normalizedMode + return additionalParams, nil + } + } + return append(additionalParams, &dmsCommonV1.AdditionalParam{ + Name: redisConnectionModeParam, + Value: normalizedMode, + Type: string(pkgParams.ParamTypeString), + }), nil +} diff --git a/internal/dms/service/db_service.go b/internal/dms/service/db_service.go index 5987f6b89..e72e75737 100644 --- a/internal/dms/service/db_service.go +++ b/internal/dms/service/db_service.go @@ -41,11 +41,23 @@ func (d *DMSService) UpdateDBService(ctx context.Context, req *dmsV2.UpdateDBSer return err } for _, additionalParam := range req.DBService.AdditionalParams { + if isRedisConnectionModeParam(req.DBService.DBType, additionalParam.Name) { + if err := setRedisConnectionModeParam(&additionalParams, additionalParam.Value); err != nil { + return err + } + continue + } err = additionalParams.SetParamValue(additionalParam.Name, additionalParam.Value) if err != nil { return fmt.Errorf("set param value failed,invalid db type: %s", req.DBService.DBType) } } + if err := normalizeRedisConnectionModeParams(req.DBService.DBType, &additionalParams); err != nil { + return err + } + if err := validateDBServiceUser(req.DBService.DBType, req.DBService.User, additionalParams); err != nil { + return err + } args := &biz.BizDBServiceArgs{ DBType: req.DBService.DBType, @@ -80,6 +92,9 @@ func (d *DMSService) UpdateDBService(ctx context.Context, req *dmsV2.UpdateDBSer } func (d *DMSService) CheckDBServiceIsConnectable(ctx context.Context, req *dmsV1.CheckDBServiceIsConnectableReq) (reply *dmsV1.CheckDBServiceIsConnectableReply, err error) { + if err := normalizeCheckDbConnectable(&req.DBService); err != nil { + return nil, err + } results, err := d.DBServiceUsecase.IsConnectable(ctx, req.DBService) if err != nil { @@ -112,6 +127,10 @@ func (d *DMSService) CheckDBServiceIsConnectableById(ctx context.Context, req *d Value: item.Value, }) } + additionParams, err = appendRedisConnectionModeIfMissing(dbService.DBType, additionParams) + if err != nil { + return nil, err + } checkDbConnectableParams := dmsCommonV1.CheckDbConnectable{ DBType: dbService.DBType, @@ -236,12 +255,27 @@ func (d *DMSService) AddDBService(ctx context.Context, req *dmsV1.AddDBServiceRe }() additionalParams, err := d.DBServiceUsecase.GetDriverParamsByDBType(ctx, req.DBService.DBType) + if err != nil { + return nil, err + } for _, additionalParam := range req.DBService.AdditionalParams { + if isRedisConnectionModeParam(req.DBService.DBType, additionalParam.Name) { + if err := setRedisConnectionModeParam(&additionalParams, additionalParam.Value); err != nil { + return nil, err + } + continue + } err = additionalParams.SetParamValue(additionalParam.Name, additionalParam.Value) if err != nil { return nil, fmt.Errorf("set param value failed,invalid db type: %s", req.DBService.DBType) } } + if err := normalizeRedisConnectionModeParams(req.DBService.DBType, &additionalParams); err != nil { + return nil, err + } + if err := validateDBServiceUser(req.DBService.DBType, req.DBService.User, additionalParams); err != nil { + return nil, err + } args := &biz.BizDBServiceArgs{ Name: req.DBService.Name, @@ -291,12 +325,27 @@ func (d *DMSService) AddDBServiceV2(ctx context.Context, req *dmsV2.AddDBService }() additionalParams, err := d.DBServiceUsecase.GetDriverParamsByDBType(ctx, req.DBService.DBType) + if err != nil { + return nil, err + } for _, additionalParam := range req.DBService.AdditionalParams { + if isRedisConnectionModeParam(req.DBService.DBType, additionalParam.Name) { + if err := setRedisConnectionModeParam(&additionalParams, additionalParam.Value); err != nil { + return nil, err + } + continue + } err = additionalParams.SetParamValue(additionalParam.Name, additionalParam.Value) if err != nil { return nil, fmt.Errorf("set param value failed,invalid db type: %s", req.DBService.DBType) } } + if err := normalizeRedisConnectionModeParams(req.DBService.DBType, &additionalParams); err != nil { + return nil, err + } + if err := validateDBServiceUser(req.DBService.DBType, req.DBService.User, additionalParams); err != nil { + return nil, err + } args := &biz.BizDBServiceArgs{ Name: req.DBService.Name, @@ -671,18 +720,20 @@ func (d *DMSService) ListDBServices(ctx context.Context, req *dmsCommonV2.ListDB ret[i].LastConnectionTestErrorMessage = *u.LastConnectionErrorMsg } - if u.AdditionalParams != nil { - additionalParams := make([]*dmsCommonV1.AdditionalParam, 0, len(u.AdditionalParams)) - for _, item := range u.AdditionalParams { - additionalParams = append(additionalParams, &dmsCommonV1.AdditionalParam{ - Name: item.Key, - Value: item.Value, - Description: item.Desc, - Type: string(item.Type), - }) - } - ret[i].AdditionalParams = additionalParams + additionalParams := make([]*dmsCommonV1.AdditionalParam, 0, len(u.AdditionalParams)+1) + for _, item := range u.AdditionalParams { + additionalParams = append(additionalParams, &dmsCommonV1.AdditionalParam{ + Name: item.Key, + Value: item.Value, + Description: item.Desc, + Type: string(item.Type), + }) + } + additionalParams, err = appendRedisConnectionModeIfMissing(u.DBType, additionalParams) + if err != nil { + return nil, err } + ret[i].AdditionalParams = additionalParams if u.SQLEConfig != nil { sqlConfig := &dmsCommonV1.SQLEConfig{ diff --git a/internal/dms/service/redis_connection_mode.go b/internal/dms/service/redis_connection_mode.go new file mode 100644 index 000000000..f3057634d --- /dev/null +++ b/internal/dms/service/redis_connection_mode.go @@ -0,0 +1,151 @@ +package service + +import ( + "fmt" + "strings" + + pkgConst "github.com/actiontech/dms/internal/dms/pkg/constant" + dmsCommonV1 "github.com/actiontech/dms/pkg/dms-common/api/dms/v1" + "github.com/actiontech/dms/pkg/params" +) + +const ( + redisConnectionModeParam = "connection_mode" + redisConnectionModeStandalone = "standalone" + redisConnectionModeCluster = "cluster" +) + +func isRedisDBType(dbType string) bool { + return strings.EqualFold(dbType, string(pkgConst.DBTypeRedis)) +} + +func isRedisConnectionModeParam(dbType, name string) bool { + return isRedisDBType(dbType) && name == redisConnectionModeParam +} + +func normalizeRedisConnectionModeValue(value string) (string, error) { + switch strings.TrimSpace(value) { + case "", redisConnectionModeStandalone: + return redisConnectionModeStandalone, nil + case redisConnectionModeCluster: + return redisConnectionModeCluster, nil + default: + return "", fmt.Errorf("invalid redis connection_mode: %s", value) + } +} + +func setRedisConnectionModeParam(additionalParams *params.Params, value string) error { + mode, err := normalizeRedisConnectionModeValue(value) + if err != nil { + return err + } + if additionalParams == nil { + return fmt.Errorf("additional params is nil") + } + if param := additionalParams.GetParam(redisConnectionModeParam); param != nil { + param.Value = mode + return nil + } + *additionalParams = append(*additionalParams, ¶ms.Param{ + Key: redisConnectionModeParam, + Value: mode, + Desc: "Redis connection mode", + Type: params.ParamTypeString, + }) + return nil +} + +func normalizeRedisConnectionModeParams(dbType string, additionalParams *params.Params) error { + if !isRedisDBType(dbType) { + return nil + } + value := "" + if additionalParams != nil { + value = additionalParams.GetParam(redisConnectionModeParam).String() + } + return setRedisConnectionModeParam(additionalParams, value) +} + +func validateDBServiceUser(dbType, user string, additionalParams params.Params) error { + if isRedisDBType(dbType) { + mode, err := normalizeRedisConnectionModeValue(additionalParams.GetParam(redisConnectionModeParam).String()) + if err != nil { + return err + } + if mode == redisConnectionModeCluster { + return nil + } + } + if user == "" { + return fmt.Errorf("db service user can't be empty") + } + return nil +} + +func normalizeCheckDbConnectable(dbService *dmsCommonV1.CheckDbConnectable) error { + if dbService == nil || !isRedisDBType(dbService.DBType) { + if dbService != nil && dbService.User == "" { + return fmt.Errorf("db service user can't be empty") + } + return nil + } + mode := "" + for _, item := range dbService.AdditionalParams { + if item != nil && item.Name == redisConnectionModeParam { + mode = item.Value + break + } + } + normalizedMode, err := normalizeRedisConnectionModeValue(mode) + if err != nil { + return err + } + found := false + for _, item := range dbService.AdditionalParams { + if item != nil && item.Name == redisConnectionModeParam { + item.Value = normalizedMode + found = true + break + } + } + if !found { + dbService.AdditionalParams = append(dbService.AdditionalParams, &dmsCommonV1.AdditionalParam{ + Name: redisConnectionModeParam, + Value: normalizedMode, + Type: string(params.ParamTypeString), + }) + } + if normalizedMode != redisConnectionModeCluster && dbService.User == "" { + return fmt.Errorf("db service user can't be empty") + } + return nil +} + +func appendRedisConnectionModeIfMissing(dbType string, additionalParams []*dmsCommonV1.AdditionalParam) ([]*dmsCommonV1.AdditionalParam, error) { + if !isRedisDBType(dbType) { + return additionalParams, nil + } + mode := "" + for _, item := range additionalParams { + if item != nil && item.Name == redisConnectionModeParam { + mode = item.Value + break + } + } + normalizedMode, err := normalizeRedisConnectionModeValue(mode) + if err != nil { + return nil, err + } + for _, item := range additionalParams { + if item != nil && item.Name == redisConnectionModeParam { + item.Value = normalizedMode + return additionalParams, nil + } + } + return append(additionalParams, &dmsCommonV1.AdditionalParam{ + Name: redisConnectionModeParam, + Value: normalizedMode, + Description: "Redis connection mode", + Type: string(params.ParamTypeString), + }), nil +}