Zhou Zheng
4 years ago
261 changed files with 10693 additions and 4237 deletions
@ -0,0 +1,81 @@
|
||||
/* |
||||
* 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.controller; |
||||
|
||||
import org.apache.dolphinscheduler.api.service.WorkFlowLineageService; |
||||
import org.apache.dolphinscheduler.api.utils.Result; |
||||
import org.apache.dolphinscheduler.common.utils.ParameterUtils; |
||||
import io.swagger.annotations.ApiParam; |
||||
import org.apache.dolphinscheduler.dao.entity.WorkFlowLineage; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.http.HttpStatus; |
||||
import org.springframework.web.bind.annotation.*; |
||||
import springfox.documentation.annotations.ApiIgnore; |
||||
|
||||
import java.util.HashSet; |
||||
import java.util.List; |
||||
import java.util.Map; |
||||
import java.util.Set; |
||||
|
||||
import static org.apache.dolphinscheduler.api.enums.Status.QUERY_WORKFLOW_LINEAGE_ERROR; |
||||
|
||||
@RestController |
||||
@RequestMapping("lineages/{projectId}") |
||||
public class WorkFlowLineageController extends BaseController { |
||||
private static final Logger logger = LoggerFactory.getLogger(WorkFlowLineageController.class); |
||||
|
||||
@Autowired |
||||
private WorkFlowLineageService workFlowLineageService; |
||||
|
||||
@GetMapping(value="/list-name") |
||||
@ResponseStatus(HttpStatus.OK) |
||||
public Result<List<WorkFlowLineage>> queryWorkFlowLineageByName(@ApiIgnore @RequestParam(value = "searchVal", required = false) String searchVal, @ApiParam(name = "projectId", value = "PROJECT_ID", required = true) @PathVariable int projectId) { |
||||
try { |
||||
searchVal = ParameterUtils.handleEscapes(searchVal); |
||||
Map<String, Object> result = workFlowLineageService.queryWorkFlowLineageByName(searchVal,projectId); |
||||
return returnDataList(result); |
||||
} catch (Exception e){ |
||||
logger.error(QUERY_WORKFLOW_LINEAGE_ERROR.getMsg(),e); |
||||
return error(QUERY_WORKFLOW_LINEAGE_ERROR.getCode(), QUERY_WORKFLOW_LINEAGE_ERROR.getMsg()); |
||||
} |
||||
} |
||||
|
||||
@GetMapping(value="/list-ids") |
||||
@ResponseStatus(HttpStatus.OK) |
||||
public Result<Map<String, Object>> queryWorkFlowLineageByIds(@ApiIgnore @RequestParam(value = "ids", required = false) String ids,@ApiParam(name = "projectId", value = "PROJECT_ID", required = true) @PathVariable int projectId) { |
||||
|
||||
try { |
||||
ids = ParameterUtils.handleEscapes(ids); |
||||
Set<Integer> idsSet = new HashSet<>(); |
||||
if(ids != null) { |
||||
String[] idsStr = ids.split(","); |
||||
for (String id : idsStr) |
||||
{ |
||||
idsSet.add(Integer.parseInt(id)); |
||||
} |
||||
} |
||||
|
||||
Map<String, Object> result = workFlowLineageService.queryWorkFlowLineageByIds(idsSet, projectId); |
||||
return returnDataList(result); |
||||
} catch (Exception e){ |
||||
logger.error(QUERY_WORKFLOW_LINEAGE_ERROR.getMsg(),e); |
||||
return error(QUERY_WORKFLOW_LINEAGE_ERROR.getCode(), QUERY_WORKFLOW_LINEAGE_ERROR.getMsg()); |
||||
} |
||||
} |
||||
} |
||||
@ -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,97 @@
|
||||
/* |
||||
* 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.common.Constants; |
||||
import org.apache.dolphinscheduler.dao.mapper.WorkFlowLineageMapper; |
||||
import org.apache.dolphinscheduler.dao.entity.WorkFlowLineage; |
||||
import org.apache.dolphinscheduler.dao.entity.WorkFlowRelation; |
||||
import org.springframework.beans.factory.annotation.Autowired; |
||||
import org.springframework.stereotype.Service; |
||||
|
||||
import java.util.*; |
||||
|
||||
@Service |
||||
public class WorkFlowLineageService extends BaseService { |
||||
|
||||
@Autowired |
||||
private WorkFlowLineageMapper workFlowLineageMapper; |
||||
|
||||
public Map<String, Object> queryWorkFlowLineageByName(String workFlowName, int projectId) { |
||||
Map<String, Object> result = new HashMap<>(5); |
||||
List<WorkFlowLineage> workFlowLineageList = workFlowLineageMapper.queryByName(workFlowName, projectId); |
||||
result.put(Constants.DATA_LIST, workFlowLineageList); |
||||
putMsg(result, Status.SUCCESS); |
||||
return result; |
||||
} |
||||
|
||||
private List<WorkFlowRelation> getWorkFlowRelationRecursion(Set<Integer> ids, List<WorkFlowRelation> workFlowRelations,Set<Integer> sourceIds) { |
||||
for(int id : ids) { |
||||
sourceIds.addAll(ids); |
||||
List<WorkFlowRelation> workFlowRelationsTmp = workFlowLineageMapper.querySourceTarget(id); |
||||
if(workFlowRelationsTmp != null && !workFlowRelationsTmp.isEmpty()) { |
||||
Set<Integer> idsTmp = new HashSet<>(); |
||||
for(WorkFlowRelation workFlowRelation:workFlowRelationsTmp) { |
||||
if(!sourceIds.contains(workFlowRelation.getTargetWorkFlowId())){ |
||||
idsTmp.add(workFlowRelation.getTargetWorkFlowId()); |
||||
} |
||||
} |
||||
workFlowRelations.addAll(workFlowRelationsTmp); |
||||
getWorkFlowRelationRecursion(idsTmp, workFlowRelations,sourceIds); |
||||
} |
||||
} |
||||
return workFlowRelations; |
||||
} |
||||
|
||||
public Map<String, Object> queryWorkFlowLineageByIds(Set<Integer> ids,int projectId) { |
||||
Map<String, Object> result = new HashMap<>(5); |
||||
List<WorkFlowLineage> workFlowLineageList = workFlowLineageMapper.queryByIds(ids, projectId); |
||||
Map<String, Object> workFlowLists = new HashMap<>(5); |
||||
Set<Integer> idsV = new HashSet<>(); |
||||
if(ids == null || ids.isEmpty()){ |
||||
for(WorkFlowLineage workFlowLineage:workFlowLineageList) { |
||||
idsV.add(workFlowLineage.getWorkFlowId()); |
||||
} |
||||
} else { |
||||
idsV = ids; |
||||
} |
||||
List<WorkFlowRelation> workFlowRelations = new ArrayList<>(); |
||||
Set<Integer> sourceIds = new HashSet<>(); |
||||
getWorkFlowRelationRecursion(idsV, workFlowRelations, sourceIds); |
||||
|
||||
Set<Integer> idSet = new HashSet<>(); |
||||
//If the incoming parameter is not empty, you need to add downstream workflow detail attributes
|
||||
if(ids != null && !ids.isEmpty()) { |
||||
for(WorkFlowRelation workFlowRelation : workFlowRelations) { |
||||
idSet.add(workFlowRelation.getTargetWorkFlowId()); |
||||
} |
||||
for(int id : ids){ |
||||
idSet.remove(id); |
||||
} |
||||
if(!idSet.isEmpty()) { |
||||
workFlowLineageList.addAll(workFlowLineageMapper.queryByIds(idSet, projectId)); |
||||
} |
||||
} |
||||
|
||||
workFlowLists.put("workFlowList",workFlowLineageList); |
||||
workFlowLists.put("workFlowRelationList",workFlowRelations); |
||||
result.put(Constants.DATA_LIST, workFlowLists); |
||||
putMsg(result, Status.SUCCESS); |
||||
return result; |
||||
} |
||||
} |
||||
@ -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 + String.valueOf(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,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,69 @@
|
||||
/* |
||||
* 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.controller; |
||||
|
||||
import org.apache.dolphinscheduler.api.enums.Status; |
||||
import org.apache.dolphinscheduler.api.utils.Result; |
||||
import org.apache.dolphinscheduler.common.utils.JSONUtils; |
||||
import org.junit.Assert; |
||||
import org.junit.Test; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
import org.springframework.http.MediaType; |
||||
import org.springframework.test.web.servlet.MvcResult; |
||||
import org.springframework.util.LinkedMultiValueMap; |
||||
import org.springframework.util.MultiValueMap; |
||||
|
||||
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; |
||||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content; |
||||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
||||
|
||||
public class WorkFlowLineageControllerTest extends AbstractControllerTest { |
||||
private static Logger logger = LoggerFactory.getLogger(WorkFlowLineageControllerTest.class); |
||||
|
||||
@Test |
||||
public void testQueryWorkFlowLineageByName() throws Exception { |
||||
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>(); |
||||
paramsMap.add("searchVal","test"); |
||||
MvcResult mvcResult = mockMvc.perform(get("/lineages/1/list-name") |
||||
.header("sessionId", sessionId) |
||||
.params(paramsMap)) |
||||
.andExpect(status().isOk()) |
||||
.andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8)) |
||||
.andReturn(); |
||||
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class); |
||||
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue()); |
||||
logger.info(mvcResult.getResponse().getContentAsString()); |
||||
|
||||
} |
||||
|
||||
@Test |
||||
public void testQueryWorkFlowLineageByIds() throws Exception { |
||||
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>(); |
||||
paramsMap.add("ids","1"); |
||||
MvcResult mvcResult = mockMvc.perform(get("/lineages/1/list-ids") |
||||
.header("sessionId", sessionId) |
||||
.params(paramsMap)) |
||||
.andExpect(status().isOk()) |
||||
.andExpect(content().contentType(MediaType.APPLICATION_JSON_UTF8)) |
||||
.andReturn(); |
||||
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class); |
||||
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue()); |
||||
logger.info(mvcResult.getResponse().getContentAsString()); |
||||
} |
||||
|
||||
} |
||||
@ -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,88 @@
|
||||
/* |
||||
* 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.Constants; |
||||
import org.apache.dolphinscheduler.common.utils.EncryptionUtils; |
||||
import org.apache.dolphinscheduler.dao.entity.WorkFlowLineage; |
||||
import org.apache.dolphinscheduler.dao.entity.WorkFlowRelation; |
||||
import org.apache.dolphinscheduler.dao.mapper.WorkFlowLineageMapper; |
||||
import org.junit.Assert; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.mockito.InjectMocks; |
||||
import org.mockito.Mock; |
||||
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
||||
import java.util.*; |
||||
|
||||
import static org.mockito.Mockito.when; |
||||
|
||||
@RunWith(MockitoJUnitRunner.class) |
||||
public class WorkFlowLineageServiceTest { |
||||
|
||||
@InjectMocks |
||||
private WorkFlowLineageService workFlowLineageService; |
||||
|
||||
@Mock |
||||
private WorkFlowLineageMapper workFlowLineageMapper; |
||||
|
||||
@Test |
||||
public void testQueryWorkFlowLineageByName() { |
||||
String searchVal = "test"; |
||||
when(workFlowLineageMapper.queryByName(searchVal, 1)).thenReturn(getWorkFlowLineages()); |
||||
Map<String, Object> result = workFlowLineageService.queryWorkFlowLineageByName(searchVal,1); |
||||
List<WorkFlowLineage> workFlowLineageList = (List<WorkFlowLineage>)result.get(Constants.DATA_LIST); |
||||
Assert.assertTrue(workFlowLineageList.size()>0); |
||||
} |
||||
|
||||
@Test |
||||
public void testQueryWorkFlowLineageByIds() { |
||||
|
||||
Set<Integer> ids = new HashSet<>(); |
||||
ids.add(1); |
||||
ids.add(2); |
||||
|
||||
when(workFlowLineageMapper.queryByIds(ids, 1)).thenReturn(getWorkFlowLineages()); |
||||
when(workFlowLineageMapper.querySourceTarget(1)).thenReturn(getWorkFlowRelation()); |
||||
Map<String, Object> result = workFlowLineageService.queryWorkFlowLineageByIds(ids,1); |
||||
Map<String, Object> workFlowLists = (Map<String, Object>)result.get(Constants.DATA_LIST); |
||||
List<WorkFlowLineage> workFlowLineages = (List<WorkFlowLineage>)workFlowLists.get("workFlowList"); |
||||
List<WorkFlowRelation> workFlowRelations = (List<WorkFlowRelation>)workFlowLists.get("workFlowRelationList"); |
||||
Assert.assertTrue(workFlowLineages.size()>0); |
||||
Assert.assertTrue(workFlowRelations.size()>0); |
||||
} |
||||
|
||||
private List<WorkFlowLineage> getWorkFlowLineages() { |
||||
List<WorkFlowLineage> workFlowLineages = new ArrayList<>(); |
||||
WorkFlowLineage workFlowLineage = new WorkFlowLineage(); |
||||
workFlowLineage.setWorkFlowId(1); |
||||
workFlowLineage.setWorkFlowName("testdag"); |
||||
workFlowLineages.add(workFlowLineage); |
||||
return workFlowLineages; |
||||
} |
||||
|
||||
private List<WorkFlowRelation> getWorkFlowRelation(){ |
||||
List<WorkFlowRelation> workFlowRelations = new ArrayList<>(); |
||||
WorkFlowRelation workFlowRelation = new WorkFlowRelation(); |
||||
workFlowRelation.setSourceWorkFlowId(1); |
||||
workFlowRelation.setTargetWorkFlowId(2); |
||||
workFlowRelations.add(workFlowRelation); |
||||
return workFlowRelations; |
||||
} |
||||
|
||||
} |
||||
File diff suppressed because one or more lines are too long
@ -0,0 +1,156 @@
|
||||
/* |
||||
* 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.utils; |
||||
|
||||
import org.apache.dolphinscheduler.common.Constants; |
||||
import org.apache.http.auth.AuthSchemeProvider; |
||||
import org.apache.http.auth.AuthScope; |
||||
import org.apache.http.auth.Credentials; |
||||
import org.apache.http.client.config.AuthSchemes; |
||||
import org.apache.http.client.methods.HttpGet; |
||||
import org.apache.http.config.Lookup; |
||||
import org.apache.http.config.RegistryBuilder; |
||||
import org.apache.http.impl.auth.SPNegoSchemeFactory; |
||||
import org.apache.http.impl.client.BasicCredentialsProvider; |
||||
import org.apache.http.impl.client.CloseableHttpClient; |
||||
import org.apache.http.impl.client.HttpClientBuilder; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
import javax.security.auth.Subject; |
||||
import javax.security.auth.kerberos.KerberosPrincipal; |
||||
import javax.security.auth.login.AppConfigurationEntry; |
||||
import javax.security.auth.login.Configuration; |
||||
import javax.security.auth.login.LoginContext; |
||||
import javax.security.auth.login.LoginException; |
||||
import java.security.Principal; |
||||
import java.security.PrivilegedAction; |
||||
import java.util.HashMap; |
||||
import java.util.HashSet; |
||||
import java.util.Map; |
||||
import java.util.Set; |
||||
|
||||
/** |
||||
* kerberos http client |
||||
*/ |
||||
public class KerberosHttpClient { |
||||
|
||||
public static final Logger logger = LoggerFactory.getLogger(KerberosHttpClient.class); |
||||
|
||||
private String principal; |
||||
private String keyTabLocation; |
||||
|
||||
public KerberosHttpClient(String principal, String keyTabLocation) { |
||||
super(); |
||||
this.principal = principal; |
||||
this.keyTabLocation = keyTabLocation; |
||||
} |
||||
|
||||
public KerberosHttpClient(String principal, String keyTabLocation, boolean isDebug) { |
||||
this(principal, keyTabLocation); |
||||
if (isDebug) { |
||||
System.setProperty("sun.security.spnego.debug", "true"); |
||||
System.setProperty("sun.security.krb5.debug", "true"); |
||||
} |
||||
} |
||||
|
||||
public KerberosHttpClient(String principal, String keyTabLocation, String krb5Location, boolean isDebug) { |
||||
this(principal, keyTabLocation, isDebug); |
||||
System.setProperty("java.security.krb5.conf", krb5Location); |
||||
} |
||||
|
||||
private static CloseableHttpClient buildSpengoHttpClient() { |
||||
HttpClientBuilder builder = HttpClientBuilder.create(); |
||||
Lookup<AuthSchemeProvider> authSchemeRegistry = RegistryBuilder.<AuthSchemeProvider>create() |
||||
.register(AuthSchemes.SPNEGO, new SPNegoSchemeFactory(true)).build(); |
||||
builder.setDefaultAuthSchemeRegistry(authSchemeRegistry); |
||||
BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider(); |
||||
credentialsProvider.setCredentials(new AuthScope(null, -1, null), new Credentials() { |
||||
@Override |
||||
public Principal getUserPrincipal() { |
||||
return null; |
||||
} |
||||
|
||||
@Override |
||||
public String getPassword() { |
||||
return null; |
||||
} |
||||
}); |
||||
builder.setDefaultCredentialsProvider(credentialsProvider); |
||||
return builder.build(); |
||||
} |
||||
|
||||
public String get(final String url, final String userId) { |
||||
logger.info("Calling KerberosHttpClient {} {} {}", this.principal, this.keyTabLocation, url); |
||||
Configuration config = new Configuration() { |
||||
@SuppressWarnings("serial") |
||||
@Override |
||||
public AppConfigurationEntry[] getAppConfigurationEntry(String name) { |
||||
Map<String, Object> options = new HashMap<>(9); |
||||
options.put("useTicketCache", "false"); |
||||
options.put("useKeyTab", "true"); |
||||
options.put("keyTab", keyTabLocation); |
||||
options.put("refreshKrb5Config", "true"); |
||||
options.put("principal", principal); |
||||
options.put("storeKey", "true"); |
||||
options.put("doNotPrompt", "true"); |
||||
options.put("isInitiator", "true"); |
||||
options.put("debug", "true"); |
||||
return new AppConfigurationEntry[] { |
||||
new AppConfigurationEntry("com.sun.security.auth.module.Krb5LoginModule", |
||||
AppConfigurationEntry.LoginModuleControlFlag.REQUIRED, options) }; |
||||
} |
||||
}; |
||||
Set<Principal> princ = new HashSet<>(1); |
||||
princ.add(new KerberosPrincipal(userId)); |
||||
Subject sub = new Subject(false, princ, new HashSet<>(), new HashSet<>()); |
||||
|
||||
LoginContext lc; |
||||
try { |
||||
lc = new LoginContext("", sub, null, config); |
||||
lc.login(); |
||||
Subject serviceSubject = lc.getSubject(); |
||||
return Subject.doAs(serviceSubject, (PrivilegedAction<String>) () -> { |
||||
CloseableHttpClient httpClient = buildSpengoHttpClient(); |
||||
HttpGet httpget = new HttpGet(url); |
||||
return HttpUtils.getResponseContentString(httpget, httpClient); |
||||
}); |
||||
} catch (LoginException le) { |
||||
logger.error("Kerberos authentication failed ", le); |
||||
} |
||||
return null; |
||||
} |
||||
|
||||
/** |
||||
* get http request content by kerberosClient |
||||
* |
||||
* @param url url |
||||
* @return http get request response content |
||||
*/ |
||||
public static String get(String url) { |
||||
|
||||
String responseContent; |
||||
KerberosHttpClient kerberosHttpClient = new KerberosHttpClient( |
||||
PropertyUtils.getString(Constants.LOGIN_USER_KEY_TAB_USERNAME), |
||||
PropertyUtils.getString(Constants.LOGIN_USER_KEY_TAB_PATH), |
||||
PropertyUtils.getString(Constants.JAVA_SECURITY_KRB5_CONF_PATH), true); |
||||
responseContent = kerberosHttpClient.get(url, PropertyUtils.getString(Constants.LOGIN_USER_KEY_TAB_USERNAME)); |
||||
return responseContent; |
||||
|
||||
} |
||||
|
||||
} |
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue