You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@linkis.apache.org by ki...@apache.org on 2026/06/11 09:50:52 UTC
(linkis) 08/48: #299 Provide configuration service in module
This is an automated email from the ASF dual-hosted git repository.
aiceflower pushed a commit to branch release-0.9.4
in repository https://gitbox.apache.org/repos/asf/linkis.git
commit d30af34bdce29d14c96494c7d9d828cfbbba3ce7
Author: Davidhua1996 <da...@hotmail.com>
AuthorDate: Sun Feb 23 16:09:32 2020 +0800
#299 Provide configuration service in module
---
.../core/service/BmlAppService.java | 48 +++
.../core/service/DataSourceInfoService.java | 121 +++++++
.../core/service/DataSourceOpService.java | 17 +
.../core/service/DataSourceRelateService.java | 54 +++
.../core/service/MetadataOperateService.java | 36 ++
.../core/service/impl/BmlAppServiceImpl.java | 102 ++++++
.../service/impl/DataSourceInfoServiceImpl.java | 370 +++++++++++++++++++++
.../service/impl/DataSourceRelateServiceImpl.java | 58 ++++
.../service/impl/MetadataOperateServiceImpl.java | 110 ++++++
9 files changed, 916 insertions(+)
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/BmlAppService.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/BmlAppService.java
new file mode 100644
index 0000000000..97251129ad
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/BmlAppService.java
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service;
+
+import com.webank.wedatasphere.linkis.common.exception.ErrorException;
+
+import java.io.InputStream;
+
+/**
+ * BML application service
+ * @author davidhua
+ * 2020/02/14
+ */
+public interface BmlAppService {
+
+ /**
+ * Upload resource by bml client
+ * @param fileName file name
+ * @param inputStream inputStream
+ * @return resource id
+ */
+ String clientUploadResource(String userName, String fileName, InputStream inputStream) throws ErrorException;
+
+ /**
+ * Remove resource by bml client
+ * @param resourceId resource id
+ */
+ void clientRemoveResource(String userName, String resourceId) throws ErrorException;
+
+ /**
+ * Update resource by bml client
+ * @param resourceId resource id
+ * @param inputStream input stream
+ * @return version
+ */
+ String clientUpdateResource(String userName, String resourceId, InputStream inputStream) throws ErrorException;
+}
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceInfoService.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceInfoService.java
new file mode 100644
index 0000000000..8592d58c55
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceInfoService.java
@@ -0,0 +1,121 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service;
+
+import com.webank.wedatasphere.linkis.common.exception.ErrorException;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSource;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSourceEnv;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.vo.DataSourceEnvVo;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.vo.DataSourceVo;
+
+import java.util.List;
+
+
+/**
+ * @author davidhua
+ * 2020/02/13
+ */
+public interface DataSourceInfoService {
+
+ /**
+ * Save data source information
+ * @param dataSource data source
+ */
+ void saveDataSourceInfo(DataSource dataSource) throws ErrorException;
+
+ /**
+ * Add parameters of data source environment
+ * @param dataSourceEnvId data source environment
+ * @param dataSource data source
+ */
+ void addEnvParamsToDataSource(Long dataSourceEnvId, DataSource dataSource);
+
+ /**
+ * Get data source
+ * @param dataSourceId id
+ * @param createSystem system name
+ * @return data source entity
+ */
+ DataSource getDataSourceInfo(Long dataSourceId, String createSystem);
+
+ /**
+ * Get data source brief information
+ * @param dataSourceId data source id
+ * @param createSystem system
+ * @return
+ */
+ DataSource getDataSourceInfoBrief(Long dataSourceId, String createSystem);
+ /**
+ * Remove data source
+ * @param dataSourceId id
+ * @param createSystem system name
+ * @return
+ */
+ Long removeDataSourceInfo(Long dataSourceId, String createSystem);
+
+ /**
+ * Update data source
+ * @param updatedOne updated data source
+ * @param storedOne stored data source
+ */
+ void updateDataSourceInfo(DataSource updatedOne, DataSource storedOne) throws ErrorException;
+
+ /**
+ * Page query of data source
+ * @param dataSourceVo data source view entity
+ * @return
+ */
+ List<DataSource> queryDataSourceInfoPage(DataSourceVo dataSourceVo);
+
+ /**
+ * Save data source environment
+ * @param dataSourceEnv data source environment
+ */
+ void saveDataSourceEnv(DataSourceEnv dataSourceEnv) throws ErrorException;
+
+ /**
+ * List data source environments
+ * @param dataSourceTypeId type id
+ * @return
+ */
+ List<DataSourceEnv> listDataSourceEnvByType(Long dataSourceTypeId);
+
+ /**
+ * Get data source environment
+ * @param envId environment id
+ * @return
+ */
+ DataSourceEnv getDataSourceEnv(Long envId);
+
+ /**
+ * Remove data source environment
+ * @param envId environment id
+ * @return
+ */
+ Long removeDataSourceEnv(Long envId);
+
+ /**
+ * Update data source environment
+ * @param updatedOne
+ * @param storedOne
+ */
+ void updateDataSourceEnv(DataSourceEnv updatedOne, DataSourceEnv storedOne) throws ErrorException;
+
+ /**
+ * Page query of data source environment
+ * @param dataSourceEnvVo
+ * @return
+ */
+ List<DataSourceEnv> queryDataSourceEnvPage(DataSourceEnvVo dataSourceEnvVo);
+}
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceOpService.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceOpService.java
new file mode 100644
index 0000000000..16b0d863f4
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceOpService.java
@@ -0,0 +1,17 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service;
+
+public interface DataSourceOpService {
+}
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceRelateService.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceRelateService.java
new file mode 100644
index 0000000000..165773d9c1
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/DataSourceRelateService.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service;
+
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSourceParamKeyDefinition;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSourceType;
+
+import java.util.List;
+
+/**
+ * @author davidhua
+ * 2020/02/14
+ */
+public interface DataSourceRelateService {
+ /**
+ * Get key definitions by data source type and scope
+ * @param dataSourceTypeId data source type id
+ * @param scope scope
+ * @return
+ */
+ List<DataSourceParamKeyDefinition> getKeyDefinitionsByType(Long dataSourceTypeId,
+ DataSourceParamKeyDefinition.Scope scope);
+
+ /**
+ * Get key definitions by data source type and scope
+ * @param dataSourceTypeId data source type id
+ * @return
+ */
+ List<DataSourceParamKeyDefinition> getKeyDefinitionsByType(Long dataSourceTypeId);
+
+ /**
+ * Get all data source types
+ * @return
+ */
+ List<DataSourceType> getAllDataSourceTypes();
+
+ /**
+ * Get data source type
+ * @param typeId
+ * @return
+ */
+ DataSourceType getDataSourceType(Long typeId);
+}
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/MetadataOperateService.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/MetadataOperateService.java
new file mode 100644
index 0000000000..ca46888025
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/MetadataOperateService.java
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service;
+
+import com.webank.wedatasphere.linkis.common.exception.ErrorException;
+import com.webank.wedatasphere.linkis.common.exception.WarnException;
+
+import java.util.Map;
+
+/**
+ * Metadata service
+ * @author davidhua
+ * 2020/02/14
+ */
+public interface MetadataOperateService {
+
+ /**
+ * Build connection with parameters in request
+ * @param mdRemoteServiceName metadata remote service
+ * @param operator operate user
+ * @param connectParams parameters
+ * @throws ErrorException
+ */
+ void doRemoteConnect(String mdRemoteServiceName, String operator, Map<String, Object> connectParams) throws WarnException;
+}
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/BmlAppServiceImpl.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/BmlAppServiceImpl.java
new file mode 100644
index 0000000000..ad3c26e96e
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/BmlAppServiceImpl.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service.impl;
+
+import com.webank.wedatasphere.linkis.bml.client.BmlClient;
+import com.webank.wedatasphere.linkis.bml.client.BmlClientFactory;
+import com.webank.wedatasphere.linkis.bml.protocol.BmlDeleteResponse;
+import com.webank.wedatasphere.linkis.bml.protocol.BmlUpdateResponse;
+import com.webank.wedatasphere.linkis.bml.protocol.BmlUploadResponse;
+import com.webank.wedatasphere.linkis.common.exception.ErrorException;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.ServiceErrorCode;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.service.BmlAppService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.cloud.context.config.annotation.RefreshScope;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.io.InputStream;
+
+/**
+ * Wrap the communication between Bml service
+ * @author davidhua
+ * 2020/02/15
+ */
+@Service
+@RefreshScope
+public class BmlAppServiceImpl implements BmlAppService {
+ private static final Logger LOG = LoggerFactory.getLogger(BmlAppService.class);
+ /**
+ * Bml client
+ */
+ private BmlClient client;
+
+ @PostConstruct
+ public void buildClient(){
+ client = BmlClientFactory.createBmlClient();
+ }
+ @Override
+ public String clientUploadResource(String userName, String fileName,
+ InputStream inputStream) throws ErrorException{
+ LOG.info("Upload resource to bml server: [ proxy_to_user: " + userName +
+ ", file name:" + fileName + " ]");
+ try{
+ BmlUploadResponse response = client.uploadResource(userName, fileName, inputStream);
+ if(!response.isSuccess()){
+ throw new ErrorException(ServiceErrorCode.BML_SERVICE_ERROR.getValue(), "");
+ }
+ return response.resourceId();
+ }catch(Exception e){
+ LOG.error("Failed to upload resource to bml server[上传资源文件失败], [ proxy_to_user: " + userName +
+ ", file name:" + fileName + " ]", e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void clientRemoveResource(String userName, String resourceId) throws ErrorException{
+ LOG.info("Remove resource to bml server: [ proxy_to_user: " + userName +
+ ", resource id:" + resourceId + " ]");
+ try{
+ BmlDeleteResponse response = client.deleteResource(userName, resourceId);
+ if(!response.isSuccess()){
+ throw new ErrorException(ServiceErrorCode.BML_SERVICE_ERROR.getValue(), "");
+ }
+ }catch(Exception e){
+ LOG.error("Fail to remove resource to bml server[删除资源文件失败], [ proxy_to_user: " + userName +
+ ", resource id:" + resourceId + " ]");
+ throw e;
+ }
+ }
+
+ @Override
+ public String clientUpdateResource(String userName, String resourceId,
+ InputStream inputStream) throws ErrorException{
+ LOG.info("Update resource to bml server: [ proxy_to_user: " + userName +
+ ", resource id:" + resourceId + " ]");
+ try{
+ //File name is invalid;
+ BmlUpdateResponse response = client.updateResource(userName, resourceId, "filename", inputStream);
+ if(!response.isSuccess()){
+ throw new ErrorException(ServiceErrorCode.BML_SERVICE_ERROR.getValue(), "");
+ }
+ return response.version();
+ }catch(Exception e){
+ LOG.error("Fail to update resource to bml server[更新资源文件失败], [ proxy_to_user: " + userName +
+ ", resource id:" + resourceId + " ]");
+ throw e;
+ }
+ }
+}
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/DataSourceInfoServiceImpl.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/DataSourceInfoServiceImpl.java
new file mode 100644
index 0000000000..c948ed9952
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/DataSourceInfoServiceImpl.java
@@ -0,0 +1,370 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service.impl;
+
+import com.github.pagehelper.PageHelper;
+import com.github.pagehelper.PageInfo;
+import com.webank.wedatasphere.linkis.common.exception.ErrorException;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.dao.DataSourceParamKeyDao;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.dao.DataSourceTypeEnvDao;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.formdata.FormStreamContent;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.util.json.Json;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.dao.DataSourceDao;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.dao.DataSourceEnvDao;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSource;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSourceEnv;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSourceParamKeyDefinition;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.service.BmlAppService;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.service.DataSourceInfoService;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.vo.DataSourceEnvVo;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.vo.DataSourceVo;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+
+/**
+ * @author davidhua
+ * 2020/02/13
+ */
+@Service
+public class DataSourceInfoServiceImpl implements DataSourceInfoService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DataSourceInfoService.class);
+ @Autowired
+ private BmlAppService bmlAppService;
+
+ @Autowired
+ private DataSourceTypeEnvDao dataSourceTypeEnvDao;
+
+ @Autowired
+ private DataSourceDao dataSourceDao;
+
+ @Autowired
+ private DataSourceEnvDao dataSourceEnvDao;
+
+ @Autowired
+ private DataSourceParamKeyDao dataSourceParamKeyDao;
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void saveDataSourceInfo(DataSource dataSource) throws ErrorException {
+ storeConnectParams(dataSource.getCreateUser(), dataSource.getKeyDefinitions(),
+ dataSource.getConnectParams(), parameter ->{
+ dataSource.setParameter(parameter);
+ //Save information into database
+ dataSourceDao.insertOne(dataSource);
+ });
+ }
+
+ @Override
+ public void addEnvParamsToDataSource(Long dataSourceEnvId, DataSource dataSource) {
+ DataSourceEnv dataSourceEnv = dataSourceEnvDao.selectOneDetail(dataSourceEnvId);
+ if(null != dataSourceEnv){
+ Map<String, Object> envParamMap = dataSourceEnv.getConnectParams();
+ envParamMap.putAll(dataSource.getConnectParams());
+ dataSource.setConnectParams(envParamMap);
+ }
+ }
+
+ @Override
+ public DataSource getDataSourceInfo(Long dataSourceId, String createSystem) {
+ return dataSourceDao.selectOneDetail(dataSourceId, createSystem);
+ }
+
+ @Override
+ public DataSource getDataSourceInfoBrief(Long dataSourceId, String createSystem) {
+ return dataSourceDao.selectOne(dataSourceId, createSystem);
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public Long removeDataSourceInfo(Long dataSourceId, String createSystem) {
+ DataSource dataSource = dataSourceDao.selectOne(dataSourceId, createSystem);
+ if(null != dataSource){
+ //First to delete record in db
+ int affect = dataSourceDao.removeOne(dataSourceId, createSystem);
+ if(affect > 0){
+ //Remove resource
+ Map<String, Object> connectParams = dataSource.getConnectParams();
+ List<DataSourceParamKeyDefinition> keyDefinitions = dataSourceParamKeyDao
+ .listByDataSourceType(dataSource.getDataSourceTypeId());
+ keyDefinitions.forEach(keyDefinition -> {
+ if(keyDefinition.getValueType() == DataSourceParamKeyDefinition.ValueType.FILE
+ && keyDefinition.getScope() != DataSourceParamKeyDefinition.Scope.ENV
+ && connectParams.containsKey(keyDefinition.getKey())){
+ try {
+ //Proxy creator to delete resource
+ bmlAppService.clientRemoveResource(dataSource.getCreateUser(), String
+ .valueOf(connectParams.get(keyDefinition.getKey())));
+ }catch(Exception e){
+ //Ignore remove error
+ }
+ }
+ });
+ return dataSourceId;
+ }
+ }
+ return -1L;
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void updateDataSourceInfo(DataSource updatedOne, DataSource storedOne) throws ErrorException{
+ updateConnectParams(updatedOne.getCreateUser(), updatedOne.getKeyDefinitions(),
+ updatedOne.getConnectParams(), storedOne.getConnectParams(),
+ parameter -> {
+ updatedOne.setParameter(parameter);
+ //Save information into database
+ dataSourceDao.updateOne(updatedOne);
+ });
+ }
+
+ @Override
+ public List<DataSource> queryDataSourceInfoPage(DataSourceVo dataSourceVo) {
+ PageHelper.startPage(dataSourceVo.getCurrentPage(), dataSourceVo.getPageSize());
+ List<DataSource> queryList = dataSourceDao.selectByPageVo(dataSourceVo);
+ PageInfo<DataSource> pageInfo = new PageInfo<>(queryList);
+ return pageInfo.getList();
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void saveDataSourceEnv(DataSourceEnv dataSourceEnv) throws ErrorException {
+ storeConnectParams(dataSourceEnv.getCreateUser(), dataSourceEnv.getKeyDefinitions(),
+ dataSourceEnv.getConnectParams(),
+ parameter ->{
+ dataSourceEnv.setParameter(parameter);
+ //Save environment into database
+ dataSourceEnvDao.insertOne(dataSourceEnv);
+ //Store relation
+ dataSourceTypeEnvDao.insertRelation(dataSourceEnv.getDataSourceTypeId(), dataSourceEnv.getId());
+ });
+ }
+
+ @Override
+ public List<DataSourceEnv> listDataSourceEnvByType(Long dataSourceTypeId) {
+ return dataSourceEnvDao.listByTypeId(dataSourceTypeId);
+ }
+
+ @Override
+ public DataSourceEnv getDataSourceEnv(Long envId) {
+ return dataSourceEnvDao.selectOneDetail(envId);
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public Long removeDataSourceEnv(Long envId) {
+ DataSourceEnv dataSourceEnv = dataSourceEnvDao.selectOneDetail(envId);
+ if(null != dataSourceEnv){
+ //First to delete record in db
+ int affect = dataSourceEnvDao.removeOne(envId);
+ if(affect > 0){
+ //Remove relations
+ dataSourceTypeEnvDao.removeRelationsByEnvId(envId);
+ //Remove resource
+ Map<String, Object> connectParams = dataSourceEnv.getConnectParams();
+ List<DataSourceParamKeyDefinition> keyDefinitions = dataSourceParamKeyDao
+ .listByDataSourceTypeAndScope(dataSourceEnv.getDataSourceTypeId(), DataSourceParamKeyDefinition.Scope.ENV);
+ keyDefinitions.forEach(keyDefinition -> {
+ if(keyDefinition.getValueType() == DataSourceParamKeyDefinition.ValueType.FILE
+ && connectParams.containsKey(keyDefinition.getKey())){
+ try{
+ //Proxy creator to delete resource
+ bmlAppService.clientRemoveResource(dataSourceEnv.getCreateUser(), String
+ .valueOf(connectParams.get(keyDefinition.getKey())));
+ }catch(Exception e){
+ //Ignore remove error
+ }
+ }
+ });
+ return envId;
+ }
+ }
+ return -1L;
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void updateDataSourceEnv(DataSourceEnv updatedOne, DataSourceEnv storedOne) throws ErrorException {
+ updateConnectParams(updatedOne.getCreateUser(), updatedOne.getKeyDefinitions(),
+ updatedOne.getConnectParams(), storedOne.getConnectParams(),
+ parameter ->{
+ updatedOne.setParameter(parameter);
+ //Update environment into database
+ dataSourceEnvDao.updateOne(updatedOne);
+ if(!updatedOne.getDataSourceTypeId().equals(storedOne.getDataSourceTypeId())){
+ //Remove old relation and add new relation
+ dataSourceTypeEnvDao.removeRelationsByEnvId(updatedOne.getId());
+ dataSourceTypeEnvDao.insertRelation(updatedOne.getDataSourceTypeId(), updatedOne.getId());
+ }
+ });
+ }
+
+ @Override
+ public List<DataSourceEnv> queryDataSourceEnvPage(DataSourceEnvVo dataSourceEnvVo) {
+ PageHelper.startPage(dataSourceEnvVo.getCurrentPage(), dataSourceEnvVo.getPageSize());
+ List<DataSourceEnv> queryList = dataSourceEnvDao.selectByPageVo(dataSourceEnvVo);
+ PageInfo<DataSourceEnv> pageInfo = new PageInfo<>(queryList);
+ return pageInfo.getList();
+ }
+
+ /**
+ *
+ * @param userName
+ * @param keyDefinitionList
+ * @param updatedParams
+ * @param storedParams
+ * @param parameterCallback
+ * @throws ErrorException
+ */
+ private void updateConnectParams(String userName, List<DataSourceParamKeyDefinition> keyDefinitionList,
+ Map<String, Object> updatedParams, Map<String, Object> storedParams,
+ Consumer<String> parameterCallback) throws ErrorException {
+ List<String> definedKeyNames = keyDefinitionList.stream().map(DataSourceParamKeyDefinition::getKey)
+ .collect(Collectors.toList());
+ List<String> uploadedResources = new ArrayList<>();
+ try {
+ updatedParams.entrySet().removeIf(entry -> {
+ if (!definedKeyNames.contains(entry.getKey())) {
+ return true;
+ }
+ Object paramValue = entry.getValue();
+ if (paramValue instanceof FormStreamContent) {
+ String resourceId = String.valueOf(storedParams.getOrDefault(entry.getKey(), ""));
+ if (StringUtils.isNotBlank(resourceId)) {
+ uploadFormStream(userName, (FormStreamContent) paramValue, resourceId);
+ } else {
+ resourceId = uploadFormStream(userName, (FormStreamContent) paramValue, "");
+ }
+ if (null == resourceId) {
+ return true;
+ }
+ uploadedResources.add(resourceId);
+ entry.setValue(resourceId);
+ }
+ storedParams.remove(entry.getKey());
+ return false;
+ });
+ //Found the duplicate File
+ List<String> duplicateResources = new ArrayList<>();
+ keyDefinitionList.forEach( definedKey-> {
+ if(definedKey.getValueType() == DataSourceParamKeyDefinition.ValueType.FILE
+ && storedParams.containsKey(definedKey.getKey())){
+ duplicateResources.add(String.valueOf(storedParams.get(definedKey.getKey())));
+ }
+ });
+ parameterCallback.accept(Json.toJson(updatedParams, null));
+ deleteResources(userName, duplicateResources);
+ }catch(Exception e){
+ deleteResources(userName, uploadedResources);
+ if(e.getCause() instanceof ErrorException){
+ throw (ErrorException)e.getCause();
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * Upload the form stream context in connect parameters,
+ * and serialize parameters
+ * @param keyDefinitionList
+ * @param connectParams
+ * @param parameterCallback
+ */
+ private void storeConnectParams(String userName, List<DataSourceParamKeyDefinition> keyDefinitionList,
+ Map<String, Object> connectParams,
+ Consumer<String> parameterCallback) throws ErrorException{
+ List<String> definedKeyNames = keyDefinitionList.stream().map(DataSourceParamKeyDefinition::getKey)
+ .collect(Collectors.toList());
+ List<String> uploadedResources = new ArrayList<>();
+ try{
+ connectParams.entrySet().removeIf(entry -> {
+ if(!definedKeyNames.contains(entry.getKey())){
+ return true;
+ }
+ Object paramValue = entry.getValue();
+ //Upload stream resource in connection params
+ if (paramValue instanceof FormStreamContent) {
+ String resourceId = uploadFormStream(userName, (FormStreamContent)paramValue, "");
+ if(null == resourceId) {
+ return true;
+ }
+ uploadedResources.add(resourceId);
+ entry.setValue(resourceId);
+ }
+ return false;
+ });
+ parameterCallback.accept(Json.toJson(connectParams, null));
+ }catch(Exception e){
+ deleteResources(userName, uploadedResources);
+ if(e.getCause() instanceof ErrorException){
+ throw (ErrorException)e.getCause();
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * Upload form stream
+ * @param userName user name
+ * @param streamContent stream content
+ * @param resourceId resource id
+ * @return resource id or version tab
+ */
+ private String uploadFormStream(String userName, FormStreamContent streamContent, String resourceId){
+ String fileName = streamContent.getFileName();
+ InputStream inputStream = streamContent.getStream();
+ if (null != inputStream){
+ //Proxy creator to upload resource
+ try{
+ return StringUtils.isBlank(resourceId)? bmlAppService.clientUploadResource(userName, fileName, inputStream)
+ : bmlAppService.clientUpdateResource(userName, resourceId, inputStream);
+ }catch(Exception e){
+ //Wrap with runtime exception
+ throw new RuntimeException(e);
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Delete uploaded resources
+ * @param userName user name
+ * @param uploadedResources resource id list
+ */
+ private void deleteResources(String userName, List<String> uploadedResources){
+ if(!uploadedResources.isEmpty()){
+ //Remove duplicated resource
+ for (String resourceId : uploadedResources) {
+ try{
+ //Proxy to delete resource
+ bmlAppService.clientRemoveResource(userName, resourceId);
+ }catch(Exception ie){
+ //ignore
+ }
+ }
+ }
+ }
+}
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/DataSourceRelateServiceImpl.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/DataSourceRelateServiceImpl.java
new file mode 100644
index 0000000000..16c7ad3b6c
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/DataSourceRelateServiceImpl.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service.impl;
+
+import com.webank.wedatasphere.linkis.datasourcemanager.core.dao.DataSourceTypeDao;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.dao.DataSourceParamKeyDao;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSourceParamKeyDefinition;
+import com.webank.wedatasphere.linkis.datasourcemanager.common.domain.DataSourceType;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.service.DataSourceRelateService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+/**
+ * @author davidhua
+ * 2020/02/14
+ */
+@Service
+public class DataSourceRelateServiceImpl implements DataSourceRelateService {
+
+ @Autowired
+ private DataSourceParamKeyDao paramKeyDao;
+
+ @Autowired
+ private DataSourceTypeDao dataSourceTypeDao;
+
+ @Override
+ public List<DataSourceParamKeyDefinition> getKeyDefinitionsByType(Long dataSourceTypeId, DataSourceParamKeyDefinition.Scope scope) {
+ return paramKeyDao.listByDataSourceTypeAndScope(dataSourceTypeId, scope);
+ }
+
+ @Override
+ public List<DataSourceParamKeyDefinition> getKeyDefinitionsByType(Long dataSourceTypeId) {
+ return paramKeyDao.listByDataSourceType(dataSourceTypeId);
+ }
+
+ @Override
+ public List<DataSourceType> getAllDataSourceTypes() {
+ return dataSourceTypeDao.getAllTypes();
+ }
+
+ @Override
+ public DataSourceType getDataSourceType(Long typeId) {
+ return dataSourceTypeDao.selectOne(typeId);
+ }
+}
diff --git a/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/MetadataOperateServiceImpl.java b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/MetadataOperateServiceImpl.java
new file mode 100644
index 0000000000..910e26a943
--- /dev/null
+++ b/datasource/datasource/datasourcemanager/server/src/main/java/com/webank/wedatasphere/linkis/datasourcemanager/core/service/impl/MetadataOperateServiceImpl.java
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2019 WeBank
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.webank.wedatasphere.linkis.datasourcemanager.core.service.impl;
+
+import com.webank.wedatasphere.linkis.common.exception.ErrorException;
+import com.webank.wedatasphere.linkis.common.exception.WarnException;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.formdata.FormStreamContent;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.service.BmlAppService;
+import com.webank.wedatasphere.linkis.datasourcemanager.core.service.MetadataOperateService;
+import com.webank.wedatasphere.linkis.metadatamanager.common.protocol.MetadataConnect;
+import com.webank.wedatasphere.linkis.metadatamanager.common.protocol.MetadataResponse;
+import com.webank.wedatasphere.linkis.rpc.Sender;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static com.webank.wedatasphere.linkis.datasourcemanager.common.ServiceErrorCode.BML_SERVICE_ERROR;
+import static com.webank.wedatasphere.linkis.datasourcemanager.common.ServiceErrorCode.REMOTE_METADATA_SERVICE_ERROR;
+
+/**
+ * @author davidhua
+ * 2020/02/14
+ */
+@Service
+public class MetadataOperateServiceImpl implements MetadataOperateService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MetadataOperateService.class);
+ @Autowired
+ private BmlAppService bmlAppService;
+ @Override
+ public void doRemoteConnect(String mdRemoteServiceName ,
+ String operator, Map<String, Object> connectParams) throws WarnException {
+ List<String> uploadedResources = new ArrayList<>();
+ try{
+ connectParams.entrySet().removeIf(entry -> {
+ Object paramValue = entry.getValue();
+ //Upload stream resource in connection parameters
+ if(paramValue instanceof FormStreamContent){
+ FormStreamContent streamContent = (FormStreamContent)paramValue;
+ String fileName = streamContent.getFileName();
+ InputStream inputStream = streamContent.getStream();
+ if (null != inputStream){
+ try {
+ String resourceId = bmlAppService.clientUploadResource(operator,
+ fileName, inputStream);
+ if(null == resourceId){
+ return true;
+ }
+ uploadedResources.add(resourceId);
+ entry.setValue(resourceId);
+ } catch (ErrorException e) {
+ throw new WarnException(BML_SERVICE_ERROR.getValue(), "Fail to operate file in request[上传文件处理失败]");
+ }
+ }
+ }
+ return false;
+ });
+ LOG.info("Send request to metadata service:[" + mdRemoteServiceName + "] for building a connection");
+ //Get a sender
+ Sender sender = Sender.getSender(mdRemoteServiceName);
+ try {
+ Object object = sender.ask(new MetadataConnect(operator, connectParams, ""));
+ if (object instanceof MetadataResponse) {
+ MetadataResponse response = (MetadataResponse) object;
+ if (!response.status()) {
+ throw new WarnException(REMOTE_METADATA_SERVICE_ERROR.getValue(),
+ "Connection Failed[连接失败], Msg[" + response.data() + "]");
+ }
+ } else {
+ throw new WarnException(REMOTE_METADATA_SERVICE_ERROR.getValue(),
+ "Remote Service Error[远端服务出错, 联系运维处理]");
+ }
+ }catch(Throwable t){
+ if(!(t instanceof WarnException)) {
+ throw new WarnException(REMOTE_METADATA_SERVICE_ERROR.getValue(),
+ "Remote Service Error[远端服务出错, 联系运维处理]");
+ }
+ throw t;
+ }
+ }finally{
+ if(!uploadedResources.isEmpty()){
+ uploadedResources.forEach( resourceId ->{
+ try{
+ //Proxy to delete resource
+ bmlAppService.clientRemoveResource(operator, resourceId);
+ }catch(Exception e){
+ //ignore
+ }
+ });
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org