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 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
