feat: support Redis cluster connection mode

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

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

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 {

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
+}

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{