dailidong
4 years ago
286 changed files with 14611 additions and 6904 deletions
@ -1,54 +0,0 @@ |
|||||||
/* |
|
||||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
|
||||||
* contributor license agreements. See the NOTICE file distributed with |
|
||||||
* this work for additional information regarding copyright ownership. |
|
||||||
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service; |
|
||||||
|
|
||||||
import org.apache.dolphinscheduler.common.graph.DAG; |
|
||||||
import org.apache.dolphinscheduler.common.model.TaskNode; |
|
||||||
import org.apache.dolphinscheduler.common.model.TaskNodeRelation; |
|
||||||
import org.apache.dolphinscheduler.common.process.ProcessDag; |
|
||||||
import org.apache.dolphinscheduler.common.utils.*; |
|
||||||
import org.apache.dolphinscheduler.dao.entity.ProcessData; |
|
||||||
import org.apache.dolphinscheduler.dao.entity.ProcessInstance; |
|
||||||
import org.apache.dolphinscheduler.dao.utils.DagHelper; |
|
||||||
|
|
||||||
import java.util.List; |
|
||||||
|
|
||||||
/** |
|
||||||
* base DAG service |
|
||||||
*/ |
|
||||||
public class BaseDAGService extends BaseService{ |
|
||||||
|
|
||||||
|
|
||||||
/** |
|
||||||
* process instance to DAG |
|
||||||
* |
|
||||||
* @param processInstance input process instance |
|
||||||
* @return process instance dag. |
|
||||||
*/ |
|
||||||
public static DAG<String, TaskNode, TaskNodeRelation> processInstance2DAG(ProcessInstance processInstance) { |
|
||||||
|
|
||||||
String processDefinitionJson = processInstance.getProcessInstanceJson(); |
|
||||||
|
|
||||||
ProcessData processData = JSONUtils.parseObject(processDefinitionJson, ProcessData.class); |
|
||||||
|
|
||||||
List<TaskNode> taskNodeList = processData.getTasks(); |
|
||||||
|
|
||||||
ProcessDag processDag = DagHelper.getProcessDag(taskNodeList); |
|
||||||
|
|
||||||
return DagHelper.buildDagGraph(processDag); |
|
||||||
} |
|
||||||
} |
|
||||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,70 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessDefinitionVersion; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.User; |
||||||
|
|
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
public interface ProcessDefinitionVersionService { |
||||||
|
|
||||||
|
/** |
||||||
|
* add the newest version of one process definition |
||||||
|
* |
||||||
|
* @param processDefinition the process definition that need to record version |
||||||
|
* @return the newest version number of this process definition |
||||||
|
*/ |
||||||
|
long addProcessDefinitionVersion(ProcessDefinition processDefinition); |
||||||
|
|
||||||
|
/** |
||||||
|
* query the pagination versions info by one certain process definition id |
||||||
|
* |
||||||
|
* @param loginUser login user info to check auth |
||||||
|
* @param projectName process definition project name |
||||||
|
* @param pageNo page number |
||||||
|
* @param pageSize page size |
||||||
|
* @param processDefinitionId process definition id |
||||||
|
* @return the pagination process definition versions info of the certain process definition |
||||||
|
*/ |
||||||
|
Map<String, Object> queryProcessDefinitionVersions(User loginUser, String projectName, |
||||||
|
int pageNo, int pageSize, int processDefinitionId); |
||||||
|
|
||||||
|
/** |
||||||
|
* query one certain process definition version by version number and process definition id |
||||||
|
* |
||||||
|
* @param processDefinitionId process definition id |
||||||
|
* @param version version number |
||||||
|
* @return the process definition version info |
||||||
|
*/ |
||||||
|
ProcessDefinitionVersion queryByProcessDefinitionIdAndVersion(int processDefinitionId, |
||||||
|
long version); |
||||||
|
|
||||||
|
/** |
||||||
|
* delete one certain process definition by version number and process definition id |
||||||
|
* |
||||||
|
* @param loginUser login user info to check auth |
||||||
|
* @param projectName process definition project name |
||||||
|
* @param processDefinitionId process definition id |
||||||
|
* @param version version number |
||||||
|
* @return delele result code |
||||||
|
*/ |
||||||
|
Map<String, Object> deleteByProcessDefinitionIdAndVersion(User loginUser, String projectName, |
||||||
|
int processDefinitionId, long version); |
||||||
|
} |
||||||
@ -0,0 +1,186 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service.impl; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.api.enums.Status; |
||||||
|
import org.apache.dolphinscheduler.api.service.AccessTokenService; |
||||||
|
import org.apache.dolphinscheduler.api.service.BaseService; |
||||||
|
import org.apache.dolphinscheduler.api.utils.PageInfo; |
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.common.enums.UserType; |
||||||
|
import org.apache.dolphinscheduler.common.utils.DateUtils; |
||||||
|
import org.apache.dolphinscheduler.common.utils.EncryptionUtils; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.AccessToken; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.User; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.AccessTokenMapper; |
||||||
|
|
||||||
|
import java.util.Date; |
||||||
|
import java.util.HashMap; |
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
import org.slf4j.Logger; |
||||||
|
import org.slf4j.LoggerFactory; |
||||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||||
|
import org.springframework.stereotype.Service; |
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.metadata.IPage; |
||||||
|
import com.baomidou.mybatisplus.extension.plugins.pagination.Page; |
||||||
|
|
||||||
|
/** |
||||||
|
* access token service impl |
||||||
|
*/ |
||||||
|
@Service |
||||||
|
public class AccessTokenServiceImpl extends BaseService implements AccessTokenService { |
||||||
|
|
||||||
|
private static final Logger logger = LoggerFactory.getLogger(AccessTokenServiceImpl.class); |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private AccessTokenMapper accessTokenMapper; |
||||||
|
|
||||||
|
/** |
||||||
|
* query access token list |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param searchVal search value |
||||||
|
* @param pageNo page number |
||||||
|
* @param pageSize page size |
||||||
|
* @return token list for page number and page size |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryAccessTokenList(User loginUser, String searchVal, Integer pageNo, Integer pageSize) { |
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
|
||||||
|
PageInfo<AccessToken> pageInfo = new PageInfo<>(pageNo, pageSize); |
||||||
|
Page<AccessToken> page = new Page<>(pageNo, pageSize); |
||||||
|
int userId = loginUser.getId(); |
||||||
|
if (loginUser.getUserType() == UserType.ADMIN_USER) { |
||||||
|
userId = 0; |
||||||
|
} |
||||||
|
IPage<AccessToken> accessTokenList = accessTokenMapper.selectAccessTokenPage(page, searchVal, userId); |
||||||
|
pageInfo.setTotalCount((int) accessTokenList.getTotal()); |
||||||
|
pageInfo.setLists(accessTokenList.getRecords()); |
||||||
|
result.put(Constants.DATA_LIST, pageInfo); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
|
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* create token |
||||||
|
* |
||||||
|
* @param userId token for user |
||||||
|
* @param expireTime token expire time |
||||||
|
* @param token token string |
||||||
|
* @return create result code |
||||||
|
*/ |
||||||
|
public Map<String, Object> createToken(int userId, String expireTime, String token) { |
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
|
||||||
|
if (userId <= 0) { |
||||||
|
throw new IllegalArgumentException("User id should not less than or equals to 0."); |
||||||
|
} |
||||||
|
AccessToken accessToken = new AccessToken(); |
||||||
|
accessToken.setUserId(userId); |
||||||
|
accessToken.setExpireTime(DateUtils.stringToDate(expireTime)); |
||||||
|
accessToken.setToken(token); |
||||||
|
accessToken.setCreateTime(new Date()); |
||||||
|
accessToken.setUpdateTime(new Date()); |
||||||
|
|
||||||
|
// insert
|
||||||
|
int insert = accessTokenMapper.insert(accessToken); |
||||||
|
|
||||||
|
if (insert > 0) { |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
} else { |
||||||
|
putMsg(result, Status.CREATE_ACCESS_TOKEN_ERROR); |
||||||
|
} |
||||||
|
|
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* generate token |
||||||
|
* |
||||||
|
* @param userId token for user |
||||||
|
* @param expireTime token expire time |
||||||
|
* @return token string |
||||||
|
*/ |
||||||
|
public Map<String, Object> generateToken(int userId, String expireTime) { |
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
String token = EncryptionUtils.getMd5(userId + expireTime + System.currentTimeMillis()); |
||||||
|
result.put(Constants.DATA_LIST, token); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* delete access token |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param id token id |
||||||
|
* @return delete result code |
||||||
|
*/ |
||||||
|
public Map<String, Object> delAccessTokenById(User loginUser, int id) { |
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
|
||||||
|
AccessToken accessToken = accessTokenMapper.selectById(id); |
||||||
|
|
||||||
|
if (accessToken == null) { |
||||||
|
logger.error("access token not exist, access token id {}", id); |
||||||
|
putMsg(result, Status.ACCESS_TOKEN_NOT_EXIST); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
if (loginUser.getId() != accessToken.getUserId() && |
||||||
|
loginUser.getUserType() != UserType.ADMIN_USER) { |
||||||
|
putMsg(result, Status.USER_NO_OPERATION_PERM); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
accessTokenMapper.deleteById(id); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* update token by id |
||||||
|
* |
||||||
|
* @param id token id |
||||||
|
* @param userId token for user |
||||||
|
* @param expireTime token expire time |
||||||
|
* @param token token string |
||||||
|
* @return update result code |
||||||
|
*/ |
||||||
|
public Map<String, Object> updateToken(int id, int userId, String expireTime, String token) { |
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
|
||||||
|
AccessToken accessToken = accessTokenMapper.selectById(id); |
||||||
|
if (accessToken == null) { |
||||||
|
logger.error("access token not exist, access token id {}", id); |
||||||
|
putMsg(result, Status.ACCESS_TOKEN_NOT_EXIST); |
||||||
|
return result; |
||||||
|
} |
||||||
|
accessToken.setUserId(userId); |
||||||
|
accessToken.setExpireTime(DateUtils.stringToDate(expireTime)); |
||||||
|
accessToken.setToken(token); |
||||||
|
accessToken.setUpdateTime(new Date()); |
||||||
|
|
||||||
|
accessTokenMapper.updateById(accessToken); |
||||||
|
|
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,384 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service.impl; |
||||||
|
|
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.api.dto.CommandStateCount; |
||||||
|
import org.apache.dolphinscheduler.api.dto.DefineUserDto; |
||||||
|
import org.apache.dolphinscheduler.api.dto.TaskCountDto; |
||||||
|
import org.apache.dolphinscheduler.api.enums.Status; |
||||||
|
import org.apache.dolphinscheduler.api.service.BaseService; |
||||||
|
import org.apache.dolphinscheduler.api.service.DataAnalysisService; |
||||||
|
import org.apache.dolphinscheduler.api.service.ProjectService; |
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.common.enums.CommandType; |
||||||
|
import org.apache.dolphinscheduler.common.enums.UserType; |
||||||
|
import org.apache.dolphinscheduler.common.utils.DateUtils; |
||||||
|
import org.apache.dolphinscheduler.common.utils.StringUtils; |
||||||
|
import org.apache.dolphinscheduler.common.utils.TriFunction; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.CommandCount; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.DefinitionGroupByUser; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ExecuteStatusCount; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.Project; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.User; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.CommandMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ErrorCommandMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper; |
||||||
|
import org.apache.dolphinscheduler.service.process.ProcessService; |
||||||
|
|
||||||
|
import java.text.MessageFormat; |
||||||
|
import java.util.ArrayList; |
||||||
|
import java.util.Date; |
||||||
|
import java.util.EnumMap; |
||||||
|
import java.util.HashMap; |
||||||
|
import java.util.List; |
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
import org.slf4j.Logger; |
||||||
|
import org.slf4j.LoggerFactory; |
||||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||||
|
import org.springframework.stereotype.Service; |
||||||
|
|
||||||
|
/** |
||||||
|
* data analysis service impl |
||||||
|
*/ |
||||||
|
@Service |
||||||
|
public class DataAnalysisServiceImpl extends BaseService implements DataAnalysisService { |
||||||
|
|
||||||
|
private static final Logger logger = LoggerFactory.getLogger(DataAnalysisServiceImpl.class); |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProjectMapper projectMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProjectService projectService; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProcessInstanceMapper processInstanceMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProcessDefinitionMapper processDefinitionMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private CommandMapper commandMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ErrorCommandMapper errorCommandMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private TaskInstanceMapper taskInstanceMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProcessService processService; |
||||||
|
|
||||||
|
private static final String COMMAND_STATE = "commandState"; |
||||||
|
|
||||||
|
private static final String ERROR_COMMAND_STATE = "errorCommandState"; |
||||||
|
|
||||||
|
/** |
||||||
|
* statistical task instance status data |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param projectId project id |
||||||
|
* @param startDate start date |
||||||
|
* @param endDate end date |
||||||
|
* @return task state count data |
||||||
|
*/ |
||||||
|
public Map<String, Object> countTaskStateByProject(User loginUser, int projectId, String startDate, String endDate) { |
||||||
|
|
||||||
|
return countStateByProject( |
||||||
|
loginUser, |
||||||
|
projectId, |
||||||
|
startDate, |
||||||
|
endDate, |
||||||
|
(start, end, projectIds) -> this.taskInstanceMapper.countTaskInstanceStateByUser(start, end, projectIds)); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* statistical process instance status data |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param projectId project id |
||||||
|
* @param startDate start date |
||||||
|
* @param endDate end date |
||||||
|
* @return process instance state count data |
||||||
|
*/ |
||||||
|
public Map<String, Object> countProcessInstanceStateByProject(User loginUser, int projectId, String startDate, String endDate) { |
||||||
|
return this.countStateByProject( |
||||||
|
loginUser, |
||||||
|
projectId, |
||||||
|
startDate, |
||||||
|
endDate, |
||||||
|
(start, end, projectIds) -> this.processInstanceMapper.countInstanceStateByUser(start, end, projectIds)); |
||||||
|
} |
||||||
|
|
||||||
|
private Map<String, Object> countStateByProject(User loginUser, int projectId, String startDate, String endDate |
||||||
|
, TriFunction<Date, Date, Integer[], List<ExecuteStatusCount>> instanceStateCounter) { |
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
boolean checkProject = checkProject(loginUser, projectId, result); |
||||||
|
if (!checkProject) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
Date start; |
||||||
|
Date end; |
||||||
|
try { |
||||||
|
start = DateUtils.getScheduleDate(startDate); |
||||||
|
end = DateUtils.getScheduleDate(endDate); |
||||||
|
} catch (Exception e) { |
||||||
|
logger.error(e.getMessage(), e); |
||||||
|
putErrorRequestParamsMsg(result); |
||||||
|
return result; |
||||||
|
} |
||||||
|
Integer[] projectIdArray = getProjectIdsArrays(loginUser, projectId); |
||||||
|
List<ExecuteStatusCount> processInstanceStateCounts = |
||||||
|
instanceStateCounter.apply(start, end, projectIdArray); |
||||||
|
|
||||||
|
if (processInstanceStateCounts != null) { |
||||||
|
TaskCountDto taskCountResult = new TaskCountDto(processInstanceStateCounts); |
||||||
|
result.put(Constants.DATA_LIST, taskCountResult); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
} |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
/** |
||||||
|
* statistics the process definition quantities of certain person |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param projectId project id |
||||||
|
* @return definition count data |
||||||
|
*/ |
||||||
|
public Map<String, Object> countDefinitionByUser(User loginUser, int projectId) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
|
||||||
|
|
||||||
|
Integer[] projectIdArray = getProjectIdsArrays(loginUser, projectId); |
||||||
|
List<DefinitionGroupByUser> defineGroupByUsers = processDefinitionMapper.countDefinitionGroupByUser( |
||||||
|
loginUser.getId(), projectIdArray, isAdmin(loginUser)); |
||||||
|
|
||||||
|
DefineUserDto dto = new DefineUserDto(defineGroupByUsers); |
||||||
|
result.put(Constants.DATA_LIST, dto); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
/** |
||||||
|
* statistical command status data |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param projectId project id |
||||||
|
* @param startDate start date |
||||||
|
* @param endDate end date |
||||||
|
* @return command state count data |
||||||
|
*/ |
||||||
|
public Map<String, Object> countCommandState(User loginUser, int projectId, String startDate, String endDate) { |
||||||
|
|
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
boolean checkProject = checkProject(loginUser, projectId, result); |
||||||
|
if (!checkProject) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* find all the task lists in the project under the user |
||||||
|
* statistics based on task status execution, failure, completion, wait, total |
||||||
|
*/ |
||||||
|
Date start = null; |
||||||
|
Date end = null; |
||||||
|
|
||||||
|
if (startDate != null && endDate != null) { |
||||||
|
try { |
||||||
|
start = DateUtils.getScheduleDate(startDate); |
||||||
|
end = DateUtils.getScheduleDate(endDate); |
||||||
|
} catch (Exception e) { |
||||||
|
logger.error(e.getMessage(), e); |
||||||
|
putErrorRequestParamsMsg(result); |
||||||
|
return result; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
Integer[] projectIdArray = getProjectIdsArrays(loginUser, projectId); |
||||||
|
// count command state
|
||||||
|
List<CommandCount> commandStateCounts = |
||||||
|
commandMapper.countCommandState( |
||||||
|
loginUser.getId(), |
||||||
|
start, |
||||||
|
end, |
||||||
|
projectIdArray); |
||||||
|
|
||||||
|
// count error command state
|
||||||
|
List<CommandCount> errorCommandStateCounts = |
||||||
|
errorCommandMapper.countCommandState( |
||||||
|
start, end, projectIdArray); |
||||||
|
|
||||||
|
// enumMap
|
||||||
|
Map<CommandType, Map<String, Integer>> dataMap = new EnumMap<>(CommandType.class); |
||||||
|
|
||||||
|
Map<String, Integer> commonCommand = new HashMap<>(); |
||||||
|
commonCommand.put(COMMAND_STATE, 0); |
||||||
|
commonCommand.put(ERROR_COMMAND_STATE, 0); |
||||||
|
|
||||||
|
|
||||||
|
// init data map
|
||||||
|
/** |
||||||
|
* START_PROCESS, START_CURRENT_TASK_PROCESS, RECOVER_TOLERANCE_FAULT_PROCESS, RECOVER_SUSPENDED_PROCESS, |
||||||
|
START_FAILURE_TASK_PROCESS,COMPLEMENT_DATA,SCHEDULER, REPEAT_RUNNING,PAUSE,STOP,RECOVER_WAITTING_THREAD; |
||||||
|
*/ |
||||||
|
dataMap.put(CommandType.START_PROCESS, commonCommand); |
||||||
|
dataMap.put(CommandType.START_CURRENT_TASK_PROCESS, commonCommand); |
||||||
|
dataMap.put(CommandType.RECOVER_TOLERANCE_FAULT_PROCESS, commonCommand); |
||||||
|
dataMap.put(CommandType.RECOVER_SUSPENDED_PROCESS, commonCommand); |
||||||
|
dataMap.put(CommandType.START_FAILURE_TASK_PROCESS, commonCommand); |
||||||
|
dataMap.put(CommandType.COMPLEMENT_DATA, commonCommand); |
||||||
|
dataMap.put(CommandType.SCHEDULER, commonCommand); |
||||||
|
dataMap.put(CommandType.REPEAT_RUNNING, commonCommand); |
||||||
|
dataMap.put(CommandType.PAUSE, commonCommand); |
||||||
|
dataMap.put(CommandType.STOP, commonCommand); |
||||||
|
dataMap.put(CommandType.RECOVER_WAITTING_THREAD, commonCommand); |
||||||
|
|
||||||
|
// put command state
|
||||||
|
for (CommandCount executeStatusCount : commandStateCounts) { |
||||||
|
Map<String, Integer> commandStateCountsMap = new HashMap<>(dataMap.get(executeStatusCount.getCommandType())); |
||||||
|
commandStateCountsMap.put(COMMAND_STATE, executeStatusCount.getCount()); |
||||||
|
dataMap.put(executeStatusCount.getCommandType(), commandStateCountsMap); |
||||||
|
} |
||||||
|
|
||||||
|
// put error command state
|
||||||
|
for (CommandCount errorExecutionStatus : errorCommandStateCounts) { |
||||||
|
Map<String, Integer> errorCommandStateCountsMap = new HashMap<>(dataMap.get(errorExecutionStatus.getCommandType())); |
||||||
|
errorCommandStateCountsMap.put(ERROR_COMMAND_STATE, errorExecutionStatus.getCount()); |
||||||
|
dataMap.put(errorExecutionStatus.getCommandType(), errorCommandStateCountsMap); |
||||||
|
} |
||||||
|
|
||||||
|
List<CommandStateCount> list = new ArrayList<>(); |
||||||
|
for (Map.Entry<CommandType, Map<String, Integer>> next : dataMap.entrySet()) { |
||||||
|
CommandStateCount commandStateCount = new CommandStateCount(next.getValue().get(ERROR_COMMAND_STATE), |
||||||
|
next.getValue().get(COMMAND_STATE), next.getKey()); |
||||||
|
list.add(commandStateCount); |
||||||
|
} |
||||||
|
|
||||||
|
result.put(Constants.DATA_LIST, list); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
private Integer[] getProjectIdsArrays(User loginUser, int projectId) { |
||||||
|
List<Integer> projectIds = new ArrayList<>(); |
||||||
|
if (projectId != 0) { |
||||||
|
projectIds.add(projectId); |
||||||
|
} else if (loginUser.getUserType() == UserType.GENERAL_USER) { |
||||||
|
projectIds = processService.getProjectIdListHavePerm(loginUser.getId()); |
||||||
|
if (projectIds.isEmpty()) { |
||||||
|
projectIds.add(0); |
||||||
|
} |
||||||
|
} |
||||||
|
return projectIds.toArray(new Integer[0]); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* count queue state |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param projectId project id |
||||||
|
* @return queue state count data |
||||||
|
*/ |
||||||
|
public Map<String, Object> countQueueState(User loginUser, int projectId) { |
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
|
||||||
|
boolean checkProject = checkProject(loginUser, projectId, result); |
||||||
|
if (!checkProject) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
// TODO tasksQueueList and tasksKillList is never updated.
|
||||||
|
List<String> tasksQueueList = new ArrayList<>(); |
||||||
|
List<String> tasksKillList = new ArrayList<>(); |
||||||
|
|
||||||
|
Map<String, Integer> dataMap = new HashMap<>(); |
||||||
|
if (loginUser.getUserType() == UserType.ADMIN_USER) { |
||||||
|
dataMap.put("taskQueue", tasksQueueList.size()); |
||||||
|
dataMap.put("taskKill", tasksKillList.size()); |
||||||
|
|
||||||
|
result.put(Constants.DATA_LIST, dataMap); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
int[] tasksQueueIds = new int[tasksQueueList.size()]; |
||||||
|
int[] tasksKillIds = new int[tasksKillList.size()]; |
||||||
|
|
||||||
|
int i = 0; |
||||||
|
for (String taskQueueStr : tasksQueueList) { |
||||||
|
if (StringUtils.isNotEmpty(taskQueueStr)) { |
||||||
|
String[] splits = taskQueueStr.split("_"); |
||||||
|
if (splits.length >= 4) { |
||||||
|
tasksQueueIds[i++] = Integer.parseInt(splits[3]); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
i = 0; |
||||||
|
for (String taskKillStr : tasksKillList) { |
||||||
|
if (StringUtils.isNotEmpty(taskKillStr)) { |
||||||
|
String[] splits = taskKillStr.split("-"); |
||||||
|
if (splits.length == 2) { |
||||||
|
tasksKillIds[i++] = Integer.parseInt(splits[1]); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
Integer taskQueueCount = 0; |
||||||
|
Integer taskKillCount = 0; |
||||||
|
|
||||||
|
Integer[] projectIds = getProjectIdsArrays(loginUser, projectId); |
||||||
|
if (tasksQueueIds.length != 0) { |
||||||
|
taskQueueCount = taskInstanceMapper.countTask( |
||||||
|
projectIds, |
||||||
|
tasksQueueIds); |
||||||
|
} |
||||||
|
|
||||||
|
if (tasksKillIds.length != 0) { |
||||||
|
taskKillCount = taskInstanceMapper.countTask(projectIds, tasksKillIds); |
||||||
|
} |
||||||
|
|
||||||
|
dataMap.put("taskQueue", taskQueueCount); |
||||||
|
dataMap.put("taskKill", taskKillCount); |
||||||
|
|
||||||
|
result.put(Constants.DATA_LIST, dataMap); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
private boolean checkProject(User loginUser, int projectId, Map<String, Object> result) { |
||||||
|
if (projectId != 0) { |
||||||
|
Project project = projectMapper.selectById(projectId); |
||||||
|
return projectService.hasProjectAndPerm(loginUser, project, result); |
||||||
|
} |
||||||
|
return true; |
||||||
|
} |
||||||
|
|
||||||
|
private void putErrorRequestParamsMsg(Map<String, Object> result) { |
||||||
|
result.put(Constants.STATUS, Status.REQUEST_PARAMS_NOT_VALID_ERROR); |
||||||
|
result.put(Constants.MSG, MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), "startDate,endDate")); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,146 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service.impl; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.api.enums.Status; |
||||||
|
import org.apache.dolphinscheduler.api.exceptions.ServiceException; |
||||||
|
import org.apache.dolphinscheduler.api.service.LoggerService; |
||||||
|
import org.apache.dolphinscheduler.api.utils.Result; |
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.common.utils.StringUtils; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.TaskInstance; |
||||||
|
import org.apache.dolphinscheduler.remote.utils.Host; |
||||||
|
import org.apache.dolphinscheduler.service.log.LogClientService; |
||||||
|
import org.apache.dolphinscheduler.service.process.ProcessService; |
||||||
|
|
||||||
|
import org.apache.commons.lang.ArrayUtils; |
||||||
|
|
||||||
|
import java.nio.charset.StandardCharsets; |
||||||
|
import java.util.Objects; |
||||||
|
|
||||||
|
import javax.annotation.PostConstruct; |
||||||
|
import javax.annotation.PreDestroy; |
||||||
|
|
||||||
|
import org.slf4j.Logger; |
||||||
|
import org.slf4j.LoggerFactory; |
||||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||||
|
import org.springframework.stereotype.Service; |
||||||
|
|
||||||
|
/** |
||||||
|
* log service |
||||||
|
*/ |
||||||
|
@Service |
||||||
|
public class LoggerServiceImpl implements LoggerService { |
||||||
|
|
||||||
|
private static final Logger logger = LoggerFactory.getLogger(LoggerServiceImpl.class); |
||||||
|
|
||||||
|
private static final String LOG_HEAD_FORMAT = "[LOG-PATH]: %s, [HOST]: %s%s"; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProcessService processService; |
||||||
|
|
||||||
|
private LogClientService logClient; |
||||||
|
|
||||||
|
@PostConstruct |
||||||
|
public void init() { |
||||||
|
if (Objects.isNull(this.logClient)) { |
||||||
|
this.logClient = new LogClientService(); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
@PreDestroy |
||||||
|
public void close() { |
||||||
|
if (Objects.nonNull(this.logClient) && this.logClient.isRunning()) { |
||||||
|
logClient.close(); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* view log |
||||||
|
* |
||||||
|
* @param taskInstId task instance id |
||||||
|
* @param skipLineNum skip line number |
||||||
|
* @param limit limit |
||||||
|
* @return log string data |
||||||
|
*/ |
||||||
|
@SuppressWarnings("unchecked") |
||||||
|
public Result<String> queryLog(int taskInstId, int skipLineNum, int limit) { |
||||||
|
|
||||||
|
TaskInstance taskInstance = processService.findTaskInstanceById(taskInstId); |
||||||
|
|
||||||
|
if (taskInstance == null || StringUtils.isBlank(taskInstance.getHost())) { |
||||||
|
return Result.error(Status.TASK_INSTANCE_NOT_FOUND); |
||||||
|
} |
||||||
|
|
||||||
|
String host = getHost(taskInstance.getHost()); |
||||||
|
|
||||||
|
Result<String> result = new Result<>(Status.SUCCESS.getCode(), Status.SUCCESS.getMsg()); |
||||||
|
|
||||||
|
logger.info("log host : {} , logPath : {} , logServer port : {}", host, taskInstance.getLogPath(), |
||||||
|
Constants.RPC_PORT); |
||||||
|
|
||||||
|
StringBuilder log = new StringBuilder(); |
||||||
|
if (skipLineNum == 0) { |
||||||
|
String head = String.format(LOG_HEAD_FORMAT, |
||||||
|
taskInstance.getLogPath(), |
||||||
|
host, |
||||||
|
Constants.SYSTEM_LINE_SEPARATOR); |
||||||
|
log.append(head); |
||||||
|
} |
||||||
|
|
||||||
|
log.append(logClient |
||||||
|
.rollViewLog(host, Constants.RPC_PORT, taskInstance.getLogPath(), skipLineNum, limit)); |
||||||
|
|
||||||
|
result.setData(log.toString()); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
/** |
||||||
|
* get log size |
||||||
|
* |
||||||
|
* @param taskInstId task instance id |
||||||
|
* @return log byte array |
||||||
|
*/ |
||||||
|
public byte[] getLogBytes(int taskInstId) { |
||||||
|
TaskInstance taskInstance = processService.findTaskInstanceById(taskInstId); |
||||||
|
if (taskInstance == null || StringUtils.isBlank(taskInstance.getHost())) { |
||||||
|
throw new ServiceException("task instance is null or host is null"); |
||||||
|
} |
||||||
|
String host = getHost(taskInstance.getHost()); |
||||||
|
byte[] head = String.format(LOG_HEAD_FORMAT, |
||||||
|
taskInstance.getLogPath(), |
||||||
|
host, |
||||||
|
Constants.SYSTEM_LINE_SEPARATOR).getBytes(StandardCharsets.UTF_8); |
||||||
|
return ArrayUtils.addAll(head, |
||||||
|
logClient.getLogBytes(host, Constants.RPC_PORT, taskInstance.getLogPath())); |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
/** |
||||||
|
* get host |
||||||
|
* |
||||||
|
* @param address address |
||||||
|
* @return old version return true ,otherwise return false |
||||||
|
*/ |
||||||
|
private String getHost(String address) { |
||||||
|
if (Boolean.TRUE.equals(Host.isOldVersion(address))) { |
||||||
|
return address; |
||||||
|
} |
||||||
|
return Host.of(address).getIp(); |
||||||
|
} |
||||||
|
} |
||||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,181 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service.impl; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.api.enums.Status; |
||||||
|
import org.apache.dolphinscheduler.api.service.BaseService; |
||||||
|
import org.apache.dolphinscheduler.api.service.ProcessDefinitionVersionService; |
||||||
|
import org.apache.dolphinscheduler.api.service.ProjectService; |
||||||
|
import org.apache.dolphinscheduler.api.utils.PageInfo; |
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessDefinitionVersion; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.Project; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.User; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionVersionMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; |
||||||
|
|
||||||
|
import java.util.HashMap; |
||||||
|
import java.util.List; |
||||||
|
import java.util.Map; |
||||||
|
import java.util.Objects; |
||||||
|
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||||
|
import org.springframework.stereotype.Service; |
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.metadata.IPage; |
||||||
|
import com.baomidou.mybatisplus.extension.plugins.pagination.Page; |
||||||
|
import com.google.common.collect.ImmutableMap; |
||||||
|
|
||||||
|
@Service |
||||||
|
public class ProcessDefinitionVersionServiceImpl extends BaseService implements |
||||||
|
ProcessDefinitionVersionService { |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProcessDefinitionVersionMapper processDefinitionVersionMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProjectService projectService; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProjectMapper projectMapper; |
||||||
|
|
||||||
|
/** |
||||||
|
* add the newest version of one process definition |
||||||
|
* |
||||||
|
* @param processDefinition the process definition that need to record version |
||||||
|
* @return the newest version number of this process definition |
||||||
|
*/ |
||||||
|
public long addProcessDefinitionVersion(ProcessDefinition processDefinition) { |
||||||
|
|
||||||
|
long version = this.queryMaxVersionByProcessDefinitionId(processDefinition.getId()) + 1; |
||||||
|
|
||||||
|
ProcessDefinitionVersion processDefinitionVersion = ProcessDefinitionVersion |
||||||
|
.newBuilder() |
||||||
|
.processDefinitionId(processDefinition.getId()) |
||||||
|
.version(version) |
||||||
|
.processDefinitionJson(processDefinition.getProcessDefinitionJson()) |
||||||
|
.description(processDefinition.getDescription()) |
||||||
|
.locations(processDefinition.getLocations()) |
||||||
|
.connects(processDefinition.getConnects()) |
||||||
|
.timeout(processDefinition.getTimeout()) |
||||||
|
.globalParams(processDefinition.getGlobalParams()) |
||||||
|
.createTime(processDefinition.getUpdateTime()) |
||||||
|
.receivers(processDefinition.getReceivers()) |
||||||
|
.receiversCc(processDefinition.getReceiversCc()) |
||||||
|
.resourceIds(processDefinition.getResourceIds()) |
||||||
|
.build(); |
||||||
|
|
||||||
|
processDefinitionVersionMapper.insert(processDefinitionVersion); |
||||||
|
|
||||||
|
return version; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query the max version number by the process definition id |
||||||
|
* |
||||||
|
* @param processDefinitionId process definition id |
||||||
|
* @return the max version number of this id |
||||||
|
*/ |
||||||
|
private long queryMaxVersionByProcessDefinitionId(int processDefinitionId) { |
||||||
|
Long maxVersion = processDefinitionVersionMapper.queryMaxVersionByProcessDefinitionId(processDefinitionId); |
||||||
|
if (Objects.isNull(maxVersion)) { |
||||||
|
return 0L; |
||||||
|
} else { |
||||||
|
return maxVersion; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query the pagination versions info by one certain process definition id |
||||||
|
* |
||||||
|
* @param loginUser login user info to check auth |
||||||
|
* @param projectName process definition project name |
||||||
|
* @param pageNo page number |
||||||
|
* @param pageSize page size |
||||||
|
* @param processDefinitionId process definition id |
||||||
|
* @return the pagination process definition versions info of the certain process definition |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryProcessDefinitionVersions(User loginUser, String projectName, int pageNo, int pageSize, int processDefinitionId) { |
||||||
|
|
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
|
||||||
|
// check the if pageNo or pageSize less than 1
|
||||||
|
if (pageNo <= 0 || pageSize <= 0) { |
||||||
|
putMsg(result |
||||||
|
, Status.QUERY_PROCESS_DEFINITION_VERSIONS_PAGE_NO_OR_PAGE_SIZE_LESS_THAN_1_ERROR |
||||||
|
, pageNo |
||||||
|
, pageSize); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
Project project = projectMapper.queryByName(projectName); |
||||||
|
|
||||||
|
// check project auth
|
||||||
|
Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName); |
||||||
|
Status resultStatus = (Status) checkResult.get(Constants.STATUS); |
||||||
|
if (resultStatus != Status.SUCCESS) { |
||||||
|
return checkResult; |
||||||
|
} |
||||||
|
|
||||||
|
PageInfo<ProcessDefinitionVersion> pageInfo = new PageInfo<>(pageNo, pageSize); |
||||||
|
Page<ProcessDefinitionVersion> page = new Page<>(pageNo, pageSize); |
||||||
|
IPage<ProcessDefinitionVersion> processDefinitionVersionsPaging = processDefinitionVersionMapper.queryProcessDefinitionVersionsPaging(page, processDefinitionId); |
||||||
|
List<ProcessDefinitionVersion> processDefinitionVersions = processDefinitionVersionsPaging.getRecords(); |
||||||
|
pageInfo.setLists(processDefinitionVersions); |
||||||
|
pageInfo.setTotalCount((int) processDefinitionVersionsPaging.getTotal()); |
||||||
|
return ImmutableMap.of( |
||||||
|
Constants.MSG, Status.SUCCESS.getMsg() |
||||||
|
, Constants.STATUS, Status.SUCCESS |
||||||
|
, Constants.DATA_LIST, pageInfo); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query one certain process definition version by version number and process definition id |
||||||
|
* |
||||||
|
* @param processDefinitionId process definition id |
||||||
|
* @param version version number |
||||||
|
* @return the process definition version info |
||||||
|
*/ |
||||||
|
public ProcessDefinitionVersion queryByProcessDefinitionIdAndVersion(int processDefinitionId, long version) { |
||||||
|
return processDefinitionVersionMapper.queryByProcessDefinitionIdAndVersion(processDefinitionId, version); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* delete one certain process definition by version number and process definition id |
||||||
|
* |
||||||
|
* @param loginUser login user info to check auth |
||||||
|
* @param projectName process definition project name |
||||||
|
* @param processDefinitionId process definition id |
||||||
|
* @param version version number |
||||||
|
* @return delele result code |
||||||
|
*/ |
||||||
|
public Map<String, Object> deleteByProcessDefinitionIdAndVersion(User loginUser, String projectName, int processDefinitionId, long version) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
Project project = projectMapper.queryByName(projectName); |
||||||
|
// check project auth
|
||||||
|
Map<String, Object> checkResult = projectService.checkProjectAndAuth(loginUser, project, projectName); |
||||||
|
Status resultStatus = (Status) checkResult.get(Constants.STATUS); |
||||||
|
if (resultStatus != Status.SUCCESS) { |
||||||
|
return checkResult; |
||||||
|
} |
||||||
|
processDefinitionVersionMapper.deleteByProcessDefinitionIdAndVersion(processDefinitionId, version); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,443 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service.impl; |
||||||
|
|
||||||
|
import static org.apache.dolphinscheduler.api.utils.CheckUtils.checkDesc; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.api.enums.Status; |
||||||
|
import org.apache.dolphinscheduler.api.service.BaseService; |
||||||
|
import org.apache.dolphinscheduler.api.service.ProjectService; |
||||||
|
import org.apache.dolphinscheduler.api.utils.PageInfo; |
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.common.enums.UserType; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.Project; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProjectUser; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.User; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProjectUserMapper; |
||||||
|
|
||||||
|
import java.util.ArrayList; |
||||||
|
import java.util.Date; |
||||||
|
import java.util.HashMap; |
||||||
|
import java.util.HashSet; |
||||||
|
import java.util.List; |
||||||
|
import java.util.Map; |
||||||
|
import java.util.Set; |
||||||
|
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||||
|
import org.springframework.stereotype.Service; |
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.metadata.IPage; |
||||||
|
import com.baomidou.mybatisplus.extension.plugins.pagination.Page; |
||||||
|
|
||||||
|
/** |
||||||
|
* project service implement |
||||||
|
**/ |
||||||
|
@Service |
||||||
|
public class ProjectServiceImpl extends BaseService implements ProjectService { |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProjectMapper projectMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProjectUserMapper projectUserMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProcessDefinitionMapper processDefinitionMapper; |
||||||
|
|
||||||
|
/** |
||||||
|
* create project |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param name project name |
||||||
|
* @param desc description |
||||||
|
* @return returns an error if it exists |
||||||
|
*/ |
||||||
|
public Map<String, Object> createProject(User loginUser, String name, String desc) { |
||||||
|
|
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
Map<String, Object> descCheck = checkDesc(desc); |
||||||
|
if (descCheck.get(Constants.STATUS) != Status.SUCCESS) { |
||||||
|
return descCheck; |
||||||
|
} |
||||||
|
|
||||||
|
Project project = projectMapper.queryByName(name); |
||||||
|
if (project != null) { |
||||||
|
putMsg(result, Status.PROJECT_ALREADY_EXISTS, name); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
Date now = new Date(); |
||||||
|
|
||||||
|
project = Project |
||||||
|
.newBuilder() |
||||||
|
.name(name) |
||||||
|
.description(desc) |
||||||
|
.userId(loginUser.getId()) |
||||||
|
.userName(loginUser.getUserName()) |
||||||
|
.createTime(now) |
||||||
|
.updateTime(now) |
||||||
|
.build(); |
||||||
|
|
||||||
|
if (projectMapper.insert(project) > 0) { |
||||||
|
Project insertedProject = projectMapper.queryByName(name); |
||||||
|
result.put(Constants.DATA_LIST, insertedProject); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
} else { |
||||||
|
putMsg(result, Status.CREATE_PROJECT_ERROR); |
||||||
|
} |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query project details by id |
||||||
|
* |
||||||
|
* @param projectId project id |
||||||
|
* @return project detail information |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryById(Integer projectId) { |
||||||
|
|
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
Project project = projectMapper.selectById(projectId); |
||||||
|
|
||||||
|
if (project != null) { |
||||||
|
result.put(Constants.DATA_LIST, project); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
} else { |
||||||
|
putMsg(result, Status.PROJECT_NOT_FOUNT, projectId); |
||||||
|
} |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* check project and authorization |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param project project |
||||||
|
* @param projectName project name |
||||||
|
* @return true if the login user have permission to see the project |
||||||
|
*/ |
||||||
|
public Map<String, Object> checkProjectAndAuth(User loginUser, Project project, String projectName) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
if (project == null) { |
||||||
|
putMsg(result, Status.PROJECT_NOT_FOUNT, projectName); |
||||||
|
} else if (!checkReadPermission(loginUser, project)) { |
||||||
|
// check read permission
|
||||||
|
putMsg(result, Status.USER_NO_OPERATION_PROJECT_PERM, loginUser.getUserName(), projectName); |
||||||
|
} else { |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
} |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
public boolean hasProjectAndPerm(User loginUser, Project project, Map<String, Object> result) { |
||||||
|
boolean checkResult = false; |
||||||
|
if (project == null) { |
||||||
|
putMsg(result, Status.PROJECT_NOT_FOUNT, ""); |
||||||
|
} else if (!checkReadPermission(loginUser, project)) { |
||||||
|
putMsg(result, Status.USER_NO_OPERATION_PROJECT_PERM, loginUser.getUserName(), project.getName()); |
||||||
|
} else { |
||||||
|
checkResult = true; |
||||||
|
} |
||||||
|
return checkResult; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* admin can view all projects |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param searchVal search value |
||||||
|
* @param pageSize page size |
||||||
|
* @param pageNo page number |
||||||
|
* @return project list which the login user have permission to see |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryProjectListPaging(User loginUser, Integer pageSize, Integer pageNo, String searchVal) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
PageInfo<Project> pageInfo = new PageInfo<>(pageNo, pageSize); |
||||||
|
|
||||||
|
Page<Project> page = new Page<>(pageNo, pageSize); |
||||||
|
|
||||||
|
int userId = loginUser.getUserType() == UserType.ADMIN_USER ? 0 : loginUser.getId(); |
||||||
|
IPage<Project> projectIPage = projectMapper.queryProjectListPaging(page, userId, searchVal); |
||||||
|
|
||||||
|
List<Project> projectList = projectIPage.getRecords(); |
||||||
|
if (userId != 0) { |
||||||
|
for (Project project : projectList) { |
||||||
|
project.setPerm(Constants.DEFAULT_ADMIN_PERMISSION); |
||||||
|
} |
||||||
|
} |
||||||
|
pageInfo.setTotalCount((int) projectIPage.getTotal()); |
||||||
|
pageInfo.setLists(projectList); |
||||||
|
result.put(Constants.COUNT, (int) projectIPage.getTotal()); |
||||||
|
result.put(Constants.DATA_LIST, pageInfo); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
|
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* delete project by id |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param projectId project id |
||||||
|
* @return delete result code |
||||||
|
*/ |
||||||
|
public Map<String, Object> deleteProject(User loginUser, Integer projectId) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
Project project = projectMapper.selectById(projectId); |
||||||
|
Map<String, Object> checkResult = getCheckResult(loginUser, project); |
||||||
|
if (checkResult != null) { |
||||||
|
return checkResult; |
||||||
|
} |
||||||
|
|
||||||
|
if (!hasPerm(loginUser, project.getUserId())) { |
||||||
|
putMsg(result, Status.USER_NO_OPERATION_PERM); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
List<ProcessDefinition> processDefinitionList = processDefinitionMapper.queryAllDefinitionList(projectId); |
||||||
|
|
||||||
|
if (!processDefinitionList.isEmpty()) { |
||||||
|
putMsg(result, Status.DELETE_PROJECT_ERROR_DEFINES_NOT_NULL); |
||||||
|
return result; |
||||||
|
} |
||||||
|
int delete = projectMapper.deleteById(projectId); |
||||||
|
if (delete > 0) { |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
} else { |
||||||
|
putMsg(result, Status.DELETE_PROJECT_ERROR); |
||||||
|
} |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* get check result |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param project project |
||||||
|
* @return check result |
||||||
|
*/ |
||||||
|
private Map<String, Object> getCheckResult(User loginUser, Project project) { |
||||||
|
String projectName = project == null ? null : project.getName(); |
||||||
|
Map<String, Object> checkResult = checkProjectAndAuth(loginUser, project, projectName); |
||||||
|
Status status = (Status) checkResult.get(Constants.STATUS); |
||||||
|
if (status != Status.SUCCESS) { |
||||||
|
return checkResult; |
||||||
|
} |
||||||
|
return null; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* updateProcessInstance project |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param projectId project id |
||||||
|
* @param projectName project name |
||||||
|
* @param desc description |
||||||
|
* @return update result code |
||||||
|
*/ |
||||||
|
public Map<String, Object> update(User loginUser, Integer projectId, String projectName, String desc) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
|
||||||
|
Map<String, Object> descCheck = checkDesc(desc); |
||||||
|
if (descCheck.get(Constants.STATUS) != Status.SUCCESS) { |
||||||
|
return descCheck; |
||||||
|
} |
||||||
|
|
||||||
|
Project project = projectMapper.selectById(projectId); |
||||||
|
boolean hasProjectAndPerm = hasProjectAndPerm(loginUser, project, result); |
||||||
|
if (!hasProjectAndPerm) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
Project tempProject = projectMapper.queryByName(projectName); |
||||||
|
if (tempProject != null && tempProject.getId() != projectId) { |
||||||
|
putMsg(result, Status.PROJECT_ALREADY_EXISTS, projectName); |
||||||
|
return result; |
||||||
|
} |
||||||
|
project.setName(projectName); |
||||||
|
project.setDescription(desc); |
||||||
|
project.setUpdateTime(new Date()); |
||||||
|
|
||||||
|
int update = projectMapper.updateById(project); |
||||||
|
if (update > 0) { |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
} else { |
||||||
|
putMsg(result, Status.UPDATE_PROJECT_ERROR); |
||||||
|
} |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
/** |
||||||
|
* query unauthorized project |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param userId user id |
||||||
|
* @return the projects which user have not permission to see |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryUnauthorizedProject(User loginUser, Integer userId) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
if (checkAdmin(loginUser, result)) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
/** |
||||||
|
* query all project list except specified userId |
||||||
|
*/ |
||||||
|
List<Project> projectList = projectMapper.queryProjectExceptUserId(userId); |
||||||
|
List<Project> resultList = new ArrayList<>(); |
||||||
|
Set<Project> projectSet = null; |
||||||
|
if (projectList != null && !projectList.isEmpty()) { |
||||||
|
projectSet = new HashSet<>(projectList); |
||||||
|
|
||||||
|
List<Project> authedProjectList = projectMapper.queryAuthedProjectListByUserId(userId); |
||||||
|
|
||||||
|
resultList = getUnauthorizedProjects(projectSet, authedProjectList); |
||||||
|
} |
||||||
|
result.put(Constants.DATA_LIST, resultList); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* get unauthorized project |
||||||
|
* |
||||||
|
* @param projectSet project set |
||||||
|
* @param authedProjectList authed project list |
||||||
|
* @return project list that authorization |
||||||
|
*/ |
||||||
|
private List<Project> getUnauthorizedProjects(Set<Project> projectSet, List<Project> authedProjectList) { |
||||||
|
List<Project> resultList; |
||||||
|
Set<Project> authedProjectSet = null; |
||||||
|
if (authedProjectList != null && !authedProjectList.isEmpty()) { |
||||||
|
authedProjectSet = new HashSet<>(authedProjectList); |
||||||
|
projectSet.removeAll(authedProjectSet); |
||||||
|
|
||||||
|
} |
||||||
|
resultList = new ArrayList<>(projectSet); |
||||||
|
return resultList; |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
/** |
||||||
|
* query authorized project |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param userId user id |
||||||
|
* @return projects which the user have permission to see, Except for items created by this user |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryAuthorizedProject(User loginUser, Integer userId) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
|
||||||
|
if (checkAdmin(loginUser, result)) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
List<Project> projects = projectMapper.queryAuthedProjectListByUserId(userId); |
||||||
|
result.put(Constants.DATA_LIST, projects); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
|
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query authorized project |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @return projects which the user have permission to see, Except for items created by this user |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryProjectCreatedByUser(User loginUser) { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
|
||||||
|
if (checkAdmin(loginUser, result)) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
List<Project> projects = projectMapper.queryProjectCreatedByUser(loginUser.getId()); |
||||||
|
result.put(Constants.DATA_LIST, projects); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
|
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* check whether have read permission |
||||||
|
* |
||||||
|
* @param user user |
||||||
|
* @param project project |
||||||
|
* @return true if the user have permission to see the project, otherwise return false |
||||||
|
*/ |
||||||
|
private boolean checkReadPermission(User user, Project project) { |
||||||
|
int permissionId = queryPermission(user, project); |
||||||
|
return (permissionId & Constants.READ_PERMISSION) != 0; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query permission id |
||||||
|
* |
||||||
|
* @param user user |
||||||
|
* @param project project |
||||||
|
* @return permission |
||||||
|
*/ |
||||||
|
private int queryPermission(User user, Project project) { |
||||||
|
if (user.getUserType() == UserType.ADMIN_USER) { |
||||||
|
return Constants.READ_PERMISSION; |
||||||
|
} |
||||||
|
|
||||||
|
if (project.getUserId() == user.getId()) { |
||||||
|
return Constants.ALL_PERMISSIONS; |
||||||
|
} |
||||||
|
|
||||||
|
ProjectUser projectUser = projectUserMapper.queryProjectRelation(project.getId(), user.getId()); |
||||||
|
|
||||||
|
if (projectUser == null) { |
||||||
|
return 0; |
||||||
|
} |
||||||
|
|
||||||
|
return projectUser.getPerm(); |
||||||
|
|
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query all project list that have one or more process definitions. |
||||||
|
* |
||||||
|
* @return project list |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryAllProjectList() { |
||||||
|
Map<String, Object> result = new HashMap<>(); |
||||||
|
List<Project> projects = projectMapper.selectList(null); |
||||||
|
List<ProcessDefinition> processDefinitions = processDefinitionMapper.selectList(null); |
||||||
|
if (projects != null) { |
||||||
|
Set<Integer> set = new HashSet<>(); |
||||||
|
for (ProcessDefinition processDefinition : processDefinitions) { |
||||||
|
set.add(processDefinition.getProjectId()); |
||||||
|
} |
||||||
|
List<Project> tempDeletelist = new ArrayList<>(); |
||||||
|
for (Project project : projects) { |
||||||
|
if (!set.contains(project.getId())) { |
||||||
|
tempDeletelist.add(project); |
||||||
|
} |
||||||
|
} |
||||||
|
projects.removeAll(tempDeletelist); |
||||||
|
} |
||||||
|
result.put(Constants.DATA_LIST, projects); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,158 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service.impl; |
||||||
|
|
||||||
|
import java.util.Date; |
||||||
|
import java.util.List; |
||||||
|
import java.util.UUID; |
||||||
|
|
||||||
|
import javax.servlet.http.Cookie; |
||||||
|
import javax.servlet.http.HttpServletRequest; |
||||||
|
|
||||||
|
import org.apache.commons.lang.StringUtils; |
||||||
|
import org.apache.dolphinscheduler.api.controller.BaseController; |
||||||
|
import org.apache.dolphinscheduler.api.service.BaseService; |
||||||
|
import org.apache.dolphinscheduler.api.service.SessionService; |
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.common.utils.CollectionUtils; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.Session; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.User; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.SessionMapper; |
||||||
|
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; |
||||||
|
|
||||||
|
/** |
||||||
|
* session service implement |
||||||
|
*/ |
||||||
|
@Service |
||||||
|
public class SessionServiceImpl extends BaseService implements SessionService { |
||||||
|
|
||||||
|
private static final Logger logger = LoggerFactory.getLogger(SessionService.class); |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private SessionMapper sessionMapper; |
||||||
|
|
||||||
|
/** |
||||||
|
* get user session from request |
||||||
|
* |
||||||
|
* @param request request |
||||||
|
* @return session |
||||||
|
*/ |
||||||
|
public Session getSession(HttpServletRequest request) { |
||||||
|
String sessionId = request.getHeader(Constants.SESSION_ID); |
||||||
|
|
||||||
|
if (StringUtils.isBlank(sessionId)) { |
||||||
|
Cookie cookie = getCookie(request, Constants.SESSION_ID); |
||||||
|
|
||||||
|
if (cookie != null) { |
||||||
|
sessionId = cookie.getValue(); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
if (StringUtils.isBlank(sessionId)) { |
||||||
|
return null; |
||||||
|
} |
||||||
|
|
||||||
|
String ip = BaseController.getClientIpAddress(request); |
||||||
|
logger.debug("get session: {}, ip: {}", sessionId, ip); |
||||||
|
|
||||||
|
return sessionMapper.selectById(sessionId); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* create session |
||||||
|
* |
||||||
|
* @param user user |
||||||
|
* @param ip ip |
||||||
|
* @return session string |
||||||
|
*/ |
||||||
|
@Transactional(rollbackFor = RuntimeException.class) |
||||||
|
public String createSession(User user, String ip) { |
||||||
|
Session session = null; |
||||||
|
|
||||||
|
// logined
|
||||||
|
List<Session> sessionList = sessionMapper.queryByUserId(user.getId()); |
||||||
|
|
||||||
|
Date now = new Date(); |
||||||
|
|
||||||
|
/** |
||||||
|
* if you have logged in and are still valid, return directly |
||||||
|
*/ |
||||||
|
if (CollectionUtils.isNotEmpty(sessionList)) { |
||||||
|
// is session list greater 1 ， delete other ，get one
|
||||||
|
if (sessionList.size() > 1) { |
||||||
|
for (int i = 1; i < sessionList.size(); i++) { |
||||||
|
sessionMapper.deleteById(sessionList.get(i).getId()); |
||||||
|
} |
||||||
|
} |
||||||
|
session = sessionList.get(0); |
||||||
|
if (now.getTime() - session.getLastLoginTime().getTime() <= Constants.SESSION_TIME_OUT * 1000) { |
||||||
|
/** |
||||||
|
* updateProcessInstance the latest login time |
||||||
|
*/ |
||||||
|
session.setLastLoginTime(now); |
||||||
|
sessionMapper.updateById(session); |
||||||
|
|
||||||
|
return session.getId(); |
||||||
|
|
||||||
|
} else { |
||||||
|
/** |
||||||
|
* session expired, then delete this session first |
||||||
|
*/ |
||||||
|
sessionMapper.deleteById(session.getId()); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
// assign new session
|
||||||
|
session = new Session(); |
||||||
|
|
||||||
|
session.setId(UUID.randomUUID().toString()); |
||||||
|
session.setIp(ip); |
||||||
|
session.setUserId(user.getId()); |
||||||
|
session.setLastLoginTime(now); |
||||||
|
|
||||||
|
sessionMapper.insert(session); |
||||||
|
|
||||||
|
return session.getId(); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* sign out |
||||||
|
* remove ip restrictions |
||||||
|
* |
||||||
|
* @param ip no use |
||||||
|
* @param loginUser login user |
||||||
|
*/ |
||||||
|
public void signOut(String ip, User loginUser) { |
||||||
|
try { |
||||||
|
/** |
||||||
|
* query session by user id and ip |
||||||
|
*/ |
||||||
|
Session session = sessionMapper.queryByUserIdAndIp(loginUser.getId(), ip); |
||||||
|
|
||||||
|
//delete session
|
||||||
|
sessionMapper.deleteById(session.getId()); |
||||||
|
} catch (Exception e) { |
||||||
|
logger.warn("userId : {} , ip : {} , find more one session", loginUser.getId(), ip); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,331 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service.impl; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.api.enums.Status; |
||||||
|
import org.apache.dolphinscheduler.api.service.BaseService; |
||||||
|
import org.apache.dolphinscheduler.api.service.TenantService; |
||||||
|
import org.apache.dolphinscheduler.api.utils.PageInfo; |
||||||
|
import org.apache.dolphinscheduler.api.utils.Result; |
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.common.utils.CollectionUtils; |
||||||
|
import org.apache.dolphinscheduler.common.utils.HadoopUtils; |
||||||
|
import org.apache.dolphinscheduler.common.utils.PropertyUtils; |
||||||
|
import org.apache.dolphinscheduler.common.utils.StringUtils; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessInstance; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.Tenant; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.User; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.TenantMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.UserMapper; |
||||||
|
|
||||||
|
import java.util.Date; |
||||||
|
import java.util.HashMap; |
||||||
|
import java.util.List; |
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
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 com.baomidou.mybatisplus.core.metadata.IPage; |
||||||
|
import com.baomidou.mybatisplus.extension.plugins.pagination.Page; |
||||||
|
|
||||||
|
/** |
||||||
|
* tenant service |
||||||
|
*/ |
||||||
|
@Service |
||||||
|
public class TenantServiceImpl extends BaseService implements TenantService { |
||||||
|
|
||||||
|
private static final Logger logger = LoggerFactory.getLogger(TenantServiceImpl.class); |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private TenantMapper tenantMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProcessInstanceMapper processInstanceMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private ProcessDefinitionMapper processDefinitionMapper; |
||||||
|
|
||||||
|
@Autowired |
||||||
|
private UserMapper userMapper; |
||||||
|
|
||||||
|
/** |
||||||
|
* create tenant |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param tenantCode tenant code |
||||||
|
* @param tenantName tenant name |
||||||
|
* @param queueId queue id |
||||||
|
* @param desc description |
||||||
|
* @return create result code |
||||||
|
* @throws Exception exception |
||||||
|
*/ |
||||||
|
@Transactional(rollbackFor = Exception.class) |
||||||
|
public Map<String, Object> createTenant(User loginUser, |
||||||
|
String tenantCode, |
||||||
|
String tenantName, |
||||||
|
int queueId, |
||||||
|
String desc) throws Exception { |
||||||
|
|
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
result.put(Constants.STATUS, false); |
||||||
|
if (checkAdmin(loginUser, result)) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
if (checkTenantExists(tenantCode)) { |
||||||
|
putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, tenantCode); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
Tenant tenant = new Tenant(); |
||||||
|
Date now = new Date(); |
||||||
|
|
||||||
|
if (!tenantCode.matches("^[0-9a-zA-Z_.-]{1,}$") || tenantCode.startsWith("-") || tenantCode.startsWith(".")) { |
||||||
|
putMsg(result, Status.VERIFY_TENANT_CODE_ERROR); |
||||||
|
return result; |
||||||
|
} |
||||||
|
tenant.setTenantCode(tenantCode); |
||||||
|
tenant.setTenantName(tenantName); |
||||||
|
tenant.setQueueId(queueId); |
||||||
|
tenant.setDescription(desc); |
||||||
|
tenant.setCreateTime(now); |
||||||
|
tenant.setUpdateTime(now); |
||||||
|
|
||||||
|
// save
|
||||||
|
tenantMapper.insert(tenant); |
||||||
|
|
||||||
|
// if hdfs startup
|
||||||
|
if (PropertyUtils.getResUploadStartupState()) { |
||||||
|
createTenantDirIfNotExists(tenantCode); |
||||||
|
} |
||||||
|
|
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
|
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query tenant list paging |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param searchVal search value |
||||||
|
* @param pageNo page number |
||||||
|
* @param pageSize page size |
||||||
|
* @return tenant list page |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryTenantList(User loginUser, String searchVal, Integer pageNo, Integer pageSize) { |
||||||
|
|
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
if (checkAdmin(loginUser, result)) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
Page<Tenant> page = new Page<>(pageNo, pageSize); |
||||||
|
IPage<Tenant> tenantIPage = tenantMapper.queryTenantPaging(page, searchVal); |
||||||
|
PageInfo<Tenant> pageInfo = new PageInfo<>(pageNo, pageSize); |
||||||
|
pageInfo.setTotalCount((int) tenantIPage.getTotal()); |
||||||
|
pageInfo.setLists(tenantIPage.getRecords()); |
||||||
|
result.put(Constants.DATA_LIST, pageInfo); |
||||||
|
|
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
|
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* updateProcessInstance tenant |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param id tennat id |
||||||
|
* @param tenantCode tennat code |
||||||
|
* @param tenantName tennat name |
||||||
|
* @param queueId queue id |
||||||
|
* @param desc description |
||||||
|
* @return update result code |
||||||
|
* @throws Exception exception |
||||||
|
*/ |
||||||
|
public Map<String, Object> updateTenant(User loginUser, int id, String tenantCode, String tenantName, int queueId, |
||||||
|
String desc) throws Exception { |
||||||
|
|
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
result.put(Constants.STATUS, false); |
||||||
|
|
||||||
|
if (checkAdmin(loginUser, result)) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
Tenant tenant = tenantMapper.queryById(id); |
||||||
|
|
||||||
|
if (tenant == null) { |
||||||
|
putMsg(result, Status.TENANT_NOT_EXIST); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
// updateProcessInstance tenant
|
||||||
|
/** |
||||||
|
* if the tenant code is modified, the original resource needs to be copied to the new tenant. |
||||||
|
*/ |
||||||
|
if (!tenant.getTenantCode().equals(tenantCode)) { |
||||||
|
if (checkTenantExists(tenantCode)) { |
||||||
|
// if hdfs startup
|
||||||
|
if (PropertyUtils.getResUploadStartupState()) { |
||||||
|
String resourcePath = HadoopUtils.getHdfsDataBasePath() + "/" + tenantCode + "/resources"; |
||||||
|
String udfsPath = HadoopUtils.getHdfsUdfDir(tenantCode); |
||||||
|
//init hdfs resource
|
||||||
|
HadoopUtils.getInstance().mkdir(resourcePath); |
||||||
|
HadoopUtils.getInstance().mkdir(udfsPath); |
||||||
|
} |
||||||
|
} else { |
||||||
|
putMsg(result, Status.TENANT_CODE_HAS_ALREADY_EXISTS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
Date now = new Date(); |
||||||
|
|
||||||
|
if (StringUtils.isNotEmpty(tenantCode)) { |
||||||
|
tenant.setTenantCode(tenantCode); |
||||||
|
} |
||||||
|
|
||||||
|
if (StringUtils.isNotEmpty(tenantName)) { |
||||||
|
tenant.setTenantName(tenantName); |
||||||
|
} |
||||||
|
|
||||||
|
if (queueId != 0) { |
||||||
|
tenant.setQueueId(queueId); |
||||||
|
} |
||||||
|
tenant.setDescription(desc); |
||||||
|
tenant.setUpdateTime(now); |
||||||
|
tenantMapper.updateById(tenant); |
||||||
|
|
||||||
|
result.put(Constants.STATUS, Status.SUCCESS); |
||||||
|
result.put(Constants.MSG, Status.SUCCESS.getMsg()); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* delete tenant |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @param id tenant id |
||||||
|
* @return delete result code |
||||||
|
* @throws Exception exception |
||||||
|
*/ |
||||||
|
@Transactional(rollbackFor = Exception.class) |
||||||
|
public Map<String, Object> deleteTenantById(User loginUser, int id) throws Exception { |
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
|
||||||
|
if (checkAdmin(loginUser, result)) { |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
Tenant tenant = tenantMapper.queryById(id); |
||||||
|
if (tenant == null) { |
||||||
|
putMsg(result, Status.TENANT_NOT_EXIST); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
List<ProcessInstance> processInstances = getProcessInstancesByTenant(tenant); |
||||||
|
if (CollectionUtils.isNotEmpty(processInstances)) { |
||||||
|
putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL, processInstances.size()); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
List<ProcessDefinition> processDefinitions = |
||||||
|
processDefinitionMapper.queryDefinitionListByTenant(tenant.getId()); |
||||||
|
if (CollectionUtils.isNotEmpty(processDefinitions)) { |
||||||
|
putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL_DEFINES, processDefinitions.size()); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
List<User> userList = userMapper.queryUserListByTenant(tenant.getId()); |
||||||
|
if (CollectionUtils.isNotEmpty(userList)) { |
||||||
|
putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL_USERS, userList.size()); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
// if resource upload startup
|
||||||
|
if (PropertyUtils.getResUploadStartupState()) { |
||||||
|
String tenantPath = HadoopUtils.getHdfsDataBasePath() + "/" + tenant.getTenantCode(); |
||||||
|
|
||||||
|
if (HadoopUtils.getInstance().exists(tenantPath)) { |
||||||
|
HadoopUtils.getInstance().delete(tenantPath, true); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
tenantMapper.deleteById(id); |
||||||
|
processInstanceMapper.updateProcessInstanceByTenantId(id, -1); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
private List<ProcessInstance> getProcessInstancesByTenant(Tenant tenant) { |
||||||
|
return processInstanceMapper.queryByTenantIdAndStatus(tenant.getId(), Constants.NOT_TERMINATED_STATES); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* query tenant list |
||||||
|
* |
||||||
|
* @param loginUser login user |
||||||
|
* @return tenant list |
||||||
|
*/ |
||||||
|
public Map<String, Object> queryTenantList(User loginUser) { |
||||||
|
|
||||||
|
Map<String, Object> result = new HashMap<>(5); |
||||||
|
|
||||||
|
List<Tenant> resourceList = tenantMapper.selectList(null); |
||||||
|
result.put(Constants.DATA_LIST, resourceList); |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
|
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* verify tenant code |
||||||
|
* |
||||||
|
* @param tenantCode tenant code |
||||||
|
* @return true if tenant code can user, otherwise return false |
||||||
|
*/ |
||||||
|
public Result verifyTenantCode(String tenantCode) { |
||||||
|
Result result = new Result(); |
||||||
|
if (checkTenantExists(tenantCode)) { |
||||||
|
putMsg(result, Status.TENANT_NAME_EXIST, tenantCode); |
||||||
|
} else { |
||||||
|
putMsg(result, Status.SUCCESS); |
||||||
|
} |
||||||
|
return result; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* check tenant exists |
||||||
|
* |
||||||
|
* @param tenantCode tenant code |
||||||
|
* @return ture if the tenant code exists, otherwise return false |
||||||
|
*/ |
||||||
|
private boolean checkTenantExists(String tenantCode) { |
||||||
|
List<Tenant> tenants = tenantMapper.queryByTenantCode(tenantCode); |
||||||
|
return CollectionUtils.isNotEmpty(tenants); |
||||||
|
} |
||||||
|
} |
||||||
@ -1,50 +0,0 @@ |
|||||||
/* |
|
||||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
|
||||||
* contributor license agreements. See the NOTICE file distributed with |
|
||||||
* this work for additional information regarding copyright ownership. |
|
||||||
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service; |
|
||||||
|
|
||||||
import org.apache.dolphinscheduler.common.graph.DAG; |
|
||||||
import org.apache.dolphinscheduler.common.model.TaskNode; |
|
||||||
import org.apache.dolphinscheduler.common.model.TaskNodeRelation; |
|
||||||
import org.apache.dolphinscheduler.dao.entity.ProcessInstance; |
|
||||||
import org.junit.Assert; |
|
||||||
import org.junit.Test; |
|
||||||
import org.junit.runner.RunWith; |
|
||||||
import org.mockito.junit.MockitoJUnitRunner; |
|
||||||
|
|
||||||
@RunWith(MockitoJUnitRunner.class) |
|
||||||
public class BaseDAGServiceTest { |
|
||||||
|
|
||||||
@Test |
|
||||||
public void testProcessInstance2DAG(){ |
|
||||||
|
|
||||||
ProcessInstance processInstance = new ProcessInstance(); |
|
||||||
processInstance.setProcessInstanceJson("{\"globalParams\":[],\"tasks\":[{\"type\":\"SHELL\",\"id\":\"tasks-61567\"," + |
|
||||||
"\"name\":\"开始\",\"params\":{\"resourceList\":[],\"localParams\":[],\"rawScript\":\"echo '1'\"}," + |
|
||||||
"\"description\":\"\",\"runFlag\":\"NORMAL\",\"dependence\":{},\"maxRetryTimes\":\"0\",\"retryInterval\":\"1\"," + |
|
||||||
"\"timeout\":{\"strategy\":\"\",\"interval\":null,\"enable\":false},\"taskInstancePriority\":\"MEDIUM\"," + |
|
||||||
"\"workerGroupId\":-1,\"preTasks\":[]},{\"type\":\"SHELL\",\"id\":\"tasks-6-3ug5ej\",\"name\":\"结束\"," + |
|
||||||
"\"params\":{\"resourceList\":[],\"localParams\":[],\"rawScript\":\"echo '1'\"},\"description\":\"\"," + |
|
||||||
"\"runFlag\":\"NORMAL\",\"dependence\":{},\"maxRetryTimes\":\"0\",\"retryInterval\":\"1\"," + |
|
||||||
"\"timeout\":{\"strategy\":\"\",\"interval\":null,\"enable\":false},\"taskInstancePriority\":\"MEDIUM\"," + |
|
||||||
"\"workerGroupId\":-1,\"preTasks\":[\"开始\"]}],\"tenantId\":-1,\"timeout\":0}"); |
|
||||||
|
|
||||||
DAG<String, TaskNode, TaskNodeRelation> relationDAG = BaseDAGService.processInstance2DAG(processInstance); |
|
||||||
|
|
||||||
Assert.assertTrue(relationDAG.containsNode("开始")); |
|
||||||
|
|
||||||
} |
|
||||||
} |
|
||||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,274 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.api.service; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.api.enums.Status; |
||||||
|
import org.apache.dolphinscheduler.api.service.impl.ProcessDefinitionVersionServiceImpl; |
||||||
|
import org.apache.dolphinscheduler.api.service.impl.ProjectServiceImpl; |
||||||
|
import org.apache.dolphinscheduler.api.utils.PageInfo; |
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.common.enums.UserType; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.ProcessDefinitionVersion; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.Project; |
||||||
|
import org.apache.dolphinscheduler.dao.entity.User; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionVersionMapper; |
||||||
|
import org.apache.dolphinscheduler.dao.mapper.ProjectMapper; |
||||||
|
|
||||||
|
import java.text.MessageFormat; |
||||||
|
import java.util.HashMap; |
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
import org.junit.Assert; |
||||||
|
import org.junit.Test; |
||||||
|
import org.junit.runner.RunWith; |
||||||
|
import org.mockito.InjectMocks; |
||||||
|
import org.mockito.Mock; |
||||||
|
import org.mockito.Mockito; |
||||||
|
import org.mockito.junit.MockitoJUnitRunner; |
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.extension.plugins.pagination.Page; |
||||||
|
import com.google.common.collect.Lists; |
||||||
|
|
||||||
|
@RunWith(MockitoJUnitRunner.class) |
||||||
|
public class ProcessDefinitionVersionServiceTest { |
||||||
|
|
||||||
|
@InjectMocks |
||||||
|
private ProcessDefinitionVersionServiceImpl processDefinitionVersionService; |
||||||
|
|
||||||
|
@Mock |
||||||
|
private ProcessDefinitionVersionMapper processDefinitionVersionMapper; |
||||||
|
|
||||||
|
@Mock |
||||||
|
private ProjectMapper projectMapper; |
||||||
|
|
||||||
|
@Mock |
||||||
|
private ProjectServiceImpl projectService; |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testAddProcessDefinitionVersion() { |
||||||
|
long expectedVersion = 5L; |
||||||
|
ProcessDefinition processDefinition = getProcessDefinition(); |
||||||
|
Mockito.when(processDefinitionVersionMapper |
||||||
|
.queryMaxVersionByProcessDefinitionId(processDefinition.getId())) |
||||||
|
.thenReturn(expectedVersion); |
||||||
|
|
||||||
|
long version = processDefinitionVersionService.addProcessDefinitionVersion(processDefinition); |
||||||
|
|
||||||
|
Assert.assertEquals(expectedVersion + 1, version); |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
@SuppressWarnings("unchecked") |
||||||
|
public void testQueryProcessDefinitionVersions() { |
||||||
|
// pageNo <= 0
|
||||||
|
int pageNo = -1; |
||||||
|
int pageSize = 10; |
||||||
|
int processDefinitionId = 66; |
||||||
|
|
||||||
|
String projectName = "project_test1"; |
||||||
|
User loginUser = new User(); |
||||||
|
loginUser.setId(-1); |
||||||
|
loginUser.setUserType(UserType.GENERAL_USER); |
||||||
|
Map<String, Object> resultMap1 = processDefinitionVersionService.queryProcessDefinitionVersions( |
||||||
|
loginUser |
||||||
|
, projectName |
||||||
|
, pageNo |
||||||
|
, pageSize |
||||||
|
, processDefinitionId); |
||||||
|
Assert.assertEquals(Status.QUERY_PROCESS_DEFINITION_VERSIONS_PAGE_NO_OR_PAGE_SIZE_LESS_THAN_1_ERROR |
||||||
|
, resultMap1.get(Constants.STATUS)); |
||||||
|
|
||||||
|
// pageSize <= 0
|
||||||
|
pageNo = 1; |
||||||
|
pageSize = -1; |
||||||
|
Map<String, Object> resultMap2 = processDefinitionVersionService.queryProcessDefinitionVersions( |
||||||
|
loginUser |
||||||
|
, projectName |
||||||
|
, pageNo |
||||||
|
, pageSize |
||||||
|
, processDefinitionId); |
||||||
|
Assert.assertEquals(Status.QUERY_PROCESS_DEFINITION_VERSIONS_PAGE_NO_OR_PAGE_SIZE_LESS_THAN_1_ERROR |
||||||
|
, resultMap2.get(Constants.STATUS)); |
||||||
|
|
||||||
|
Map<String, Object> res = new HashMap<>(); |
||||||
|
putMsg(res, Status.PROJECT_NOT_FOUNT); |
||||||
|
Project project = getProject(projectName); |
||||||
|
Mockito.when(projectMapper.queryByName(projectName)) |
||||||
|
.thenReturn(project); |
||||||
|
Mockito.when(projectService.checkProjectAndAuth(loginUser, project, projectName)) |
||||||
|
.thenReturn(res); |
||||||
|
|
||||||
|
// project auth fail
|
||||||
|
pageNo = 1; |
||||||
|
pageSize = 10; |
||||||
|
Map<String, Object> resultMap3 = processDefinitionVersionService.queryProcessDefinitionVersions( |
||||||
|
loginUser |
||||||
|
, projectName |
||||||
|
, pageNo |
||||||
|
, pageSize |
||||||
|
, processDefinitionId); |
||||||
|
Assert.assertEquals(Status.PROJECT_NOT_FOUNT, resultMap3.get(Constants.STATUS)); |
||||||
|
|
||||||
|
putMsg(res, Status.SUCCESS); |
||||||
|
|
||||||
|
Mockito.when(projectService.checkProjectAndAuth(loginUser, project, projectName)) |
||||||
|
.thenReturn(res); |
||||||
|
|
||||||
|
ProcessDefinitionVersion processDefinitionVersion = getProcessDefinitionVersion(getProcessDefinition()); |
||||||
|
|
||||||
|
Mockito.when(processDefinitionVersionMapper |
||||||
|
.queryProcessDefinitionVersionsPaging(Mockito.any(Page.class), Mockito.eq(processDefinitionId))) |
||||||
|
.thenReturn(new Page<ProcessDefinitionVersion>() |
||||||
|
.setRecords(Lists.newArrayList(processDefinitionVersion))); |
||||||
|
|
||||||
|
Map<String, Object> resultMap4 = processDefinitionVersionService.queryProcessDefinitionVersions( |
||||||
|
loginUser |
||||||
|
, projectName |
||||||
|
, pageNo |
||||||
|
, pageSize |
||||||
|
, processDefinitionId); |
||||||
|
Assert.assertEquals(Status.SUCCESS, resultMap4.get(Constants.STATUS)); |
||||||
|
Assert.assertEquals(processDefinitionVersion |
||||||
|
, ((PageInfo<ProcessDefinitionVersion>) resultMap4.get(Constants.DATA_LIST)) |
||||||
|
.getLists().get(0)); |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testQueryByProcessDefinitionIdAndVersion() { |
||||||
|
|
||||||
|
ProcessDefinitionVersion expectedProcessDefinitionVersion = |
||||||
|
getProcessDefinitionVersion(getProcessDefinition()); |
||||||
|
|
||||||
|
int processDefinitionId = 66; |
||||||
|
long version = 10; |
||||||
|
Mockito.when(processDefinitionVersionMapper.queryByProcessDefinitionIdAndVersion(processDefinitionId, version)) |
||||||
|
.thenReturn(expectedProcessDefinitionVersion); |
||||||
|
|
||||||
|
ProcessDefinitionVersion processDefinitionVersion = processDefinitionVersionService |
||||||
|
.queryByProcessDefinitionIdAndVersion(processDefinitionId, version); |
||||||
|
|
||||||
|
Assert.assertEquals(expectedProcessDefinitionVersion, processDefinitionVersion); |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testDeleteByProcessDefinitionIdAndVersion() { |
||||||
|
String projectName = "project_test1"; |
||||||
|
int processDefinitionId = 66; |
||||||
|
long version = 10; |
||||||
|
Project project = getProject(projectName); |
||||||
|
Mockito.when(projectMapper.queryByName(projectName)) |
||||||
|
.thenReturn(project); |
||||||
|
|
||||||
|
User loginUser = new User(); |
||||||
|
loginUser.setId(-1); |
||||||
|
loginUser.setUserType(UserType.GENERAL_USER); |
||||||
|
|
||||||
|
// project auth fail
|
||||||
|
Mockito.when(projectService.checkProjectAndAuth(loginUser, project, projectName)) |
||||||
|
.thenReturn(new HashMap<>()); |
||||||
|
|
||||||
|
Map<String, Object> resultMap1 = processDefinitionVersionService.deleteByProcessDefinitionIdAndVersion( |
||||||
|
loginUser |
||||||
|
, projectName |
||||||
|
, processDefinitionId |
||||||
|
, version); |
||||||
|
|
||||||
|
Assert.assertEquals(0, resultMap1.size()); |
||||||
|
|
||||||
|
Map<String, Object> res = new HashMap<>(); |
||||||
|
putMsg(res, Status.SUCCESS); |
||||||
|
|
||||||
|
Mockito.when(processDefinitionVersionMapper.deleteByProcessDefinitionIdAndVersion(processDefinitionId, version)) |
||||||
|
.thenReturn(1); |
||||||
|
Mockito.when(projectService.checkProjectAndAuth(loginUser, project, projectName)) |
||||||
|
.thenReturn(res); |
||||||
|
|
||||||
|
Map<String, Object> resultMap2 = processDefinitionVersionService.deleteByProcessDefinitionIdAndVersion( |
||||||
|
loginUser |
||||||
|
, projectName |
||||||
|
, processDefinitionId |
||||||
|
, version); |
||||||
|
|
||||||
|
Assert.assertEquals(Status.SUCCESS, resultMap2.get(Constants.STATUS)); |
||||||
|
|
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* get mock processDefinitionVersion by processDefinition |
||||||
|
* |
||||||
|
* @return processDefinitionVersion |
||||||
|
*/ |
||||||
|
private ProcessDefinitionVersion getProcessDefinitionVersion(ProcessDefinition processDefinition) { |
||||||
|
return ProcessDefinitionVersion |
||||||
|
.newBuilder() |
||||||
|
.processDefinitionId(processDefinition.getId()) |
||||||
|
.version(1) |
||||||
|
.processDefinitionJson(processDefinition.getProcessDefinitionJson()) |
||||||
|
.description(processDefinition.getDescription()) |
||||||
|
.locations(processDefinition.getLocations()) |
||||||
|
.connects(processDefinition.getConnects()) |
||||||
|
.timeout(processDefinition.getTimeout()) |
||||||
|
.globalParams(processDefinition.getGlobalParams()) |
||||||
|
.createTime(processDefinition.getUpdateTime()) |
||||||
|
.receivers(processDefinition.getReceivers()) |
||||||
|
.receiversCc(processDefinition.getReceiversCc()) |
||||||
|
.resourceIds(processDefinition.getResourceIds()) |
||||||
|
.build(); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* get mock processDefinition |
||||||
|
* |
||||||
|
* @return ProcessDefinition |
||||||
|
*/ |
||||||
|
private ProcessDefinition getProcessDefinition() { |
||||||
|
|
||||||
|
ProcessDefinition processDefinition = new ProcessDefinition(); |
||||||
|
processDefinition.setId(66); |
||||||
|
processDefinition.setName("test_pdf"); |
||||||
|
processDefinition.setProjectId(2); |
||||||
|
processDefinition.setTenantId(1); |
||||||
|
processDefinition.setDescription(""); |
||||||
|
|
||||||
|
return processDefinition; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* get mock Project |
||||||
|
* |
||||||
|
* @param projectName projectName |
||||||
|
* @return Project |
||||||
|
*/ |
||||||
|
private Project getProject(String projectName) { |
||||||
|
Project project = new Project(); |
||||||
|
project.setId(1); |
||||||
|
project.setName(projectName); |
||||||
|
project.setUserId(1); |
||||||
|
return project; |
||||||
|
} |
||||||
|
|
||||||
|
private void putMsg(Map<String, Object> result, Status status, Object... statusParams) { |
||||||
|
result.put(Constants.STATUS, status); |
||||||
|
if (statusParams != null && statusParams.length > 0) { |
||||||
|
result.put(Constants.MSG, MessageFormat.format(status.getMsg(), statusParams)); |
||||||
|
} else { |
||||||
|
result.put(Constants.MSG, status.getMsg()); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,23 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.common.enums; |
||||||
|
|
||||||
|
public enum AlertEvent { |
||||||
|
|
||||||
|
SERVER_DOWN,TIME_OUT |
||||||
|
} |
||||||
@ -0,0 +1,23 @@ |
|||||||
|
/* |
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||||
|
* contributor license agreements. See the NOTICE file distributed with |
||||||
|
* this work for additional information regarding copyright ownership. |
||||||
|
* The ASF licenses this file to You 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 org.apache.dolphinscheduler.common.enums; |
||||||
|
|
||||||
|
public enum AlertWarnLevel { |
||||||
|
|
||||||
|
MIDDLE,SERIOUS |
||||||
|
} |
||||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue