Skip to content

J
jz-dmp-service
Commit cd4b5211 authored by mcb's avatar mcb
Browse files
Options

数据运维修改

parent 5682aa10
Hide whitespace changes
Inline Side-by-side
Showing with 490 additions and 406 deletions
src/main/java/com/jz/dmp/azkaban/dao/ExecutionFlowsMapper.java
View file @ cd4b5211
......@@ -37,4 +37,8 @@ public interface ExecutionFlowsMapper {
* @since 2021-02-03
*/
List<Map> queryExamplesLogByExecId(String execId) throws Exception;
Map<String, Object> queryTaskInstanceStatus() throws Exception;
List<Map> queryLastStatus(@Param("taskName") String[] taskName) throws Exception;
}
src/main/java/com/jz/dmp/modules/controller/dataOperation/DmpDevTaskController.java
View file @ cd4b5211
......@@ -3,6 +3,7 @@ package com.jz.dmp.modules.controller.dataOperation;
import com.jz.common.constant.JsonResult;
import com.jz.common.constant.ResultCode;
import com.jz.common.page.PageInfoResponse;
import com.jz.dmp.modules.controller.DataIntegration.bean.ConflictCheckReq;
import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListDto;
import com.jz.dmp.modules.controller.DataIntegration.bean.RealTimeSyncListReq;
import com.jz.dmp.modules.controller.dataOperation.bean.DataDevTaskListDto;
......@@ -74,4 +75,28 @@ public class DmpDevTaskController {
JsonResult list = dmpDevelopTaskService.runTaskByTaskId(taskId);
return list;
}
/**
* 运维大屏--获取任务状态
*
* @return
* @author Bellamy
* @since 2021-02-22
*/
@ApiOperation(value = "运维大屏--重点关注任务状态", notes = "重点关注")
@PostMapping(value = "/getTaskStatus")
public JsonResult getTaskStatus(@RequestParam String projectId) throws Exception {
if (StringUtils.isEmpty(projectId)) {
return new JsonResult(ResultCode.PARAMS_ERROR, "projectId不能为空！");
}
JsonResult result = new JsonResult();
try {
result = dmpDevelopTaskService.getTaskStatus(projectId);
} catch (Exception e) {
result.setMessage(e.getMessage());
result.setCode(ResultCode.INTERNAL_SERVER_ERROR);
e.printStackTrace();
}
return result;
}
}
src/main/java/com/jz/dmp/modules/controller/dataOperation/bean/DataDevTaskListReq.java
View file @ cd4b5211
......@@ -44,6 +44,9 @@ public class DataDevTaskListReq extends BasePageBean {
@ApiModelProperty(value = "任务类型")
private String taskType;
@ApiModelProperty(value = "最后执行状态")
private String status;
public String getProjectId() {
return projectId;
}
......@@ -75,4 +78,12 @@ public class DataDevTaskListReq extends BasePageBean {
public void setTaskType(String taskType) {
this.taskType = taskType;
}
public String getStatus() {
return status;
}
public void setStatus(String status) {
this.status = status;
}
}
src/main/java/com/jz/dmp/modules/controller/dataService/bean/AuthApiListReq.java
View file @ cd4b5211
......@@ -18,12 +18,16 @@ import java.io.Serializable;
@ApiModel("授权api列表请求对象")
public class AuthApiListReq extends BasePageBean implements Serializable {
@ApiModelProperty(value = "文件夹id", required = false)
@ApiModelProperty(value = "文件夹id",required = false)
private Long fileId;
@ApiModelProperty(value = "组织编码", required = false)
@ApiModelProperty(value = "文件来源:,1 API制做,2 组织创建",required = false)
private String fileSource;
@ApiModelProperty(value = "组织编码",required = false)
private String orgCode;
@ApiModelProperty(value = "组织名称", required = false)
@ApiModelProperty(value = "组织名称",required = false)
private String orgName;
@ApiModelProperty(value = "api名称", required = false)
@ApiModelProperty(value = "api名称",required = false)
private String apiName;
@ApiModelProperty(value = "项目id")
private Long projectId;
}
src/main/java/com/jz/dmp/modules/controller/dataService/bean/OrganizationManageListQueryReq.java
View file @ cd4b5211
......@@ -30,4 +30,7 @@ public class OrganizationManageListQueryReq extends BasePageBean implements Seri
@ApiModelProperty(value = "文件夹id")
private String fileId;
@ApiModelProperty(value = "工程id")
private Long projectId;
}
src/main/java/com/jz/dmp/modules/dao/DmpRealtimeSyncInfoDao.java
View file @ cd4b5211
......@@ -195,4 +195,13 @@ public interface DmpRealtimeSyncInfoDao {
* @since 2021-02-02
*/
int insertRealtimeHistory(DmpRealtimeTaskHistory taskHistory) throws Exception;
/**
* 获取实时任务状态
*
* @return
* @author Bellamy
* @since 2021-02-02
*/
Map<String, Object> queryTaskStatus(String projectId) throws Exception;
}
\ No newline at end of file
src/main/java/com/jz/dmp/modules/service/DmpDevelopTaskService.java
View file @ cd4b5211
......@@ -174,4 +174,13 @@ public interface DmpDevelopTaskService {
* @since 2021-02-03
*/
JsonResult queryExamplesLogByExecId(String execId) throws Exception;
/**
* 运维大屏--获取任务状态
*
* @return
* @author Bellamy
* @since 2021-02-22
*/
JsonResult getTaskStatus(String projectId) throws Exception;
}
\ No newline at end of file
src/main/java/com/jz/dmp/modules/service/impl/DmpDevelopTaskServiceImpl.java
View file @ cd4b5211
package com.jz.dmp.modules.service.impl;
import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.regex.Pattern;
import javax.servlet.http.HttpServletRequest;
import com.alibaba.fastjson.JSONObject;
import com.jz.common.utils.web.HttpClientUtils;
import com.mysql.jdbc.Blob;
import org.apache.tomcat.jni.Mmap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.ThrowsAdvice;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.github.pagehelper.Page;
......@@ -41,15 +15,7 @@ import com.jz.common.constant.StatuConstant;
import com.jz.common.enums.ModuleLogEnum;
import com.jz.common.page.PageInfoResponse;
import com.jz.common.persistence.BaseService;
import com.jz.common.utils.AzkabanApiUtils2;
import com.jz.common.utils.CodeGeneratorUtils;
import com.jz.common.utils.DateUtils;
import com.jz.common.utils.FileUtils;
import com.jz.common.utils.FlowParseTool;
import com.jz.common.utils.GZIPUtils;
import com.jz.common.utils.JsonMapper;
import com.jz.common.utils.StringUtils;
import com.jz.common.utils.ZipUtils;
import com.jz.common.utils.*;
import com.jz.common.utils.web.XmlUtils;
import com.jz.dmp.agent.DmpAgentResult;
import com.jz.dmp.azkaban.dao.ExecutionFlowsMapper;
......@@ -63,27 +29,25 @@ import com.jz.dmp.modules.controller.dataOperation.bean.DataDevExamplesListReq;
import com.jz.dmp.modules.controller.dataOperation.bean.DataDevTaskListDto;
import com.jz.dmp.modules.controller.dataOperation.bean.DataDevTaskListReq;
import com.jz.dmp.modules.controller.projconfig.bean.DmpProjectConfigInfoDto;
import com.jz.dmp.modules.dao.DmpDevelopTaskDao;
import com.jz.dmp.modules.dao.DmpNavigationTreeDao;
import com.jz.dmp.modules.dao.DmpProjectDao;
import com.jz.dmp.modules.dao.DmpSyncingDatasourceTypeDao;
import com.jz.dmp.modules.dao.*;
import com.jz.dmp.modules.dao.projconfig.DmpProjectConfigInfoMapper;
import com.jz.dmp.modules.model.DmpAgentDatasourceInfo;
import com.jz.dmp.modules.model.DmpDevelopTask;
import com.jz.dmp.modules.model.DmpDevelopTaskHistory;
import com.jz.dmp.modules.model.DmpModuleOperateLog;
import com.jz.dmp.modules.model.DmpNavigationTree;
import com.jz.dmp.modules.model.DmpProjectConfigInfo;
import com.jz.dmp.modules.model.DmpProjectSystemInfo;
import com.jz.dmp.modules.model.DmpSyncingDatasource;
import com.jz.dmp.modules.model.DmpSyncingDatasourceType;
import com.jz.dmp.modules.service.DmpDevelopTaskHistoryService;
import com.jz.dmp.modules.service.DmpDevelopTaskService;
import com.jz.dmp.modules.service.DmpModuleOperateLogService;
import com.jz.dmp.modules.service.DmpSyncingDatasourceService;
import com.jz.dmp.modules.service.FlowService;
import com.jz.dmp.modules.service.OfflineSynchService;
import com.jz.dmp.modules.service.projconfig.DmpProjectConfigInfoService;
import com.jz.dmp.modules.model.*;
import com.jz.dmp.modules.service.*;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import javax.servlet.http.HttpServletRequest;
import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.regex.Pattern;
/**
* 任务开发(DmpDevelopTask)表服务实现类
......@@ -117,28 +81,31 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
@Autowired
private ExecutionFlowsMapper executionFlowsMapper;
@Autowired
private DmpModuleOperateLogService dmpModuleOperateLogService;
@Autowired
private DmpNavigationTreeDao dmpNavigationTreeDao;
@Autowired
private ObjectMapper objectMapper;
@Autowired
private FlowService flowService;
@Autowired
private DmpDevelopTaskHistoryService dmpDevelopTaskHistoryService;
@Autowired
private DmpProjectConfigInfoMapper dmpProjectConfigInfoMapper;
@Autowired
private RedisTemplate redisTemplate;
@Autowired
private DmpRealtimeSyncInfoDao dmpRealtimeSyncInfoDao;
/**
* 添加保存dmp数据(包含校验数据)
*
......@@ -811,6 +778,7 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
public PageInfoResponse<DataDevTaskListDto> queryDevTaskListPage(DataDevTaskListReq req) throws Exception {
PageInfoResponse<DataDevTaskListDto> pageInfoResponse = new PageInfoResponse<>();
if (StringUtils.isNotBlank(req.getTreeIdOrName())) {
//判断是否为整数 是整数返回true,否则返回false
Pattern pattern = Pattern.compile("^[-\\+]?[\\d]*$");
......@@ -829,6 +797,13 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
pageInfoResponse.setCode(ResultCode.SUCCESS);
pageInfoResponse.setMessage("查询成功");
pageInfoResponse.setData(pageInfo);
String taskName = "";
List<DataDevTaskListDto> listObj = (List<DataDevTaskListDto>) pageInfo;
for (DataDevTaskListDto str : listObj) {
taskName += "," + str.getTaskName();
}
List<Map> flowList = executionFlowsMapper.queryLastStatus(taskName.substring(1).split(","));
return pageInfoResponse;
}
......@@ -983,17 +958,17 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
}
//未删除数据
param.put("dataStatus", "1");
String obderByStr = dmpDevelopTaskRequest.takeOrderByStr(DmpDevelopTaskRequest.class);
Page page = null;
if (obderByStr==null) {
page = PageHelper.startPage(dmpDevelopTaskRequest.getPageNum(), dmpDevelopTaskRequest.getPageSize());
} else {
page = PageHelper.startPage(dmpDevelopTaskRequest.getPageNum(), dmpDevelopTaskRequest.getPageSize(), obderByStr);
}
if (obderByStr == null) {
page = PageHelper.startPage(dmpDevelopTaskRequest.getPageNum(), dmpDevelopTaskRequest.getPageSize());
} else {
page = PageHelper.startPage(dmpDevelopTaskRequest.getPageNum(), dmpDevelopTaskRequest.getPageSize(), obderByStr);
}
List<DmpDevelopTask> list = dmpDevelopTaskDao.findList(param);
List<DmpDevelopTaskDto> dtos = MyDmpDevelopTaskConverter.INSTANCE().domain2dto(list);
PageInfo<DmpDevelopTaskDto> pageInfo = new PageInfo<>(dtos);
......@@ -1047,10 +1022,10 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
//通过任务名称，去查询开发实例
PageHelper.startPage(req.getPageNum(), req.getPageSize());
Map saveParams = new HashMap();
if (org.apache.commons.lang3.StringUtils.isNotEmpty(req.getBusinessTime())) { //业务时间范围
String[] cretime = req.getBusinessTime().split("-");
saveParams.put("startTime", cretime[0].trim() + " 00:00:00");
saveParams.put("endTime", cretime[1].trim() + " 23:59:59");
if (StringUtils.isNotBlank(req.getStartTime()) && StringUtils.isNotBlank(req.getEndTime())) { //业务时间范围
//String[] cretime = req.getBusinessTime().split("-");
saveParams.put("startTime", req.getStartTime().trim() + " 00:00:00");
saveParams.put("endTime", req.getEndTime().trim() + " 23:59:59");
} else if (StringUtils.isNotBlank(req.getCreTimeType())) {
if ("01".equals(req.getCreTimeType())) { //昨天
saveParams.put("businessTime", DateUtils.getYesterdayStr());
......@@ -1086,9 +1061,9 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
}
return pageInfoResponse;
}
/**
*根据treeId取得最新版提交的同步脚本文件名（不含后缀）及版本信息
* 根据treeId取得最新版提交的同步脚本文件名（不含后缀）及版本信息
*/
@Override
public String getConfigFileNameNotSuffix4Published(Long treeId) throws Exception {
......@@ -1096,8 +1071,8 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
String fileName = XmlUtils.getPropertyValue(log.getOperateLog(), "name");
return fileName + "@" + log.getRemark();
}
private DmpModuleOperateLog getLastVersion4SubmitSyncInfo(Integer treeId)throws Exception {
private DmpModuleOperateLog getLastVersion4SubmitSyncInfo(Integer treeId) throws Exception {
DmpModuleOperateLog o = null;
try {
o = dmpModuleOperateLogService.getLastOperateLog(treeId, ModuleLogEnum.VERSION_SYNC_XML);
......@@ -1118,48 +1093,48 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
}
return o;
}
/*
* (non-Javadoc)
*
* @see
* com.ycxc.vmts.service.DmpDevelopTaskService#add(com.ycxc.vmts.entity.DmpDevelopTask,
* javax.servlet.http.HttpServletRequest)
*/
@Override
@Transactional(rollbackFor = Exception.class)
public BaseBeanResponse<DmpDevelopTask> add(DmpDevelopTask dmpDevelopTask, HttpServletRequest httpRequest) throws Exception {
BaseBeanResponse<DmpDevelopTask> baseBeanResponse = new BaseBeanResponse<>();
//参数校验
String taskName = dmpDevelopTask.getName();
if (StringUtils.isEmpty(taskName)) {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseBeanResponse.setMessage("任务名称不能为空");
return baseBeanResponse;
}
//任务类型校验
String taskType = dmpDevelopTask.getTaskType();
String treeType = "";
if (StringUtils.isEmpty(taskType)) {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseBeanResponse.setMessage("任务类型不能为空");
return baseBeanResponse;
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVELOP)) {
treeType = CommConstant.TREE_TYPE_DEVELOP;
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSHELL)) {
treeType = CommConstant.TREE_TYPE_DEVSHELL;
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSQL)) {
treeType = CommConstant.TREE_TYPE_DEVSQL;
} else {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseBeanResponse.setMessage("任务类型错误");
return baseBeanResponse;
}
//添加树叶子节点
DmpNavigationTree tree = new DmpNavigationTree();
* (non-Javadoc)
*
* @see
* com.ycxc.vmts.service.DmpDevelopTaskService#add(com.ycxc.vmts.entity.DmpDevelopTask,
* javax.servlet.http.HttpServletRequest)
*/
@Override
@Transactional(rollbackFor = Exception.class)
public BaseBeanResponse<DmpDevelopTask> add(DmpDevelopTask dmpDevelopTask, HttpServletRequest httpRequest) throws Exception {
BaseBeanResponse<DmpDevelopTask> baseBeanResponse = new BaseBeanResponse<>();
//参数校验
String taskName = dmpDevelopTask.getName();
if (StringUtils.isEmpty(taskName)) {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseBeanResponse.setMessage("任务名称不能为空");
return baseBeanResponse;
}
//任务类型校验
String taskType = dmpDevelopTask.getTaskType();
String treeType = "";
if (StringUtils.isEmpty(taskType)) {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseBeanResponse.setMessage("任务类型不能为空");
return baseBeanResponse;
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVELOP)) {
treeType = CommConstant.TREE_TYPE_DEVELOP;
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSHELL)) {
treeType = CommConstant.TREE_TYPE_DEVSHELL;
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSQL)) {
treeType = CommConstant.TREE_TYPE_DEVSQL;
} else {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseBeanResponse.setMessage("任务类型错误");
return baseBeanResponse;
}
//添加树叶子节点
DmpNavigationTree tree = new DmpNavigationTree();
tree.setName(taskName);
tree.setParentId(dmpDevelopTask.getParentId());
tree.setProjectId(dmpDevelopTask.getProjectId());
......@@ -1169,15 +1144,15 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
tree.setIsEnable("1");
tree.setDataStatus("1");
tree.setTreeSort(dmpNavigationTreeDao.getMaxSortById(tree));
int cnt = dmpNavigationTreeDao.countTreeByName(tree);
if (cnt > 0) {
baseBeanResponse.setCode(StatuConstant.CODE_DATA_EXISTED);
baseBeanResponse.setMessage("当前项目已存在同名的任务名称");
return baseBeanResponse;
baseBeanResponse.setCode(StatuConstant.CODE_DATA_EXISTED);
baseBeanResponse.setMessage("当前项目已存在同名的任务名称");
return baseBeanResponse;
}
dmpNavigationTreeDao.insert(tree);
//将新增任务保存到数据库
dmpDevelopTask.setTaskType(taskType);
dmpDevelopTask.setType("W");
......@@ -1188,7 +1163,7 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
int revision = 1;
ObjectNode flowHeaderNode = objectMapper.createObjectNode();
flowHeaderNode.put("name", dmpDevelopTask.getName() );
flowHeaderNode.put("name", dmpDevelopTask.getName());
flowHeaderNode.put("description", dmpDevelopTask.getTaskDesc());
flowHeaderNode.put("revision", revision);
dmpDevelopTask.setFlowHeader(flowHeaderNode.toString());
......@@ -1197,7 +1172,7 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
scriptNode.put("id", "canvas");
scriptNode.put("resourceId", "canvas");
ObjectNode stencilSetNode = objectMapper.createObjectNode();
stencilSetNode.put("namespace","http://b3mn.org/stencilset/bpmn2.0#");
stencilSetNode.put("namespace", "http://b3mn.org/stencilset/bpmn2.0#");
scriptNode.put("stencilset", stencilSetNode);
byte[] data = null;
try {
......@@ -1207,105 +1182,105 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
}
dmpDevelopTask.setData(data);
// 创建人
dmpDevelopTask.setCreateUserId(Optional.ofNullable(getHttpRequestUserId(httpRequest)).orElse(0).toString());
// 创建时间
dmpDevelopTask.setCreateTime(new Date());
//版本设置
//获取已存在版本
String version = CodeGeneratorUtils.generatorNextTaskVesion(null);
dmpDevelopTask.setVersion(version);
dmpDevelopTaskDao.insert(dmpDevelopTask);
baseBeanResponse.setCode(StatuConstant.SUCCESS_CODE);
baseBeanResponse.setMessage("新增成功");
baseBeanResponse.setData(dmpDevelopTask);
return baseBeanResponse;
}
/*
* (non-Javadoc)
*
* @see
* com.ycxc.vmts.service.DmpDevelopTaskService#findByTreeId(com.ycxc.vmts.controller.
* bean.DmpDevelopTaskRequest, javax.servlet.http.HttpServletRequest)
*/
@Override
public BaseBeanResponse<DmpDevelopTaskDto> findByTreeId(Integer treeId, HttpServletRequest httpRequest)
throws Exception {
BaseBeanResponse<DmpDevelopTaskDto> baseBeanResponse = new BaseBeanResponse<>();
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.selectTaskInfoByParam(treeId);
DmpDevelopTaskDto dmpDevelopTaskDto = MyDmpDevelopTaskConverter.INSTANCE().domain2dto(dmpDevelopTask);
baseBeanResponse.setCode(StatuConstant.SUCCESS_CODE);
baseBeanResponse.setMessage("查询成功");
baseBeanResponse.setData(dmpDevelopTaskDto);
return baseBeanResponse;
}
/**
*任务修改
*/
@Override
@Transactional(rollbackFor = Exception.class)
public BaseBeanResponse<DmpDevelopTask> edit(DmpDevelopTask dmpDevelopTask, HttpServletRequest httpRequest)
throws Exception {
BaseBeanResponse<DmpDevelopTask> baseBeanResponse = new BaseBeanResponse<DmpDevelopTask>();
// 修改人
dmpDevelopTask.setUpdateUserId(getHttpRequestUserId(httpRequest).toString());
// 修改时间
dmpDevelopTask.setUpdateTime(new Date());
// 版本设置
// 获取已存在版本
DmpDevelopTask dmpDevelopTaskDb = dmpDevelopTaskDao.selectTaskById(dmpDevelopTask.getId().toString());
String version = CodeGeneratorUtils.generatorNextTaskVesion(dmpDevelopTaskDb.getVersion());
dmpDevelopTask.setVersion(version);
dmpDevelopTaskDao.update(dmpDevelopTask);
DmpDevelopTask dmpDevelopTaskDb2 = dmpDevelopTaskDao.selectTaskById(dmpDevelopTask.getId().toString());
DmpDevelopTaskHistory dmpDevelopTaskHistory = MyDmpDevelopTaskHistoryConverter.INSTANCE().task2history(dmpDevelopTaskDb2);
// 保存版本为版本数据
dmpDevelopTaskHistoryService.add(dmpDevelopTaskHistory, httpRequest);
baseBeanResponse.setCode(StatuConstant.SUCCESS_CODE);
baseBeanResponse.setMessage("修改成功");
baseBeanResponse.setData(dmpDevelopTaskDb2);
return baseBeanResponse;
}
/**
*任务流程发布
*/
@Override
public BaseResponse flowSubmit(Long treeId, HttpServletRequest httpRequest)
throws Exception {
BaseResponse baseResponse = new BaseResponse();
// 参数校验
if (treeId == null) {
baseResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseResponse.setMessage("treeId不能为空");
return baseResponse;
}
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.get(treeId);
FlowPro flowPro = MyDmpDevelopTaskConverter.INSTANCE().task2flowpro(dmpDevelopTask);
// 创建人
dmpDevelopTask.setCreateUserId(Optional.ofNullable(getHttpRequestUserId(httpRequest)).orElse(0).toString());
// 创建时间
dmpDevelopTask.setCreateTime(new Date());
//版本设置
//获取已存在版本
String version = CodeGeneratorUtils.generatorNextTaskVesion(null);
dmpDevelopTask.setVersion(version);
dmpDevelopTaskDao.insert(dmpDevelopTask);
baseBeanResponse.setCode(StatuConstant.SUCCESS_CODE);
baseBeanResponse.setMessage("新增成功");
baseBeanResponse.setData(dmpDevelopTask);
return baseBeanResponse;
}
/*
* (non-Javadoc)
*
* @see
* com.ycxc.vmts.service.DmpDevelopTaskService#findByTreeId(com.ycxc.vmts.controller.
* bean.DmpDevelopTaskRequest, javax.servlet.http.HttpServletRequest)
*/
@Override
public BaseBeanResponse<DmpDevelopTaskDto> findByTreeId(Integer treeId, HttpServletRequest httpRequest)
throws Exception {
BaseBeanResponse<DmpDevelopTaskDto> baseBeanResponse = new BaseBeanResponse<>();
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.selectTaskInfoByParam(treeId);
DmpDevelopTaskDto dmpDevelopTaskDto = MyDmpDevelopTaskConverter.INSTANCE().domain2dto(dmpDevelopTask);
baseBeanResponse.setCode(StatuConstant.SUCCESS_CODE);
baseBeanResponse.setMessage("查询成功");
baseBeanResponse.setData(dmpDevelopTaskDto);
return baseBeanResponse;
}
/**
* 任务修改
*/
@Override
@Transactional(rollbackFor = Exception.class)
public BaseBeanResponse<DmpDevelopTask> edit(DmpDevelopTask dmpDevelopTask, HttpServletRequest httpRequest)
throws Exception {
BaseBeanResponse<DmpDevelopTask> baseBeanResponse = new BaseBeanResponse<DmpDevelopTask>();
// 修改人
dmpDevelopTask.setUpdateUserId(getHttpRequestUserId(httpRequest).toString());
// 修改时间
dmpDevelopTask.setUpdateTime(new Date());
// 版本设置
// 获取已存在版本
DmpDevelopTask dmpDevelopTaskDb = dmpDevelopTaskDao.selectTaskById(dmpDevelopTask.getId().toString());
String version = CodeGeneratorUtils.generatorNextTaskVesion(dmpDevelopTaskDb.getVersion());
dmpDevelopTask.setVersion(version);
dmpDevelopTaskDao.update(dmpDevelopTask);
DmpDevelopTask dmpDevelopTaskDb2 = dmpDevelopTaskDao.selectTaskById(dmpDevelopTask.getId().toString());
DmpDevelopTaskHistory dmpDevelopTaskHistory = MyDmpDevelopTaskHistoryConverter.INSTANCE().task2history(dmpDevelopTaskDb2);
// 保存版本为版本数据
dmpDevelopTaskHistoryService.add(dmpDevelopTaskHistory, httpRequest);
baseBeanResponse.setCode(StatuConstant.SUCCESS_CODE);
baseBeanResponse.setMessage("修改成功");
baseBeanResponse.setData(dmpDevelopTaskDb2);
return baseBeanResponse;
}
/**
* 任务流程发布
*/
@Override
public BaseResponse flowSubmit(Long treeId, HttpServletRequest httpRequest)
throws Exception {
BaseResponse baseResponse = new BaseResponse();
// 参数校验
if (treeId == null) {
baseResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseResponse.setMessage("treeId不能为空");
return baseResponse;
}
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.get(treeId);
FlowPro flowPro = MyDmpDevelopTaskConverter.INSTANCE().task2flowpro(dmpDevelopTask);
DmpNavigationTree dmpNavigationTree = dmpNavigationTreeDao.queryById(treeId.intValue());
Long projectId = Long.parseLong(dmpNavigationTree.getProjectId().toString());
flowPro.setPublishedToProjectId(projectId);
......@@ -1327,93 +1302,93 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
}
updateDevelopTask.setIsSubmit(flowPro.getIsSubmit());
updateDevelopTask.setIsGziped(1);
//设置发布版本
updateDevelopTask.setPublishVersion(queryDmpDevelopTask.getVersion());
dmpDevelopTaskDao.update(updateDevelopTask);
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setMessage("工作流程保存发布成功");
} else {
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setMessage(baseResponsePublish.getMessage());
}
return baseResponse;
}
/**
*根据treeId获取离线任务xmlFileName
*/
@Override
public String getExecXmlFileName(Long syncTaskTreeId) throws Exception {
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.selectTaskInfoByParam(syncTaskTreeId);
if (!dmpDevelopTask.getTaskType().equals(CommConstant.TASK_TYPE_OFFLINE)) {
throw new RuntimeException("非离线任务！");
}
String xmlContent = convert2SyncXmlContent(dmpDevelopTask);
String xmlFileName = XmlUtils.getPropertyValue(xmlContent, "name");
return xmlFileName;
}
/**
*运行任务
*/
@Override
public BaseResponse taskAzkabanRun(Long treeId, HttpServletRequest httpRequest) throws Exception {
BaseResponse baseResponse = new BaseResponse();
DmpDevelopTask developTask = dmpDevelopTaskDao.get(treeId);
//先发布任务
boolean flag = publishAndExecute(developTask);
if (flag) {
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setMessage("立即运行成功");
} else {
baseResponse.setCode(StatuConstant.FAILURE_CODE);
baseResponse.setMessage("立即运行失败");
}
return baseResponse;
}
/**
}
return baseResponse;
}
/**
* 根据treeId获取离线任务xmlFileName
*/
@Override
public String getExecXmlFileName(Long syncTaskTreeId) throws Exception {
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.selectTaskInfoByParam(syncTaskTreeId);
if (!dmpDevelopTask.getTaskType().equals(CommConstant.TASK_TYPE_OFFLINE)) {
throw new RuntimeException("非离线任务！");
}
String xmlContent = convert2SyncXmlContent(dmpDevelopTask);
String xmlFileName = XmlUtils.getPropertyValue(xmlContent, "name");
return xmlFileName;
}
/**
* 运行任务
*/
@Override
public BaseResponse taskAzkabanRun(Long treeId, HttpServletRequest httpRequest) throws Exception {
BaseResponse baseResponse = new BaseResponse();
DmpDevelopTask developTask = dmpDevelopTaskDao.get(treeId);
//先发布任务
boolean flag = publishAndExecute(developTask);
if (flag) {
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setMessage("立即运行成功");
} else {
baseResponse.setCode(StatuConstant.FAILURE_CODE);
baseResponse.setMessage("立即运行失败");
}
return baseResponse;
}
/**
* 发布流程
*/
private boolean publishAndExecute(DmpDevelopTask dmpDevelopTask) throws Exception {
Integer taskId = dmpDevelopTask.getId(); //任务id
Integer projectId = dmpDevelopTask.getProjectId(); //项目id
String treeName = dmpDevelopTask.getName(); //任务流程名称
//获取项目配置信息
Map<String, Object> param = new HashMap<String, Object>();
param.put("projectId", projectId);
List<DmpProjectConfigInfoDto> list = dmpProjectConfigInfoMapper.findList(param);
if (CollectionUtils.isEmpty(list)) {
throw new RuntimeException("项目没有设置配置信息，请联系管理员！");
}
throw new RuntimeException("项目没有设置配置信息，请联系管理员！");
}
DmpProjectConfigInfoDto dmpProjectConfigInfoDto = list.get(0);
//校验任务类型并配置参数
String taskType = dmpDevelopTask.getTaskType();
String azkabanExectorTaskExec = "";
String taskAlias = "";
if (StringUtils.isEmpty(taskType)) {
throw new RuntimeException("任务的任务类型为空！");
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSHELL)) {
azkabanExectorTaskExec = dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanExectorShellExec(); //执行shell任务命令
taskAlias = "shell";
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSQL)) {
azkabanExectorTaskExec = dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanExectorSqlExec(); //执行sql任务命令
taskAlias = "sql";
} else {
throw new RuntimeException("该任务类型不能调用此方法发布，请检查任务类型！");
}
throw new RuntimeException("任务的任务类型为空！");
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSHELL)) {
azkabanExectorTaskExec = dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanExectorShellExec(); //执行shell任务命令
taskAlias = "shell";
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSQL)) {
azkabanExectorTaskExec = dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanExectorSqlExec(); //执行sql任务命令
taskAlias = "sql";
} else {
throw new RuntimeException("该任务类型不能调用此方法发布，请检查任务类型！");
}
String azkabanLocalTaskFilePath = dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanLocalTaskFilePath(); //文件路径
String azkabanMonitorUrl = dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanMonitorUrl();//AZKABAN WEB服务地址
......@@ -1467,126 +1442,126 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
//上次zip包到azkaban
String localTaskZipAbsolutePath = localTaskZipPath + "/" + localZipTargetFileName;
AzkabanApiUtils2 azkabanApiUtils = new AzkabanApiUtils2(azkabanMonitorUrl, redisTemplate);
return azkabanApiUtils.loginCreateProjectuploadZipAndExecute("jz_localflow_"+taskAlias+"_" + projectId, "local_"+taskAlias+"_project", localTaskZipAbsolutePath, treeName);
return azkabanApiUtils.loginCreateProjectuploadZipAndExecute("jz_localflow_" + taskAlias + "_" + projectId, "local_" + taskAlias + "_project", localTaskZipAbsolutePath, treeName);
}
/**
*停止任务
*/
@Override
public BaseBeanResponse<String> taskAzkabanStop(Long treeId, HttpServletRequest httpRequest) throws Exception {
// TODO Auto-generated method stub
BaseBeanResponse<String> baseBeanResponse = new BaseBeanResponse<String>();
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.get(treeId);
if (dmpDevelopTask==null) {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseBeanResponse.setMessage("任务不存在");
return baseBeanResponse;
}
Integer projectId = dmpDevelopTask.getProjectId();
String treeName = dmpDevelopTask.getName();
//获取项目配置信息
/**
* 停止任务
*/
@Override
public BaseBeanResponse<String> taskAzkabanStop(Long treeId, HttpServletRequest httpRequest) throws Exception {
// TODO Auto-generated method stub
BaseBeanResponse<String> baseBeanResponse = new BaseBeanResponse<String>();
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.get(treeId);
if (dmpDevelopTask == null) {
baseBeanResponse.setCode(StatuConstant.CODE_ERROR_PARAMETER);
baseBeanResponse.setMessage("任务不存在");
return baseBeanResponse;
}
Integer projectId = dmpDevelopTask.getProjectId();
String treeName = dmpDevelopTask.getName();
//获取项目配置信息
Map<String, Object> param = new HashMap<String, Object>();
List<DmpProjectConfigInfoDto> list = dmpProjectConfigInfoMapper.findList(param);
if (CollectionUtils.isEmpty(list)) {
throw new RuntimeException("项目没有设置配置信息，请联系管理员！");
}
throw new RuntimeException("项目没有设置配置信息，请联系管理员！");
}
DmpProjectConfigInfoDto dmpProjectConfigInfoDto = list.get(0);
//校验任务类型并配置参数
//校验任务类型并配置参数
String taskType = dmpDevelopTask.getTaskType();
String taskAlias = "";
if (StringUtils.isEmpty(taskType)) {
throw new RuntimeException("任务的任务类型为空！");
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSHELL)) {
taskAlias = "shell";
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSQL)) {
taskAlias = "sql";
} else {
throw new RuntimeException("该任务类型不能调用此方法停止，请检查任务类型！");
}
String azkabanMonitorUrl = dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanMonitorUrl();//AZKABAN WEB服务地址
AzkabanApiUtils2 azkabanApiUtils = new AzkabanApiUtils2(azkabanMonitorUrl, redisTemplate);
String exeIdsStr = azkabanApiUtils.stopFlow("jz_localflow_"+taskAlias+"_" + projectId, treeName);
baseBeanResponse.setCode(StatuConstant.SUCCESS_CODE);
baseBeanResponse.setMessage("停止成功");
baseBeanResponse.setData(exeIdsStr);
return baseBeanResponse;
}
/**
*SHELL/SQL版本发布
*/
@Override
public BaseResponse taskPublish(Long treeId, HttpServletRequest httpRequest) throws Exception {
BaseResponse baseResponse = new BaseResponse();
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.get(treeId);
// 修改人
dmpDevelopTask.setUpdateUserId(getHttpRequestUserId(httpRequest).toString());
// 修改时间
dmpDevelopTask.setUpdateTime(new Date());
//设置发布版本
dmpDevelopTask.setPublishVersion(dmpDevelopTask.getVersion());
dmpDevelopTaskDao.update(dmpDevelopTask);
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setMessage("发布成功");
return baseResponse;
}
/**
*软删除任务
*/
@Override
@Transactional(rollbackFor = Exception.class)
public BaseResponse softDeleteByTreeId(Integer treeId, HttpServletRequest httpRequest) throws Exception {
BaseResponse baseResponse = new BaseResponse();
DmpDevelopTask developTask = dmpDevelopTaskDao.get(treeId.longValue());
String taskType = developTask.getTaskType();
if (taskType.equals(CommConstant.TASK_TYPE_DEVELOP)) {
//取消发布的任务
BaseResponse baseResponseAzkaban = flowService.deleteAzkabanFlow(developTask);
if (baseResponseAzkaban.getCode().equals(StatuConstant.FAILURE_CODE)) {
throw new RuntimeException("azkaban取消发布任务失败");
}
}else if (taskType.equals(CommConstant.TASK_TYPE_DEVSHELL) || taskType.equals(CommConstant.TASK_TYPE_DEVSQL)) {
//什么也不做
}else {
baseResponse.setCode(StatuConstant.CODE_DATA_NOTMEET);
baseResponse.setMessage("任务类型不适合调用该方法");
return baseResponse;
}
//软删除树
DmpNavigationTree dmpNavigationTree = new DmpNavigationTree();
dmpNavigationTree.setId(treeId);
dmpNavigationTree.setDataStatus("0");
dmpNavigationTreeDao.update(dmpNavigationTree);
//软删除任务
developTask.setDataStatus("0");
dmpDevelopTaskDao.update(developTask);
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setMessage("软删除成功");
return baseResponse;
}
throw new RuntimeException("任务的任务类型为空！");
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSHELL)) {
taskAlias = "shell";
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSQL)) {
taskAlias = "sql";
} else {
throw new RuntimeException("该任务类型不能调用此方法停止，请检查任务类型！");
}
String azkabanMonitorUrl = dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanMonitorUrl();//AZKABAN WEB服务地址
AzkabanApiUtils2 azkabanApiUtils = new AzkabanApiUtils2(azkabanMonitorUrl, redisTemplate);
String exeIdsStr = azkabanApiUtils.stopFlow("jz_localflow_" + taskAlias + "_" + projectId, treeName);
baseBeanResponse.setCode(StatuConstant.SUCCESS_CODE);
baseBeanResponse.setMessage("停止成功");
baseBeanResponse.setData(exeIdsStr);
return baseBeanResponse;
}
/**
* SHELL/SQL版本发布
*/
@Override
public BaseResponse taskPublish(Long treeId, HttpServletRequest httpRequest) throws Exception {
BaseResponse baseResponse = new BaseResponse();
DmpDevelopTask dmpDevelopTask = dmpDevelopTaskDao.get(treeId);
// 修改人
dmpDevelopTask.setUpdateUserId(getHttpRequestUserId(httpRequest).toString());
// 修改时间
dmpDevelopTask.setUpdateTime(new Date());
//设置发布版本
dmpDevelopTask.setPublishVersion(dmpDevelopTask.getVersion());
dmpDevelopTaskDao.update(dmpDevelopTask);
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setMessage("发布成功");
return baseResponse;
}
/**
* 软删除任务
*/
@Override
@Transactional(rollbackFor = Exception.class)
public BaseResponse softDeleteByTreeId(Integer treeId, HttpServletRequest httpRequest) throws Exception {
BaseResponse baseResponse = new BaseResponse();
DmpDevelopTask developTask = dmpDevelopTaskDao.get(treeId.longValue());
String taskType = developTask.getTaskType();
if (taskType.equals(CommConstant.TASK_TYPE_DEVELOP)) {
//取消发布的任务
BaseResponse baseResponseAzkaban = flowService.deleteAzkabanFlow(developTask);
if (baseResponseAzkaban.getCode().equals(StatuConstant.FAILURE_CODE)) {
throw new RuntimeException("azkaban取消发布任务失败");
}
} else if (taskType.equals(CommConstant.TASK_TYPE_DEVSHELL) || taskType.equals(CommConstant.TASK_TYPE_DEVSQL)) {
//什么也不做
} else {
baseResponse.setCode(StatuConstant.CODE_DATA_NOTMEET);
baseResponse.setMessage("任务类型不适合调用该方法");
return baseResponse;
}
//软删除树
DmpNavigationTree dmpNavigationTree = new DmpNavigationTree();
dmpNavigationTree.setId(treeId);
dmpNavigationTree.setDataStatus("0");
dmpNavigationTreeDao.update(dmpNavigationTree);
//软删除任务
developTask.setDataStatus("0");
dmpDevelopTaskDao.update(developTask);
baseResponse.setCode(StatuConstant.SUCCESS_CODE);
baseResponse.setMessage("软删除成功");
return baseResponse;
}
/**
* 获取执行实例的日志详情
*
......@@ -1597,7 +1572,7 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
@Override
public JsonResult queryExamplesLogByExecId(String execId) throws Exception {
List<Map> list = executionFlowsMapper.queryExamplesLogByExecId(execId);
if(list.size() > 0 && list != null) {
if (list.size() > 0 && list != null) {
list.forEach(map -> {
//map.put("log", map.get("log"));
});
......@@ -1605,4 +1580,35 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
return JsonResult.ok(list);
}
/**
* 运维大屏--获取任务状态
*
* @param projectId
* @return
* @author Bellamy
* @since 2021-02-22
*/
@Override
public JsonResult getTaskStatus(String projectId) throws Exception {
Map<String, Integer> taskStatus = new HashMap<>();
//RUNNING, FAILED,SUCCESS
//实时任务状态
Map<String, Object> realTimeTask = dmpRealtimeSyncInfoDao.queryTaskStatus(projectId);
Integer realTimeRunNum = Integer.valueOf(String.valueOf(realTimeTask.get("running")));
Integer realTimeFailedNum = Integer.valueOf(String.valueOf(realTimeTask.get("failed")));
//业务流程实例状态 :30-正在运行,50-运行成功,60-kill任务,70-运行失败
Map<String, Object> flowStatus = executionFlowsMapper.queryTaskInstanceStatus();
Integer flowRunNum = Integer.valueOf(String.valueOf(flowStatus.get("running")));
Integer flowSuccessNum = Integer.valueOf(String.valueOf(flowStatus.get("success")));
Integer flowFailedNum = Integer.valueOf(String.valueOf(flowStatus.get("failed")));
Integer flowManualNum = Integer.valueOf(String.valueOf(flowStatus.get("manualNum")));
taskStatus.put("running", realTimeRunNum + flowRunNum);
taskStatus.put("success", flowSuccessNum);
taskStatus.put("failed", realTimeFailedNum + flowFailedNum);
taskStatus.put("manual", flowManualNum);
return JsonResult.ok(taskStatus);
}
}
\ No newline at end of file
src/main/resources/application-test.yml
View file @ cd4b5211
# 测试环境配置
server:
port: 7181
port: 7183
#contextPath: /resource
management:
port: 54001
port: 54003
health:
mail:
enabled: false
......
src/main/resources/mapper/dmp/DmpRealtimeSyncInfoMapper.xml
View file @ cd4b5211
......@@ -546,4 +546,17 @@
#{item}
</foreach>
</update>
<!--实时任务状态-->
<select id="queryTaskStatus" resultType="java.util.Map" parameterType="string">
select
sum(case when t1.status='FAILED' then 1 else 0 end) failed,
sum(case when t1.status='RUNNING' then 1 else 0 end) running,
sum(case when t1.status='PAUSED' then 1 else 0 end) paused
from
dmp_realtime_sync_info t1
inner join dmp_navigation_tree t2 on t1.tree_id=t2.id
where 1=1 and t1.data_status='1' and t2.data_status='1'
and t1.project_id =#{projectId}
</select>
</mapper>
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment