dolphin 增加钉钉通知-功能

dolphin 1.6已经增加了企业微信的功能。

公司用的钉钉，那么增加功能到钉钉指定分组，可以针对分组将错误的任务或者被杀死的任务进行提示。

alert模块，增加配置：

 enterprise.dingtalk.enable=true
 enterprise.dingtalk.secret=钉钉秘钥
 enterprise.dingtalk.url=钉钉地址
# enterprise.dingtalk.url=https://oapi.dingtalk.com/robot/send?access_token=token

增加配置，Constants增加配置

    public static final String ENTERPRISE_DINGTALK_ENABLE="enterprise.dingtalk.enable";

    public static final String ENTERPRISE_DINGTALK_SECRET="enterprise.dingtalk.secret";

    public static final String ENTERPRISE_DINGTALK_URL="enterprise.dingtalk.url";

增加工具类

package org.apache.dolphinscheduler.alert.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.dolphinscheduler.common.utils.HttpUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.io.IOException;
import java.net.URLEncoder;
import java.util.Base64;

public class EnterpriseDingTalkUtils {
    public static final Logger logger = LoggerFactory.getLogger(EnterpriseWeChatUtils.class);

//    private static final boolean ENTERPRISE_DINGTALK_ENABLE = PropertyUtils.getBoolean(Constants.ENTERPRISE_DINGTALK_ENABLE);

    private static final String ENTERPRISE_DINGTALK_URL = PropertyUtils.getString(Constants.ENTERPRISE_DINGTALK_URL);

    private static final String ENTERPRISE_DINGTALK_SECRET = PropertyUtils.getString(Constants.ENTERPRISE_DINGTALK_SECRET);

    /**
     * get Enterprise Ding Ding is enable
     * @return isEnable
     */
    public static boolean isEnable(){
        return PropertyUtils.getBoolean(Constants.ENTERPRISE_DINGTALK_ENABLE);
    }

    /**
     * get Secret string and time number
     */
    public static String encodeKey(){
        try {
            //获取时间戳
            Long timestamp = System.currentTimeMillis();
            //把时间戳和密钥拼接成字符串，中间加入一个换行符
            String stringToSign = timestamp + "
" + ENTERPRISE_DINGTALK_SECRET;
            //声明一个Mac对象，用来操作字符串
            Mac mac = null;
            mac = Mac.getInstance("HmacSHA256");

            //初始化，设置Mac对象操作的字符串是UTF-8类型，加密方式是SHA256
            mac.init(new SecretKeySpec(ENTERPRISE_DINGTALK_SECRET.getBytes("UTF-8"), "HmacSHA256"));
            //把字符串转化成字节形式
            byte[] signData = mac.doFinal(stringToSign.getBytes("UTF-8"));
            //新建一个Base64编码对象
            Base64.Encoder encoder = Base64.getEncoder();
            //把上面的字符串进行Base64加密后再进行URL编码
            String sign = URLEncoder.encode(new String(encoder.encodeToString(signData)),"UTF-8");
            System.out.println(timestamp);
            System.out.println(sign);

            String result = "&timestamp=" + timestamp + "&sign=" + sign;
            return result;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
    public static String getJsonBodyString(String msg){
        JSONObject result = new JSONObject();
        JSONObject text = new JSONObject();
        text.put("content", msg);
        result.put("text", text);
        result.put("msgtype", "text");
        String jsonString = JSON.toJSONString(result);
        return jsonString;
    }

    public static void sendMessageToDingTalk(String msg){
        String enterDingTalkUrl = ENTERPRISE_DINGTALK_URL + encodeKey();
        String jsonBodyStr = getJsonBodyString(msg);
        CloseableHttpClient httpClient = HttpClients.createDefault();
        try {
            HttpPost httpPost = new HttpPost(enterDingTalkUrl);
            httpPost.setHeader("Content-Type", "application/json;charset=utf8");
            httpPost.setEntity(new StringEntity(jsonBodyStr, Constants.UTF_8));
            CloseableHttpResponse response = httpClient.execute(httpPost);
            String resp;
            try {
                HttpEntity entity = response.getEntity();
                resp = EntityUtils.toString(entity, Constants.UTF_8);
                EntityUtils.consume(entity);
            } finally {
                response.close();
            }
            logger.info("Enterprise DingTalk send [{}], param:{}, resp:{}",
                    ENTERPRISE_DINGTALK_URL, msg, resp);
        } catch (ClientProtocolException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                httpClient.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}

在server模块的worker运行逻辑（TaskExecuteThread）加入：

    @Override
    public void run() {
        String result = "";
        TaskExecuteResponseCommand responseCommand = new TaskExecuteResponseCommand(taskExecutionContext.getTaskInstanceId());
        try {
            logger.info("script path : {}", taskExecutionContext.getExecutePath());
            // task node
            TaskNode taskNode = JSONObject.parseObject(taskExecutionContext.getTaskJson(), TaskNode.class);

            // copy hdfs/minio file to local
            downloadResource(taskExecutionContext.getExecutePath(),
                    taskExecutionContext.getResources(),
                    logger);

            taskExecutionContext.setTaskParams(taskNode.getParams());
            taskExecutionContext.setEnvFile(CommonUtils.getSystemEnvPath());
            taskExecutionContext.setDefinedParams(getGlobalParamsMap());

            // set task timeout
            setTaskTimeout(taskExecutionContext, taskNode);

            taskExecutionContext.setTaskAppId(String.format("%s_%s_%s",
                    taskExecutionContext.getProcessDefineId(),
                    taskExecutionContext.getProcessInstanceId(),
                    taskExecutionContext.getTaskInstanceId()));

            task = TaskManager.newTask(taskExecutionContext, taskLogger);

            // task init
            task.init();
            result = String.format("task name=%s,task-id=%s,type=%s error",taskNode.getName(),taskNode.getId(),taskNode.getType());
            // task handle
            task.handle();

            // task result process
            task.after();
            responseCommand.setStatus(task.getExitStatus().getCode());
            responseCommand.setEndTime(new Date());
            responseCommand.setProcessId(task.getProcessId());
            responseCommand.setAppIds(task.getAppIds());
            logger.info("task instance id : {},task final status : {}", taskExecutionContext.getTaskInstanceId(), task.getExitStatus());
        } catch (Exception e) {
            logger.error("task scheduler failure", e);
            kill();
            responseCommand.setStatus(ExecutionStatus.FAILURE.getCode());
            responseCommand.setEndTime(new Date());
            responseCommand.setProcessId(task.getProcessId());
            responseCommand.setAppIds(task.getAppIds());
        } finally {
            taskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
            ResponceCache.get().cache(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command(), Event.RESULT);
            taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
            clearTaskExecPath();
            //
            if(EnterpriseDingTalkUtils.isEnable()) {
                if (responseCommand.getStatus() == ExecutionStatus.FAILURE.getCode() ||
                        responseCommand.getStatus() == ExecutionStatus.KILL.getCode()) {
                    EnterpriseDingTalkUtils.sendMessageToDingTalk(result);
                }
            }
        }
    }

参考：https://segmentfault.com/a/1190000022077236