feat：kuscia支持达梦数据源（#739） (#962)

kongxiaoran · web-flow · commit 447d9b7914b9 · 2025-12-11T13:46:31.000+08:00
diff --git a/etc/conf/kuscia.yaml b/etc/conf/kuscia.yaml
@@ -88,6 +88,7 @@ dataMesh:
       dataSourceTypes:                # the type of datasource that data proxy supported
         - "odps"                      # odps also call as Aliyun MaxCompute
         - "hive"
+        - "dameng"                    # dameng database
 # Note: Currently only a single odps data source configuration is supported. If two odps configurations are defined here, the latter one will override the previous one.
 #   - endpoint: "example-dataproxy-grpc:8024"
 #     dataSourceTypes:
diff --git a/pkg/common/constants.go b/pkg/common/constants.go
@@ -229,6 +229,7 @@ const (
 	DomainDataSourceTypeODPS           = "odps"
 	DomainDataSourceTypePostgreSQL     = "postgresql"
 	DomainDataSourceTypeHive           = "hive"
+	DomainDataSourceTypeDameng         = "dameng"
 	DefaultDomainDataSourceLocalFSPath = "var/storage/data"
 )
 
diff --git a/pkg/datamesh/metaserver/service/domaindatasource.go b/pkg/datamesh/metaserver/service/domaindatasource.go
@@ -221,6 +221,7 @@ func parseDataSourceURI(sourceType string, info *datamesh.DataSourceInfo) (uri s
 	case common.DomainDataSourceTypeMysql:
 	case common.DomainDataSourceTypePostgreSQL:
 	case common.DomainDataSourceTypeHive:
+	case common.DomainDataSourceTypeDameng:
 		if isInvalid(info.Database == nil) {
 			return
 		}
@@ -231,7 +232,7 @@ func parseDataSourceURI(sourceType string, info *datamesh.DataSourceInfo) (uri s
 		}
 		uri = info.Odps.Endpoint + "/" + info.Odps.Project
 	default:
-		err = fmt.Errorf("datasource type:%q not support, only support [localfs,oss,mysql,odps,postgresql,hive]", sourceType)
+		err = fmt.Errorf("datasource type:%q not support, only support [localfs,oss,mysql,odps,postgresql,hive,dameng]", sourceType)
 		nlog.Error(err)
 		return
 	}
@@ -285,6 +286,7 @@ func decodeDataSourceInfo(sourceType string, connectionStr string) (*datamesh.Da
 	case common.DomainDataSourceTypeMysql:
 	case common.DomainDataSourceTypePostgreSQL:
 	case common.DomainDataSourceTypeHive:
+	case common.DomainDataSourceTypeDameng:
 		dsInfo.Database = &datamesh.DatabaseDataSourceInfo{}
 		err = json.Unmarshal(connectionBytes, dsInfo.Database)
 	case common.DomainDataSourceTypeLocalFS:
diff --git a/pkg/datamesh/metaserver/service/domaindatasource_test.go b/pkg/datamesh/metaserver/service/domaindatasource_test.go
@@ -70,6 +70,19 @@ func makeDomainDataSourceService(t *testing.T, conf *config.DataMeshConfig) IDom
 	return NewDomainDataSourceService(conf, makeConfigService(t))
 }
 
+func TestParseDataSourceURI_UnsupportedType(t *testing.T) {
+	info := &datamesh.DataSourceInfo{
+		Localfs: &datamesh.LocalDataSourceInfo{
+			Path: "/test/path",
+		},
+	}
+	uri, err := parseDataSourceURI("unsupported_type", info)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "not support")
+	assert.Contains(t, err.Error(), "unsupported_type")
+	assert.Empty(t, uri)
+}
+
 func makeConfigService(t *testing.T) cmservice.IConfigService {
 	privateKey, err := rsa.GenerateKey(rand.Reader, 1024)
 	assert.Nil(t, err)
diff --git a/pkg/kusciaapi/service/domaindata_source.go b/pkg/kusciaapi/service/domaindata_source.go
@@ -492,8 +492,9 @@ func validateDataSourceType(t string) error {
 		t != common.DomainDataSourceTypeLocalFS &&
 		t != common.DomainDataSourceTypeODPS &&
 		t != common.DomainDataSourceTypePostgreSQL &&
-		t != common.DomainDataSourceTypeHive {
-		return fmt.Errorf("domain data source type %q doesn't support, the available types are [localfs,oss,mysql,odps,postgresql,hive]", t)
+		t != common.DomainDataSourceTypeHive &&
+		t != common.DomainDataSourceTypeDameng {
+		return fmt.Errorf("domain data source type %q doesn't support, the available types are [localfs,oss,mysql,odps,postgresql,hive,dameng]", t)
 	}
 	return nil
 }
@@ -595,7 +596,7 @@ func decodeDataSourceInfo(sourceType string, connectionStr string) (*kusciaapi.D
 	case common.DomainDataSourceTypeOSS:
 		dsInfo.Oss = &kusciaapi.OssDataSourceInfo{}
 		err = json.Unmarshal(connectionBytes, dsInfo.Oss)
-	case common.DomainDataSourceTypeMysql, common.DomainDataSourceTypePostgreSQL, common.DomainDataSourceTypeHive:
+	case common.DomainDataSourceTypeMysql, common.DomainDataSourceTypePostgreSQL, common.DomainDataSourceTypeHive, common.DomainDataSourceTypeDameng:
 		dsInfo.Database = &kusciaapi.DatabaseDataSourceInfo{}
 		err = json.Unmarshal(connectionBytes, dsInfo.Database)
 	case common.DomainDataSourceTypeLocalFS:
@@ -649,7 +650,7 @@ func parseAndNormalizeDataSource(sourceType string, info *kusciaapi.DataSourceIn
 		// truncate slash
 		info.Localfs.Path = strings.TrimRight(info.Localfs.Path, string(filepath.Separator))
 		uri = info.Localfs.Path
-	case common.DomainDataSourceTypeMysql, common.DomainDataSourceTypePostgreSQL, common.DomainDataSourceTypeHive:
+	case common.DomainDataSourceTypeMysql, common.DomainDataSourceTypePostgreSQL, common.DomainDataSourceTypeHive, common.DomainDataSourceTypeDameng:
 		if isInvalid(info.Database == nil) {
 			return
 		}
@@ -660,7 +661,7 @@ func parseAndNormalizeDataSource(sourceType string, info *kusciaapi.DataSourceIn
 		}
 		uri = info.Odps.Endpoint + "/" + info.Odps.Project
 	default:
-		err = fmt.Errorf("datasource type:%q not support, only support [localfs,oss,mysql,odps,postgresql,hive]", sourceType)
+		err = fmt.Errorf("datasource type:%q not support, only support [localfs,oss,mysql,odps,postgresql,hive,dameng]", sourceType)
 		nlog.Error(err)
 		return
 	}
@@ -701,6 +702,7 @@ func validateDataSourceInfo(sourceType string, info *kusciaapi.DataSourceInfo) e
 	case common.DomainDataSourceTypeMysql:
 	case common.DomainDataSourceTypePostgreSQL:
 	case common.DomainDataSourceTypeHive:
+	case common.DomainDataSourceTypeDameng:
 		if info.Database == nil {
 			return fmt.Errorf("%v info is nil", sourceType)
 		}
diff --git a/pkg/kusciaapi/service/domaindata_source_test.go b/pkg/kusciaapi/service/domaindata_source_test.go
@@ -291,6 +291,44 @@ func TestListDomainDataSource_InfoErr(t *testing.T) {
 	assert.Equal(t, int32(pberrorcode.ErrorCode_KusciaAPIErrListDomainDataSource), res.Status.Code)
 }
 
+func TestValidateDataSourceType_UnsupportedType(t *testing.T) {
+	err := validateDataSourceType("unsupported_type")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "doesn't support")
+	assert.Contains(t, err.Error(), "unsupported_type")
+}
+
+func TestParseAndNormalizeDataSource_UnsupportedType(t *testing.T) {
+	info := &kusciaapi.DataSourceInfo{
+		Localfs: &kusciaapi.LocalDataSourceInfo{
+			Path: "/test/path",
+		},
+	}
+	uri, err := parseAndNormalizeDataSource("unsupported_type", info)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "not support")
+	assert.Contains(t, err.Error(), "unsupported_type")
+	assert.Empty(t, uri)
+}
+
+func TestCreateDomainDataSource_UnsupportedType(t *testing.T) {
+	conf := makeDomainDataSourceServiceConfig(t)
+	dsService := makeDomainDataSourceService(t, conf)
+	res := dsService.CreateDomainDataSource(context.Background(), &kusciaapi.CreateDomainDataSourceRequest{
+		Header:       nil,
+		DomainId:     mockDomainID,
+		DatasourceId: "ds-1",
+		Type:         "unsupported_type",
+		Info: &kusciaapi.DataSourceInfo{
+			Localfs: &kusciaapi.LocalDataSourceInfo{
+				Path: "./data",
+			},
+		},
+	})
+	assert.NotEqual(t, int32(0), res.Status.Code)
+	assert.Contains(t, res.Status.Message, "doesn't support")
+}
+
 func makeDomainDataSourceService(t *testing.T, conf *config.KusciaAPIConfig) IDomainDataSourceService {
 	return NewDomainDataSourceService(conf, makeConfigService(t))
 }
diff --git a/scripts/deploy/kuscia.sh b/scripts/deploy/kuscia.sh
@@ -648,7 +648,9 @@ function dataproxy_config() {
   dataProxyList:
     - endpoint: "dataproxy-grpc:8023"
       dataSourceTypes:
-        - "odps"'
+        - "odps"
+        - "hive"
+        - "dameng"'
   if ! grep -q "${data_proxy_config}" "${kuscia_config_file}"; then
      echo "${data_proxy_config}" >> "${kuscia_config_file}"
   fi

Original file line number	Diff line number	Diff line change
`@@ -229,6 +229,7 @@ const (`
`229`	`229`	`DomainDataSourceTypeODPS = "odps"`
`230`	`230`	`DomainDataSourceTypePostgreSQL = "postgresql"`
`231`	`231`	`DomainDataSourceTypeHive = "hive"`
	`232`	`+ DomainDataSourceTypeDameng = "dameng"`
`232`	`233`	`DefaultDomainDataSourceLocalFSPath = "var/storage/data"`
`233`	`234`	`)`
`234`	`235`