Skip to content

Commit f26bb6f

Browse files
iwanghciwanghc
authored
Merge pull request #633 from actiontech/dev/issue-2902-mongodb-odc-ce
feat(sql_workbench): sync mongodb datasource metadata to ODC (#2902)
2 parents 0c651d4 + 9daab0f commit f26bb6f

5 files changed

Lines changed: 201 additions & 65 deletions

File tree

‎internal/dms/pkg/constant/const.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -256,6 +256,8 @@ func ParseDBType(s string) (DBType, error) {
256256
return DBTypeHANA, nil
257257
case "PolarDB For MySQL":
258258
return DBTypePolarDBForMySQL, nil
259+
case "MongoDB":
260+
return DBTypeMongoDB, nil
259261
case "OceanBase For Oracle":
260262
return DBTypeOceanBaseOracle, nil
261263

@@ -281,6 +283,7 @@ const (
281283
DBTypeGaussDBForMySQL DBType = "GaussDB for MySQL"
282284
DBTypeHANA DBType = "HANA"
283285
DBTypePolarDBForMySQL DBType = "PolarDB For MySQL"
286+
DBTypeMongoDB DBType = "MongoDB"
284287
DBTypeOceanBaseOracle DBType = "OceanBase For Oracle"
285288
)
286289

‎internal/dms/pkg/constant/const_test.go‎

Lines changed: 26 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -7,10 +7,10 @@ import (
77
func TestCheckDBTypeIfDataExportSupported_NewTypes(t *testing.T) {
88
// 验证新增的数据源应在白名单中
99
newTypes := map[string]bool{
10-
"TiDB": true,
11-
"TDSQL For InnoDB": true,
12-
"GoldenDB": true,
13-
"TBase": true,
10+
"TiDB": true,
11+
"TDSQL For InnoDB": true,
12+
"GoldenDB": true,
13+
"TBase": true,
1414
// Issue #2868: GaussDB (PostgreSQL 协议) 与 GaussDB for MySQL (MySQL 协议) 是两个独立产品
1515
"GaussDB": true, // PostgreSQL 协议 GaussDB / openGauss, 走 opengauss-connector-go-pq 驱动
1616
"GaussDB for MySQL": true, // 华为云 GaussDB(for MySQL), 走 MySQL 驱动
@@ -51,9 +51,9 @@ func TestCheckDBTypeIfDataExportSupported_ExistingTypes(t *testing.T) {
5151

5252
func TestParseDBType_PolarDB(t *testing.T) {
5353
cases := map[string]struct {
54-
input string
55-
wantDBType DBType
56-
wantErr bool
54+
input string
55+
wantDBType DBType
56+
wantErr bool
5757
}{
5858
"valid PolarDB For MySQL": {
5959
input: "PolarDB For MySQL",
@@ -109,28 +109,29 @@ func TestParseDBType(t *testing.T) {
109109
expected DBType
110110
expectError bool
111111
}{
112-
"MySQL": {input: "MySQL", expected: DBTypeMySQL},
113-
"TDSQL For InnoDB": {input: "TDSQL For InnoDB", expected: DBTypeTDSQLForInnoDB},
114-
"TiDB": {input: "TiDB", expected: DBTypeTiDB},
115-
"PostgreSQL": {input: "PostgreSQL", expected: DBTypePostgreSQL},
116-
"Oracle": {input: "Oracle", expected: DBTypeOracle},
117-
"DB2": {input: "DB2", expected: DBTypeDB2},
118-
"SQL Server": {input: "SQL Server", expected: DBTypeSQLServer},
112+
"MySQL": {input: "MySQL", expected: DBTypeMySQL},
113+
"TDSQL For InnoDB": {input: "TDSQL For InnoDB", expected: DBTypeTDSQLForInnoDB},
114+
"TiDB": {input: "TiDB", expected: DBTypeTiDB},
115+
"PostgreSQL": {input: "PostgreSQL", expected: DBTypePostgreSQL},
116+
"Oracle": {input: "Oracle", expected: DBTypeOracle},
117+
"DB2": {input: "DB2", expected: DBTypeDB2},
118+
"SQL Server": {input: "SQL Server", expected: DBTypeSQLServer},
119119
"OceanBase For MySQL": {input: "OceanBase For MySQL", expected: DBTypeOceanBaseMySQL},
120-
"GoldenDB": {input: "GoldenDB", expected: DBTypeGoldenDB},
121-
"TBase": {input: "TBase", expected: DBTypeTBase},
122-
"Hive": {input: "Hive", expected: DBTypeHive},
123-
"DM": {input: "DM", expected: DBTypeDM},
120+
"GoldenDB": {input: "GoldenDB", expected: DBTypeGoldenDB},
121+
"TBase": {input: "TBase", expected: DBTypeTBase},
122+
"Hive": {input: "Hive", expected: DBTypeHive},
123+
"DM": {input: "DM", expected: DBTypeDM},
124124
// Issue #2868: 拆分 GaussDB / GaussDB for MySQL 为两个独立产品
125-
"GaussDB": {input: "GaussDB", expected: DBTypeGaussDB},
126-
"GaussDB for MySQL": {input: "GaussDB for MySQL", expected: DBTypeGaussDBForMySQL},
127-
"HANA": {input: "HANA", expected: DBTypeHANA},
125+
"GaussDB": {input: "GaussDB", expected: DBTypeGaussDB},
126+
"GaussDB for MySQL": {input: "GaussDB for MySQL", expected: DBTypeGaussDBForMySQL},
127+
"HANA": {input: "HANA", expected: DBTypeHANA},
128128
// PolarDB-MySQL 新增 (Issue #826)
129-
"PolarDB For MySQL": {input: "PolarDB For MySQL", expected: DBTypePolarDBForMySQL},
129+
"PolarDB For MySQL": {input: "PolarDB For MySQL", expected: DBTypePolarDBForMySQL},
130+
"MongoDB": {input: "MongoDB", expected: DBTypeMongoDB},
130131
// "PolarDB" 单独不应匹配
131-
"PolarDB only": {input: "PolarDB", expectError: true},
132-
"invalid type": {input: "UnknownDB", expectError: true},
133-
"empty string": {input: "", expectError: true},
132+
"PolarDB only": {input: "PolarDB", expectError: true},
133+
"invalid type": {input: "UnknownDB", expectError: true},
134+
"empty string": {input: "", expectError: true},
134135
}
135136

136137
for name, tc := range tests {

‎internal/sql_workbench/client/sql_workbench_client.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -398,7 +398,7 @@ func (c *SqlWorkbenchClient) CreateDatasources(datasource CreateDatasourceReques
398398
// 检查HTTP状态码
399399
if resp.StatusCode != http.StatusOK {
400400
c.log.Errorf("Create datasource failed with status code: %d, response: %s", resp.StatusCode, string(body))
401-
return nil, fmt.Errorf("create datasource failed with status code: %d", resp.StatusCode)
401+
return nil, fmt.Errorf("create datasource failed with status code: %d, response: %s", resp.StatusCode, string(body))
402402
}
403403

404404
// 解析响应

‎internal/sql_workbench/service/sql_workbench_service.go‎

Lines changed: 94 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -851,8 +851,20 @@ type datasourceBaseInfo struct {
851851
ServiceName *string
852852
EnvironmentID int64
853853
DefaultSchema *string
854+
Properties interface{}
855+
JDBCParams map[string]interface{}
854856
}
855857

858+
const (
859+
mongoDefaultDatabaseParam = "default_database"
860+
mongoAuthDatabaseParam = "auth_source"
861+
mongoAuthMechanismParam = "auth_mechanism"
862+
mongoReplicaSetParam = "replica_set"
863+
mongoTLSEnabledParam = "tls"
864+
mongoDirectConnectionParam = "direct_connection"
865+
mongoTLSSkipVerifyParam = "tls_skip_verify"
866+
)
867+
856868
// buildDatasourceBaseInfo 构建数据源基础信息
857869
func (sqlWorkbenchService *SqlWorkbenchService) buildDatasourceBaseInfo(ctx context.Context, dbService *biz.DBService, environmentID int64) (*datasourceBaseInfo, error) {
858870
datasourceName, err := sqlWorkbenchService.buildDatasourceName(ctx, dbService)
@@ -876,6 +888,10 @@ func (sqlWorkbenchService *SqlWorkbenchService) buildDatasourceBaseInfo(ctx cont
876888
baseInfo.ServiceName = &serviceName
877889
}
878890

891+
if dbService.DBType == string(pkgConst.DBTypeMongoDB) {
892+
baseInfo.DefaultSchema, baseInfo.Properties, baseInfo.JDBCParams = buildMongoDatasourceOptions(dbService)
893+
}
894+
879895
return baseInfo, nil
880896
}
881897

@@ -887,17 +903,19 @@ func (sqlWorkbenchService *SqlWorkbenchService) buildCreateDatasourceRequest(ctx
887903
}
888904

889905
return client.CreateDatasourceRequest{
890-
CreatorID: sqlWorkbenchUser.SqlWorkbenchUserId,
891-
Type: baseInfo.Type,
892-
Name: baseInfo.Name,
893-
Username: baseInfo.Username,
894-
Password: baseInfo.Password,
895-
Host: baseInfo.Host,
896-
Port: baseInfo.Port,
897-
ServiceName: baseInfo.ServiceName,
898-
SSLConfig: client.SSLConfig{Enabled: false},
899-
EnvironmentID: baseInfo.EnvironmentID,
900-
DefaultSchema: baseInfo.DefaultSchema,
906+
CreatorID: sqlWorkbenchUser.SqlWorkbenchUserId,
907+
Type: baseInfo.Type,
908+
Name: baseInfo.Name,
909+
Username: baseInfo.Username,
910+
Password: baseInfo.Password,
911+
Host: baseInfo.Host,
912+
Port: baseInfo.Port,
913+
ServiceName: baseInfo.ServiceName,
914+
SSLConfig: client.SSLConfig{Enabled: false},
915+
Properties: baseInfo.Properties,
916+
EnvironmentID: baseInfo.EnvironmentID,
917+
JdbcURLParameters: baseInfo.JDBCParams,
918+
DefaultSchema: baseInfo.DefaultSchema,
901919
}, nil
902920
}
903921

@@ -909,16 +927,18 @@ func (sqlWorkbenchService *SqlWorkbenchService) buildUpdateDatasourceRequest(ctx
909927
}
910928

911929
return client.UpdateDatasourceRequest{
912-
Type: baseInfo.Type,
913-
Name: &baseInfo.Name,
914-
Username: baseInfo.Username,
915-
Password: &baseInfo.Password,
916-
Host: baseInfo.Host,
917-
Port: baseInfo.Port,
918-
ServiceName: baseInfo.ServiceName,
919-
SSLConfig: client.SSLConfig{Enabled: false},
920-
EnvironmentID: baseInfo.EnvironmentID,
921-
DefaultSchema: baseInfo.DefaultSchema,
930+
Type: baseInfo.Type,
931+
Name: &baseInfo.Name,
932+
Username: baseInfo.Username,
933+
Password: &baseInfo.Password,
934+
Host: baseInfo.Host,
935+
Port: baseInfo.Port,
936+
ServiceName: baseInfo.ServiceName,
937+
SSLConfig: client.SSLConfig{Enabled: false},
938+
Properties: interfacePtr(baseInfo.Properties),
939+
EnvironmentID: baseInfo.EnvironmentID,
940+
JdbcURLParameters: mapPtr(baseInfo.JDBCParams),
941+
DefaultSchema: baseInfo.DefaultSchema,
922942
}, nil
923943
}
924944

@@ -950,6 +970,8 @@ func (sqlWorkbenchService *SqlWorkbenchService) convertDBType(dmsDBType string)
950970
return "MYSQL"
951971
case "PolarDB For MySQL":
952972
return "MYSQL"
973+
case "MongoDB":
974+
return "MONGODB"
953975
default:
954976
return dmsDBType
955977
}
@@ -966,6 +988,57 @@ func (sqlWorkbenchService *SqlWorkbenchService) SupportDBType(dbType pkgConst.DB
966988
dbType == pkgConst.DBTypePolarDBForMySQL
967989
}
968990

991+
func buildMongoDatasourceOptions(dbService *biz.DBService) (*string, interface{}, map[string]interface{}) {
992+
defaultDatabase := dbService.AdditionalParams.GetParam(mongoDefaultDatabaseParam).String()
993+
var defaultSchema *string
994+
if defaultDatabase != "" {
995+
defaultSchema = &defaultDatabase
996+
}
997+
998+
jdbcParams := map[string]interface{}{}
999+
if authDB := dbService.AdditionalParams.GetParam(mongoAuthDatabaseParam).String(); authDB != "" {
1000+
jdbcParams["authSource"] = authDB
1001+
}
1002+
if authMechanism := dbService.AdditionalParams.GetParam(mongoAuthMechanismParam).String(); authMechanism != "" {
1003+
jdbcParams["authMechanism"] = authMechanism
1004+
}
1005+
if replicaSet := dbService.AdditionalParams.GetParam(mongoReplicaSetParam).String(); replicaSet != "" {
1006+
jdbcParams["replicaSet"] = replicaSet
1007+
}
1008+
if tlsParam := dbService.AdditionalParams.GetParam(mongoTLSEnabledParam); tlsParam != nil && tlsParam.String() != "" {
1009+
if tlsParam.Bool() {
1010+
jdbcParams["tls"] = "true"
1011+
} else {
1012+
jdbcParams["tls"] = "false"
1013+
}
1014+
}
1015+
if dbService.AdditionalParams.GetParam(mongoDirectConnectionParam).Bool() {
1016+
jdbcParams["directConnection"] = true
1017+
}
1018+
if dbService.AdditionalParams.GetParam(mongoTLSSkipVerifyParam).Bool() {
1019+
jdbcParams["tlsInsecure"] = true
1020+
}
1021+
1022+
if len(jdbcParams) == 0 {
1023+
return defaultSchema, nil, nil
1024+
}
1025+
return defaultSchema, nil, jdbcParams
1026+
}
1027+
1028+
func interfacePtr(v interface{}) *interface{} {
1029+
if v == nil {
1030+
return nil
1031+
}
1032+
return &v
1033+
}
1034+
1035+
func mapPtr(v map[string]interface{}) *map[string]interface{} {
1036+
if len(v) == 0 {
1037+
return nil
1038+
}
1039+
return &v
1040+
}
1041+
9691042
// buildDatabaseUser 当是ob-mysql时需要给账号管理的账号附加租户名集群名等字符: root@oms_mysql#oms_resource_4250
9701043
func buildDatabaseUser(account string, dbServiceUser string, dbType string) string {
9711044
if dbType == string(pkgConst.DBTypeOceanBaseMySQL) {

‎internal/sql_workbench/service/sql_workbench_service_test.go‎

Lines changed: 77 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,9 @@ package sql_workbench
33
import (
44
"testing"
55

6+
"github.com/actiontech/dms/internal/dms/biz"
67
pkgConst "github.com/actiontech/dms/internal/dms/pkg/constant"
8+
pkgParams "github.com/actiontech/dms/pkg/params"
79
)
810

911
func Test_convertDBType(t *testing.T) {
@@ -12,17 +14,18 @@ func Test_convertDBType(t *testing.T) {
1214
input string
1315
expected string
1416
}{
15-
"DM": {input: "达梦(DM)", expected: "DM"},
16-
"MySQL": {input: "MySQL", expected: "MYSQL"},
17-
"PostgreSQL": {input: "PostgreSQL", expected: "POSTGRESQL"},
18-
"Oracle": {input: "Oracle", expected: "ORACLE"},
19-
"SQL Server": {input: "SQL Server", expected: "SQL_SERVER"},
20-
"OB Oracle": {input: "OceanBase For Oracle", expected: "OB_ORACLE"},
21-
"OB MySQL": {input: "OceanBase For MySQL", expected: "OB_MYSQL"},
17+
"DM": {input: "达梦(DM)", expected: "DM"},
18+
"MySQL": {input: "MySQL", expected: "MYSQL"},
19+
"PostgreSQL": {input: "PostgreSQL", expected: "POSTGRESQL"},
20+
"Oracle": {input: "Oracle", expected: "ORACLE"},
21+
"SQL Server": {input: "SQL Server", expected: "SQL_SERVER"},
22+
"OB Oracle": {input: "OceanBase For Oracle", expected: "OB_ORACLE"},
23+
"OB MySQL": {input: "OceanBase For MySQL", expected: "OB_MYSQL"},
2224
"TiDB": {input: "TiDB", expected: "TIDB"},
23-
"TDSQL For InnoDB": {input: "TDSQL For InnoDB", expected: "MYSQL"},
25+
"TDSQL For InnoDB": {input: "TDSQL For InnoDB", expected: "MYSQL"},
2426
"GoldenDB": {input: "GoldenDB", expected: "MYSQL"},
2527
"PolarDB For MySQL": {input: "PolarDB For MySQL", expected: "MYSQL"},
28+
"MongoDB": {input: "MongoDB", expected: "MONGODB"},
2629
"Unknown passthrough": {input: "UnknownDB", expected: "UnknownDB"},
2730
}
2831
for name, tc := range cases {
@@ -41,16 +44,17 @@ func Test_SupportDBType(t *testing.T) {
4144
input pkgConst.DBType
4245
expected bool
4346
}{
44-
"DM supported": {input: pkgConst.DBTypeDM, expected: true},
45-
"MySQL supported": {input: pkgConst.DBTypeMySQL, expected: true},
46-
"Oracle supported": {input: pkgConst.DBTypeOracle, expected: true},
47-
"OB MySQL supported": {input: pkgConst.DBTypeOceanBaseMySQL, expected: true},
48-
"TiDB supported": {input: pkgConst.DBTypeTiDB, expected: true},
49-
"TDSQL supported": {input: pkgConst.DBTypeTDSQLForInnoDB, expected: true},
50-
"GoldenDB supported": {input: pkgConst.DBTypeGoldenDB, expected: true},
51-
"PostgreSQL unsupported": {input: pkgConst.DBTypePostgreSQL, expected: false},
52-
"SQL Server unsupported": {input: pkgConst.DBTypeSQLServer, expected: false},
53-
"PolarDB For MySQL supported": {input: pkgConst.DBTypePolarDBForMySQL, expected: true},
47+
"DM supported": {input: pkgConst.DBTypeDM, expected: true},
48+
"MySQL supported": {input: pkgConst.DBTypeMySQL, expected: true},
49+
"Oracle supported": {input: pkgConst.DBTypeOracle, expected: true},
50+
"OB MySQL supported": {input: pkgConst.DBTypeOceanBaseMySQL, expected: true},
51+
"TiDB supported": {input: pkgConst.DBTypeTiDB, expected: true},
52+
"TDSQL supported": {input: pkgConst.DBTypeTDSQLForInnoDB, expected: true},
53+
"GoldenDB supported": {input: pkgConst.DBTypeGoldenDB, expected: true},
54+
"MongoDB unsupported": {input: pkgConst.DBTypeMongoDB, expected: false},
55+
"PostgreSQL unsupported": {input: pkgConst.DBTypePostgreSQL, expected: false},
56+
"SQL Server unsupported": {input: pkgConst.DBTypeSQLServer, expected: false},
57+
"PolarDB For MySQL supported": {input: pkgConst.DBTypePolarDBForMySQL, expected: true},
5458
}
5559
for name, tc := range cases {
5660
t.Run(name, func(t *testing.T) {
@@ -61,3 +65,58 @@ func Test_SupportDBType(t *testing.T) {
6165
})
6266
}
6367
}
68+
69+
func Test_buildMongoDatasourceOptions(t *testing.T) {
70+
defaultDB := "appdb"
71+
defaultSchema, propertiesValue, jdbcParams := buildMongoDatasourceOptions(&biz.DBService{
72+
DBType: string(pkgConst.DBTypeMongoDB),
73+
Host: "127.0.0.1",
74+
Port: "27017",
75+
AdditionalParams: pkgParams.Params{
76+
&pkgParams.Param{Key: mongoDefaultDatabaseParam, Value: defaultDB, Type: pkgParams.ParamTypeString},
77+
&pkgParams.Param{Key: mongoAuthDatabaseParam, Value: "admin", Type: pkgParams.ParamTypeString},
78+
&pkgParams.Param{Key: mongoAuthMechanismParam, Value: "SCRAM-SHA-256", Type: pkgParams.ParamTypeString},
79+
&pkgParams.Param{Key: mongoReplicaSetParam, Value: "rs0", Type: pkgParams.ParamTypeString},
80+
&pkgParams.Param{Key: mongoTLSEnabledParam, Value: "true", Type: pkgParams.ParamTypeBool},
81+
&pkgParams.Param{Key: mongoDirectConnectionParam, Value: "true", Type: pkgParams.ParamTypeBool},
82+
&pkgParams.Param{Key: mongoTLSSkipVerifyParam, Value: "true", Type: pkgParams.ParamTypeBool},
83+
},
84+
})
85+
if defaultSchema == nil || *defaultSchema != defaultDB {
86+
t.Fatalf("unexpected default schema: %#v", defaultSchema)
87+
}
88+
if propertiesValue != nil {
89+
t.Fatalf("expected nil properties, got %#v", propertiesValue)
90+
}
91+
if jdbcParams["authSource"] != "admin" {
92+
t.Fatalf("unexpected authSource: %#v", jdbcParams["authSource"])
93+
}
94+
if jdbcParams["authMechanism"] != "SCRAM-SHA-256" {
95+
t.Fatalf("unexpected authMechanism: %#v", jdbcParams["authMechanism"])
96+
}
97+
if jdbcParams["replicaSet"] != "rs0" {
98+
t.Fatalf("unexpected replicaSet: %#v", jdbcParams["replicaSet"])
99+
}
100+
if jdbcParams["tls"] != "true" {
101+
t.Fatalf("unexpected tls: %#v", jdbcParams["tls"])
102+
}
103+
if jdbcParams["directConnection"] != true || jdbcParams["tlsInsecure"] != true {
104+
t.Fatalf("unexpected jdbc params: %#v", jdbcParams)
105+
}
106+
}
107+
108+
func Test_buildMongoDatasourceOptions_tlsOnly(t *testing.T) {
109+
_, propertiesValue, jdbcParams := buildMongoDatasourceOptions(&biz.DBService{
110+
DBType: string(pkgConst.DBTypeMongoDB),
111+
AdditionalParams: pkgParams.Params{
112+
&pkgParams.Param{Key: mongoTLSEnabledParam, Value: "true", Type: pkgParams.ParamTypeBool},
113+
},
114+
})
115+
if propertiesValue != nil {
116+
t.Fatalf("expected nil properties, got %#v", propertiesValue)
117+
}
118+
if jdbcParams["tls"] != "true" {
119+
t.Fatalf("expected tls in jdbcUrlParameters when only tls is configured, got %#v", jdbcParams)
120+
}
121+
}
122+

0 commit comments

Comments
 (0)