commit 数据运维

src/main/java/com/jz/common/enums/FlowInstanceEnum.java

+package com.jz.common.enums;
+
+/**
+ * 业务流程实例状态
+ *
+ * @author Bellamy
+ * @since 2020-11-30 14:30:23
+ */
+public enum FlowInstanceEnum {
+    /**
+     * 运行
+     */
+    running("running", "30"),
+
+    /**
+     * 成功
+     */
+    success("success", "50"),
+
+    kill("kill", "60"),
+
+    failed("failed", "70"),
+    ;
+
+    private String code;
+
+    private String value;
+
+    private FlowInstanceEnum(String code, String value) {
+        this.code = code;
+        this.value = value;
+    }
+
+    public static FlowInstanceEnum get(String code) {
+        if (code == null) {
+            return null;
+        }
+        for (FlowInstanceEnum status : values()) {
+            if (status.getCode().equalsIgnoreCase(code)) {
+                return status;
+            }
+        }
+        return null;
+    }
+
+    public String getCode() {
+        return code;
+    }
+
+    public void setCode(String code) {
+        this.code = code;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+}

src/main/java/com/jz/common/utils/AzkabanApiUtils2.java

@@ -7,6 +7,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import com.jz.dmp.modules.controller.dataOperation.bean.SetSlaReq;
+import com.jz.dmp.modules.controller.dataOperation.bean.SetSlaRulesReq;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.core.io.FileSystemResource;
@@ -43,7 +45,7 @@ public class AzkabanApiUtils2 {
     private String azkabanServerUrl;
     private String userName;
     private String password;
-    
+
 	private RedisTemplate redisTemplate;
 	private static final StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
 	private static final FastJsonRedisSerializer fastJsonRedisSerializer = new FastJsonRedisSerializer(SSOUserInfo.class);
@@ -104,38 +106,38 @@ public class AzkabanApiUtils2 {
         	LOGGER.error(azkabanServerUrl+"-----"+linkedMultiValueMap+" sessionId 为空");
 			throw new RuntimeException("登陆失败");
 		}*/
-        String sessionId = SessionUtils.getSession().getId(); 
-    	
+        String sessionId = SessionUtils.getSession().getId();
+
         //通过redis方式登录Azkaban
         String redisKey = "spring:sessions:sessions:"+sessionId;
         SSOUserInfo ssoUserInfo = (SSOUserInfo) redisTemplate.opsForValue().get(redisKey);
-		
+
 		if (ssoUserInfo == null) {
 			redisTemplate.setKeySerializer(stringRedisSerializer);
 			redisTemplate.setValueSerializer(fastJsonRedisSerializer);
 			redisTemplate.opsForValue().set(redisKey, getSSOuserInfo());
 		}
-			
+
         System.err.println("----sessionId="+sessionId);
         LOGGER.info("----sessionId="+sessionId);
         return sessionId; //SessionUtils.getSession().getId();
     }
-    
-    /**  
-    * @Title: getSSOuserInfo  
-    * @Description: TODO(生成azkaban登录需要保存的实体)  
-    * @param @return    参数  
-    * @return SSOUserInfo    返回类型  
-    * @throws  
-    */ 
+
+    /**
+    * @Title: getSSOuserInfo
+    * @Description: TODO(生成azkaban登录需要保存的实体)
+    * @param @return    参数
+    * @return SSOUserInfo    返回类型
+    * @throws
+    */
     private SSOUserInfo getSSOuserInfo(){
-		
+
     	Map<String,String> rolePermissMap =new HashMap<>();
-		
+
 		DmpMember dmpMember = SessionUtils.getSecurityUser();
-		
+
 		List<DmpRole> memberProjectRoles = dmpMember.getMemberProjectRoleList();
-		
+
 		for (DmpRole role : memberProjectRoles) {
 			rolePermissMap.put(role.getRoleType(), role.getRemark());
 		}
@@ -447,7 +449,7 @@ public class AzkabanApiUtils2 {
         }
         return true;
     }
-    
+
     /**
      * 创建一个新的任务 ,包含  登录  创建项目  上传zip包  并立即执行
      *
@@ -458,7 +460,7 @@ public class AzkabanApiUtils2 {
     public boolean loginCreateProjectuploadZipAndExecute(String projectName, String projectDesc, String uploadZipPath, String flowName) {
         //登录
         String sessionId = login();
-    	
+
         //创建项目
         createAzkabanProject(sessionId, projectName, projectDesc);
 
@@ -471,7 +473,7 @@ public class AzkabanApiUtils2 {
 
         return true;
     }
-    
+
     /**
      * 调度一个工作流(one time)
      *
@@ -521,14 +523,14 @@ public class AzkabanApiUtils2 {
         LOGGER.info("调度projectName=" + projectName + ",flowName=" + flowName + "one time 成功");
         return (Integer) postForObject.get("execid");
     }
-    
+
     /**
      * 停止一个工作流调度
      *
      * @param sessionId   会话id
      * @param projectName 项目名
      * @param flowName    azkaban工作流名称
-     * @return 
+     * @return
      */
     @SuppressWarnings("unchecked")
     public String stopFlow(String projectName, String flowName) {
@@ -542,7 +544,7 @@ public class AzkabanApiUtils2 {
         }
         //登录
         String sessionId = login();
-        
+
         //找出任务正在执行的所有实例
         Integer[] execIds = getFlowRuning(sessionId, projectName, flowName);
         String execIdsStr = "";
@@ -555,7 +557,7 @@ public class AzkabanApiUtils2 {
 
         return execIdsStr;
     }
-    
+
     /**
      * 获取任务所有正在执行的实例
      *
@@ -602,18 +604,18 @@ public class AzkabanApiUtils2 {
             String message = (String) postForObject.get("message");
             throw new RuntimeException(message);
         }
-        
+
         Integer[] execIds = new Integer[] {};
-        
+
         if (postForObject.get("execIds")!=null) {
         	execIds = (Integer[])postForObject.get("execIds");
 		}
-        
+
         LOGGER.info("获取所有任务("+projectName+")正在调度execids"+execIds.toString());
-        
+
         return execIds;
     }
-    
+
     /**
      * 获取任务所有正在执行的实例
      *
@@ -651,9 +653,9 @@ public class AzkabanApiUtils2 {
             String message = (String) postForObject.get("message");
             throw new RuntimeException(message);
         }
-        
+
         LOGGER.info("停止调度（"+execid+"）成功");
-        
+
         return true;
     }
 
@@ -663,17 +665,17 @@ public class AzkabanApiUtils2 {
         System.err.println(dw_test);
     }
 
-	/**  
-	* @Title: getSyncingFlowExecution  
-	* @Description: TODO(获取同步状态结果)  
+	/**
+	* @Title: getSyncingFlowExecution
+	* @Description: TODO(获取同步状态结果)
 	* @param @param projectId
 	* @param @param flow
 	* @param @param start
 	* @param @param length
-	* @param @return    参数  
-	* @return List<FlowExecution>    返回类型  
-	* @throws  
-	*/ 
+	* @param @return    参数
+	* @return List<FlowExecution>    返回类型
+	* @throws
+	*/
 	public List<FlowExecution> getSyncingFlowExecution(Long projectId, String flow, Integer start, Integer length) {
 
 		List<FlowExecution> list = new ArrayList<FlowExecution>();
@@ -724,6 +726,55 @@ public class AzkabanApiUtils2 {
 
 		return list;
 	}
-	
+
+    /**
+     * Set SLA
+     * @author Bellamy
+     * @return 调用成功后，返回execid
+     */
+    @SuppressWarnings("unchecked")
+    public String saveSla(SetSlaReq req) {
+        String sessionId = login();
+        String executeFlowUrl = azkabanServerUrl + "/schedule?token=" + sessionId;
+        LinkedMultiValueMap<String, Object> linkedMultiValueMap = new LinkedMultiValueMap<String, Object>();
+        linkedMultiValueMap.add("session.id", sessionId);
+        linkedMultiValueMap.add("ajax", "setSla");
+        linkedMultiValueMap.add("scheduleId", req.getScheduleId());
+        linkedMultiValueMap.add("slaEmails",req.getSlaEmails());
+
+        List<SetSlaRulesReq> slaRule = req.getSlaRule();
+        for (int i = 0; i < slaRule.size(); i++) {
+            SetSlaRulesReq sla = slaRule.get(i);
+            linkedMultiValueMap.add("settings[" + i +"]", sla.getFlow() +"," + sla.getSlaRule() +"," + sla.getDuration()+"," +sla.getEmailNotification()+"," +sla.getKill());
+        }
+
+        Map<String, Object> postForObject = null;
+        try {
+            postForObject = bulidRestTemplate().postForObject(executeFlowUrl, linkedMultiValueMap, Map.class);
+        } catch (Exception e) {
+            LOGGER.error(executeFlowUrl + "----" + linkedMultiValueMap + "----设置SLA接口异常");
+            e.printStackTrace();
+            throw new RuntimeException("设置SLA接口异常");
+        }
+
+        if (postForObject.containsKey("error")){
+            throw new RuntimeException(postForObject.get("error").toString());
+        }
+        String status = (String) postForObject.get("status");
+        if ("error".equals(status)) {
+            String message = (String) postForObject.get("message");
+            throw new RuntimeException(message);
+        }
+
+        /*Integer[] execIds = new Integer[] {};
+
+        if (postForObject.get("execIds")!=null) {
+            execIds = (Integer[])postForObject.get("execIds");
+        }
+
+        LOGGER.info("获取所有任务("+projectName+")正在调度execids"+execIds.toString());*/
+
+        return null;
+    }
 }
 

src/main/java/com/jz/dmp/azkaban/dao/ExecutionFlowsMapper.java

@@ -40,5 +40,5 @@ public interface ExecutionFlowsMapper {
 
     Map<String, Object> queryTaskInstanceStatus() throws Exception;
 
-    List<Map> queryLastStatus(@Param("taskName") String[] taskName) throws Exception;
+    List<Map> queryLastStatus(@Param("taskName") String[] taskName, @Param("status") String status) throws Exception;
 }

src/main/java/com/jz/dmp/modules/controller/dataOperation/DmpDevExamplesController.java

@@ -3,10 +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.dataOperation.bean.DataDevExamplesListDto;
-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.dataOperation.bean.*;
 import com.jz.dmp.modules.service.DmpDevelopTaskService;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiImplicitParam;
@@ -83,4 +80,25 @@ public class DmpDevExamplesController {
         return result;
     }
 
+    /**
+     * 设置SLA
+     *
+     * @return
+     * @author Bellamy
+     * @since 2021-02-03
+     */
+    @ApiOperation(value = "设置SLA", notes = "设置SLA")
+    @PostMapping(value = "/setSla")
+    public JsonResult setSla(@RequestBody SetSlaReq req) {
+        JsonResult result = new JsonResult();
+        try {
+            result = dmpDevelopTaskService.setSla(req);
+        } catch (Exception e) {
+            result.setCode(ResultCode.INTERNAL_SERVER_ERROR);
+            result.setMessage("failed！");
+            e.printStackTrace();
+        }
+        return result;
+    }
+
 }

src/main/java/com/jz/dmp/modules/controller/dataOperation/DmpDevTaskController.java

@@ -84,7 +84,7 @@ public class DmpDevTaskController {
      * @since 2021-02-22
      */
     @ApiOperation(value = "运维大屏--重点关注任务状态", notes = "重点关注")
-    @PostMapping(value = "/getTaskStatus")
+    @GetMapping(value = "/getTaskStatus")
     public JsonResult getTaskStatus(@RequestParam String projectId) throws Exception {
         if (StringUtils.isEmpty(projectId)) {
             return new JsonResult(ResultCode.PARAMS_ERROR, "projectId不能为空！");

src/main/java/com/jz/dmp/modules/controller/dataOperation/bean/SetSlaReq.java

+package com.jz.dmp.modules.controller.dataOperation.bean;
+
+
+import com.jz.common.page.BasePageBean;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * @ClassName: SetSlaReq
+ * @Description: set sla请求参数
+ * @Author：Bellamy
+ * @Date 2021/01/13
+ * @Version 1.0
+ */
+@Data
+@ApiModel(value = "set sla请求参数", description = "et sla请求参数")
+public class SetSlaReq implements Serializable {
+
+    private static final long serialVersionUID = 4274039208139198292L;
+    @ApiModelProperty(value = "scheduleId")
+    private String scheduleId;
+
+    @ApiModelProperty(value = "邮件")
+    private String slaEmails;
+
+    @ApiModelProperty(value = "sla规则")
+    private List<SetSlaRulesReq> slaRule;
+
+    @ApiModelProperty(value = "projectId")
+    @NotEmpty(message = "projectId不能为空!")
+    @NotNull(message = "projectId不能为空!")
+    private String projectId;
+
+
+}

src/main/java/com/jz/dmp/modules/controller/dataOperation/bean/SetSlaRulesReq.java

+package com.jz.dmp.modules.controller.dataOperation.bean;
+
+
+import com.jz.common.page.BasePageBean;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+import java.io.Serializable;
+
+/**
+ * @ClassName: SetSlaRulesReq
+ * @Description: 数据运维-SLA请求参数
+ * @Author：Bellamy
+ * @Date 2021/01/13
+ * @Version 1.0
+ */
+@Data
+@ApiModel(value = "SLA请求参数", description = "SLA请求参数")
+public class SetSlaRulesReq implements Serializable {
+
+    private static final long serialVersionUID = 7548870996371216281L;
+    @ApiModelProperty(value = "flow/job")
+    @NotNull(message = "flow/job不能为空")
+    @NotEmpty(message = "flow/job不能为空")
+    private String flow;
+
+    @ApiModelProperty(value = "sla 规则")
+    private String slaRule;
+
+    @ApiModelProperty(value = "duration")
+    private String duration;
+
+    @ApiModelProperty(value = "邮件通知")
+    private Boolean emailNotification;
+
+    @ApiModelProperty(value = "kill")
+    private Boolean kill;
+
+}

src/main/java/com/jz/dmp/modules/service/DmpDevelopTaskService.java

@@ -9,10 +9,7 @@ import com.jz.common.page.PageInfoResponse;
 import com.jz.dmp.modules.controller.DataIntegration.bean.flow.FlowPro;
 import com.jz.dmp.modules.controller.bean.DmpDevelopTaskDto;
 import com.jz.dmp.modules.controller.bean.DmpDevelopTaskRequest;
-import com.jz.dmp.modules.controller.dataOperation.bean.DataDevExamplesListDto;
-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.dataOperation.bean.*;
 import com.jz.dmp.modules.model.DmpDevelopTask;
 
 /**
@@ -183,4 +180,12 @@ public interface DmpDevelopTaskService {
 	 * @since 2021-02-22
 	 */
 	JsonResult getTaskStatus(String projectId) throws Exception;
+	/**
+	 * 设置SLA
+	 *
+	 * @return
+	 * @author Bellamy
+	 * @since 2021-02-03
+	 */
+	JsonResult setSla(SetSlaReq req) throws Exception;
 }
\ No newline at end of file

src/main/java/com/jz/dmp/modules/service/impl/DmpDevelopTaskServiceImpl.java

@@ -12,6 +12,7 @@ import com.jz.common.constant.CommConstant;
 import com.jz.common.constant.JsonResult;
 import com.jz.common.constant.ResultCode;
 import com.jz.common.constant.StatuConstant;
+import com.jz.common.enums.FlowInstanceEnum;
 import com.jz.common.enums.ModuleLogEnum;
 import com.jz.common.page.PageInfoResponse;
 import com.jz.common.persistence.BaseService;
@@ -24,10 +25,7 @@ import com.jz.dmp.modules.controller.bean.DmpDevelopTaskDto;
 import com.jz.dmp.modules.controller.bean.DmpDevelopTaskRequest;
 import com.jz.dmp.modules.controller.bean.MyDmpDevelopTaskConverter;
 import com.jz.dmp.modules.controller.bean.MyDmpDevelopTaskHistoryConverter;
-import com.jz.dmp.modules.controller.dataOperation.bean.DataDevExamplesListDto;
-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.dataOperation.bean.*;
 import com.jz.dmp.modules.controller.projconfig.bean.DmpProjectConfigInfoDto;
 import com.jz.dmp.modules.dao.*;
 import com.jz.dmp.modules.dao.projconfig.DmpProjectConfigInfoMapper;
@@ -789,6 +787,16 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
                 req.setTreeIdOrName(req.getTreeIdOrName().trim());//节点名称
             }
         }
+        if (StringUtils.isNotBlank(req.getStatus())) {
+            if (FlowInstanceEnum.running.getCode().equals(req.getStatus())) {
+                req.setStatus(FlowInstanceEnum.running.getValue());
+            } else if (FlowInstanceEnum.failed.getCode().equals(req.getStatus())) {
+                req.setStatus(FlowInstanceEnum.failed.getValue());
+            } else {
+                //全部
+                req.setStatus("");
+            }
+        }
 
         PageHelper.startPage(req.getPageNum(), req.getPageSize());
         List<DataDevTaskListDto> list = dmpDevelopTaskDao.queryDevTaskListPage(req);
@@ -798,11 +806,21 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
         pageInfoResponse.setMessage("查询成功");
         pageInfoResponse.setData(pageInfo);
         String taskName = "";
-        List<DataDevTaskListDto> listObj = (List<DataDevTaskListDto>) pageInfo;
+        List<DataDevTaskListDto> listObj = pageInfoResponse.getData().getList();
         for (DataDevTaskListDto str : listObj) {
             taskName += "," + str.getTaskName();
         }
-        List<Map> flowList = executionFlowsMapper.queryLastStatus(taskName.substring(1).split(","));
+        //查询实例最后执行状态
+        List<Map> flowList = executionFlowsMapper.queryLastStatus(taskName.substring(1).split(","), req.getStatus());
+        if (flowList.size() > 0 && flowList != null) {
+            for (DataDevTaskListDto str : listObj) {
+                for (Map strFlow : flowList) {
+                    if (str.getTaskName().equals(strFlow.get("taskName"))) {
+                        str.setStatus(String.valueOf(strFlow.get("status")));
+                    }
+                }
+            }
+        }
 
         return pageInfoResponse;
     }
@@ -1611,4 +1629,24 @@ public class DmpDevelopTaskServiceImpl extends BaseService implements DmpDevelop
         return JsonResult.ok(taskStatus);
     }
 
+    /**
+     * 设置SLA
+     *
+     * @return
+     * @author Bellamy
+     * @since 2021-02-03
+     */
+    @Override
+    public JsonResult setSla(SetSlaReq req) throws Exception {
+        Map<String, Object> params = new HashMap<>();
+        Long projectId = Long.valueOf(req.getProjectId()); //项目id
+        DmpProjectSystemInfo publishToProjectSystemInfo = dmpProjectDao.queryProjectSystemInfo(projectId);
+
+        //调用azkaban服务
+        String azkabanApiUrl = publishToProjectSystemInfo.getAzkabanMonitorUrl();
+        AzkabanApiUtils2 azkabanApiUtils = new AzkabanApiUtils2(azkabanApiUrl, redisTemplate);
+        azkabanApiUtils.saveSla(req);
+        return null;
+    }
+
 }
\ No newline at end of file

src/main/resources/azkabanmapper/ExecutionFlowsMapper.xml

@@ -222,7 +222,20 @@
             )t
     </select>
     <select id="queryLastStatus" resultType="java.util.Map">
-
+        SELECT
+            exec_id AS execId,
+            flow_id AS taskName,
+            max(from_unixtime( submit_time / 1000, '%Y-%m-%d %H:%i:%s' )) AS startTime,
+          (case when status='30' then '运行中' when status='50' then '成功' when status='70' then '失败' end) status
+        FROM
+            execution_flows
+        WHERE 1 = 1
+        <if test="status != null and status !='' ">and status=#{status}</if>
+        and flow_id in
+        <foreach collection="taskName" item="item" open="(" separator="," close=")">
+            #{item}
+        </foreach>
+        group by flow_id
     </select>
 
 </mapper>
\ No newline at end of file

src/main/resources/mapper/dmp/DmpDevelopTaskMapper.xml

@@ -177,7 +177,7 @@
                 dmp_navigation_tree t1
                 inner join dmp_develop_task t2 on t2.TREE_ID=t1.ID and t2.data_status ='1'
                 left join dmp_member t3 on t1.create_user_id=t3.user_id
-            where 1=1 and IS_LEVEL ='1'
+            where 1=1 and IS_LEVEL ='1' and t1.data_status ='1'
             and t1.project_id = #{projectId}
             <if test="taskId != null and taskId != ''"> and t2.id =#{taskId} </if>
             <if test="treeIdOrName != null and treeIdOrName != ''"> and t1.name like concat('%',#{treeIdOrName},'%') </if>