Merge 8252d59 into f3d87e3

elasticjob-error-handler/elasticjob-error-handler-dingtalk/pom.xml

@@ -0,0 +1,75 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.shardingsphere.elasticjob</groupId>
+        <artifactId>elasticjob-error-handler</artifactId>
+        <version>3.0.0-beta-SNAPSHOT</version>
+    </parent>
+    <artifactId>elasticjob-error-handler-dingtalk</artifactId>
+    <name>${project.artifactId}</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.shardingsphere.elasticjob</groupId>
+            <artifactId>elasticjob-infra-common</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.shardingsphere.elasticjob</groupId>
+            <artifactId>elasticjob-restful</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>jcl-over-slf4j</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>log4j-over-slf4j</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

elasticjob-error-handler/elasticjob-error-handler-dingtalk/src/main/java/org/apache/shardingsphere/elasticjob/error/handler/config/DingtalkConfiguration.java

@@ -0,0 +1,38 @@
+/*
+ * 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.shardingsphere.elasticjob.error.handler.config;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.Setter;
+
+@Getter
+@Setter
+@RequiredArgsConstructor
+public class DingtalkConfiguration {
+
+    private final String webhook;
+
+    private final String keyword;
+
+    private final String secret;
+
+    private final Integer connectTimeout;
+
+    private final Integer readTimeout;
+}

elasticjob-error-handler/elasticjob-error-handler-dingtalk/src/main/java/org/apache/shardingsphere/elasticjob/error/handler/env/DingtalkEnvironment.java

@@ -0,0 +1,125 @@
+/*
+ * 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.shardingsphere.elasticjob.error.handler.env;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.elasticjob.error.handler.config.DingtalkConfiguration;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
+
+/**
+ * Bootstrap env.
+ */
+@Slf4j
+public final class DingtalkEnvironment {
+
+    private static final DingtalkEnvironment INSTANCE = new DingtalkEnvironment();
+
+    private static final String PROPERTIES_PATH = "conf/elasticjob-dingtalk.properties";
+
+    private final Properties properties;
+
+    private DingtalkEnvironment() {
+        properties = getProperties();
+    }
+
+    /**
+     * Get instance of Dingtalk env.
+     *
+     * @return instance of Dingtalk env.
+     */
+    public static DingtalkEnvironment getInstance() {
+        return INSTANCE;
+    }
+
+    private Properties getProperties() {
+        Properties result = new Properties();
+        try (InputStream fileInputStream = this.getClass().getClassLoader().getResourceAsStream(PROPERTIES_PATH)) {
+            if (fileInputStream != null) {
+                result.load(fileInputStream);
+            }
+        } catch (final IOException ex) {
+            log.warn("Can not load properties file from path: '{}'.", PROPERTIES_PATH);
+        }
+        setPropertiesByEnv(result);
+        return result;
+    }
+
+    private void setPropertiesByEnv(final Properties prop) {
+        for (EnvironmentArgument each : EnvironmentArgument.values()) {
+            String key = each.getKey();
+            String value = System.getProperties().getProperty(key);
+            if (!Strings.isNullOrEmpty(value)) {
+                log.info("Load property {} with value {} from ENV.", key, value);
+                prop.setProperty(each.getKey(), value);
+            }
+        }
+    }
+
+    /**
+     * Get dingtalk configuration.
+     *
+     * @return dingtalk configuration
+     */
+    public DingtalkConfiguration getDingtalkConfiguration() {
+        String webhook = getValue(EnvironmentArgument.WEBHOOK);
+        String keyword = getValue(EnvironmentArgument.KEYWORD);
+        String secret = getValue(EnvironmentArgument.SECRET);
+        String connectTimeout = getValue(EnvironmentArgument.CONNECT_TIMEOUT);
+        String readTimeout = getValue(EnvironmentArgument.READ_TIMEOUT);
+        return new DingtalkConfiguration(webhook, keyword, secret, Integer.parseInt(connectTimeout), Integer.parseInt(readTimeout));
+    }
+
+    private String getValue(final EnvironmentArgument environmentArgument) {
+        String result = properties.getProperty(environmentArgument.getKey(), environmentArgument.getDefaultValue());
+        if (environmentArgument.isRequired()) {
+            Preconditions.checkState(!Strings.isNullOrEmpty(result), String.format("Property `%s` is required.", environmentArgument.getKey()));
+        }
+        return result;
+    }
+
+    /**
+     * Env args.
+     */
+    @RequiredArgsConstructor
+    @Getter
+    public enum EnvironmentArgument {
+
+        WEBHOOK("elasticjob.dingtalk.webhook", "", true),
+
+        KEYWORD("elasticjob.dingtalk.keyword", "", false),
+
+        SECRET("elasticjob.dingtalk.secret", "", false),
+
+        CONNECT_TIMEOUT("elasticjob.dingtalk.connectTimeout", "3000", false),
+
+        READ_TIMEOUT("elasticjob.dingtalk.readTimeout", "5000", false);
+
+        private final String key;
+
+        private final String defaultValue;
+
+        private final boolean required;
+    }
+}

elasticjob-error-handler/elasticjob-error-handler-dingtalk/src/main/java/org/apache/shardingsphere/elasticjob/error/handler/impl/DingtalkJobErrorHandler.java

@@ -0,0 +1,144 @@
+/*
+ * 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.shardingsphere.elasticjob.error.handler.impl;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableMap;
+import com.google.gson.JsonObject;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.elasticjob.error.handler.config.DingtalkConfiguration;
+import org.apache.shardingsphere.elasticjob.error.handler.env.DingtalkEnvironment;
+import org.apache.shardingsphere.elasticjob.infra.exception.JobExecutionException;
+import org.apache.shardingsphere.elasticjob.infra.handler.error.JobErrorHandler;
+import org.apache.shardingsphere.elasticjob.infra.json.GsonFactory;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.io.UnsupportedEncodingException;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.security.InvalidKeyException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
+import java.util.Collections;
+
+/**
+ * Job error handler for dingtalk error message.
+ */
+@Slf4j
+public final class DingtalkJobErrorHandler implements JobErrorHandler {
+
+    @Setter
+    private DingtalkConfiguration dingtalkConfiguration;
+
+    @Override
+    public void handleException(final String jobName, final Throwable cause) {
+        if (null == dingtalkConfiguration) {
+            dingtalkConfiguration = DingtalkEnvironment.getInstance().getDingtalkConfiguration();
+        }
+        HttpURLConnection connection = null;
+        try {
+            URL url = getUrl();
+            connection = (HttpURLConnection) url.openConnection();
+            connection.setRequestMethod("POST");
+            connection.setDoOutput(true);
+            connection.setConnectTimeout(dingtalkConfiguration.getConnectTimeout());
+            connection.setReadTimeout(dingtalkConfiguration.getReadTimeout());
+            connection.setRequestProperty("Content-Type", "application/json; charset=utf-8");
+            connection.connect();
+            OutputStream outputStream = connection.getOutputStream();
+            String msg = getMsg(jobName, cause);
+            String paramJson = getParamJson(msg);
+            outputStream.write(paramJson.getBytes(StandardCharsets.UTF_8));
+            int code = connection.getResponseCode();
+            if (HttpURLConnection.HTTP_OK == code) {
+                InputStream resultInputStream = connection.getInputStream();
+                StringBuilder result = new StringBuilder();
+                try (BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(resultInputStream, StandardCharsets.UTF_8))) {
+                    String line;
+                    while (null != (line = bufferedReader.readLine())) {
+                        result.append(line);
+                    }
+                }
+                JsonObject resp = GsonFactory.getGson().fromJson(result.toString(), JsonObject.class);
+                if (!"0".equals(resp.get("errcode").getAsString())) {
+                    log.error("An exception has occurred in Job '{}', But failed to send alert by Dingtalk because of: {}", jobName, resp.get("errmsg").getAsString(), cause);
+                } else {
+                    log.error("An exception has occurred in Job '{}', Notification to Dingtalk was successful.", jobName, cause);
+                }
+            }
+        } catch (IOException | NoSuchAlgorithmException | InvalidKeyException ex) {
+            throw new JobExecutionException(ex);
+        } finally {
+            if (null != connection) {
+                connection.disconnect();
+            }
+        }
+    }
+
+    private String getParamJson(final String msg) {
+        return GsonFactory.getGson().toJson(ImmutableMap.of("msgtype", "text", "text", Collections.singletonMap("content", msg)));
+    }
+
+    private String getMsg(final String jobName, final Throwable cause) {
+        StringWriter sw = new StringWriter();
+        cause.printStackTrace(new PrintWriter(sw, true));
+        String msg = String.format("Job '%s' exception occur in job processing, caused by %s", jobName, sw.toString());
+        if (!Strings.isNullOrEmpty(dingtalkConfiguration.getKeyword())) {
+            msg = dingtalkConfiguration.getKeyword().concat(msg);
+        }
+        return msg;
+    }
+
+    private URL getUrl() throws NoSuchAlgorithmException, InvalidKeyException, UnsupportedEncodingException, MalformedURLException {
+        if (Strings.isNullOrEmpty(dingtalkConfiguration.getSecret())) {
+            return new URL(dingtalkConfiguration.getWebhook());
+        } else {
+            return new URL(getSignUrl());
+        }
+    }
+
+    private String getSignUrl() throws NoSuchAlgorithmException, InvalidKeyException, UnsupportedEncodingException {
+        Long timestamp = System.currentTimeMillis();
+        return String.format("%s&timestamp=%s&sign=%s", dingtalkConfiguration.getWebhook(), timestamp, sign(timestamp));
+    }
+
+    private String sign(final Long timestamp) throws NoSuchAlgorithmException, UnsupportedEncodingException, InvalidKeyException {
+        String stringToSign = timestamp + "\n" + dingtalkConfiguration.getSecret();
+        Mac mac = Mac.getInstance("HmacSHA256");
+        mac.init(new SecretKeySpec(dingtalkConfiguration.getSecret().getBytes("UTF-8"), "HmacSHA256"));
+        byte[] signData = mac.doFinal(stringToSign.getBytes("UTF-8"));
+        return URLEncoder.encode(new String(Base64.getEncoder().encode(signData)), "UTF-8");
+    }
+
+    @Override
+    public String getType() {
+        return "DINGTALK";
+    }
+}