refactor: optimize alarm code

powerjob-server/pom.xml

@@ -55,6 +55,7 @@
         <powerjob-remote-impl-akka.version>4.3.3</powerjob-remote-impl-akka.version>
         <springdoc-openapi-ui.version>1.6.14</springdoc-openapi-ui.version>
         <aliyun-sdk-oss.version>3.17.1</aliyun-sdk-oss.version>
+        <commons-collections4.version>4.4</commons-collections4.version>
     </properties>
 
     <dependencyManagement>
@@ -113,6 +114,14 @@
                 <version>${aliyun-sdk-oss.version}</version>
             </dependency>
 
+            <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-collections4 -->
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-collections4</artifactId>
+                <version>${commons-collections4.version}</version>
+            </dependency>
+
+
         </dependencies>
     </dependencyManagement>
 

powerjob-server/powerjob-server-common/pom.xml

@@ -18,4 +18,11 @@
         <maven.compiler.target>8</maven.compiler.target>
     </properties>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-collections4</artifactId>
+        </dependency>
+    </dependencies>
+
 </project>

powerjob-server/powerjob-server-common/src/main/java/tech/powerjob/server/common/utils/TestUtils.java

@@ -0,0 +1,42 @@
+package tech.powerjob.server.common.utils;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.Maps;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+
+/**
+ * 开发团队专用测试工具
+ *
+ * @author tjq
+ * @since 2023/7/31
+ */
+public class TestUtils {
+
+    private static final String TEST_CONFIG_NAME = "/.powerjob_test";
+
+    public static final String KEY_PHONE_NUMBER = "phone";
+
+    public static final String KEY_MONGO_URI = "mongoUri";
+
+    /**
+     * 获取本地的测试配置，主要用于存放一些密钥
+     * @return 测试配置
+     */
+    public static Map<String, Object> fetchTestConfig() {
+        try {
+            // 后续本地测试，密钥相关的内容统一存入 .powerjob_test 中，方便管理
+            String content = FileUtils.readFileToString(new File(System.getProperty("user.home").concat(TEST_CONFIG_NAME)), StandardCharsets.UTF_8);
+            if (StringUtils.isNotEmpty(content)) {
+                return JSONObject.parseObject(content);
+            }
+        } catch (Exception ignore) {
+        }
+        return Maps.newHashMap();
+    }
+
+}

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/alarm/AlarmCenter.java

@@ -1,19 +1,19 @@
 package tech.powerjob.server.core.alarm;
 
-import org.springframework.beans.BeanUtils;
-import org.springframework.stereotype.Component;
-import tech.powerjob.server.extension.alarm.Alarm;
-import tech.powerjob.server.extension.alarm.AlarmTarget;
-import tech.powerjob.server.extension.alarm.Alarmable;
-import tech.powerjob.server.persistence.remote.model.UserInfoDO;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Queues;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import tech.powerjob.server.extension.alarm.Alarm;
+import tech.powerjob.server.extension.alarm.AlarmTarget;
+import tech.powerjob.server.extension.alarm.Alarmable;
 
 import java.util.List;
-import java.util.concurrent.*;
+import java.util.concurrent.ExecutorService;
-import java.util.stream.Collectors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * 报警服务
@@ -40,21 +40,13 @@ public class AlarmCenter {
         });
     }
 
-    public void alarmFailed(Alarm alarm, List<UserInfoDO> targetUserList) {
+    public void alarmFailed(Alarm alarm, List<AlarmTarget> alarmTargets) {
         POOL.execute(() -> BEANS.forEach(alarmable -> {
             try {
-                alarmable.onFailed(alarm, targetUserList.stream().map(AlarmCenter::convertUserInfo2AlarmTarget).collect(Collectors.toList()));
+                alarmable.onFailed(alarm, alarmTargets);
             }catch (Exception e) {
                 log.warn("[AlarmCenter] alarm failed.", e);
             }
         }));
     }
-
-    private static AlarmTarget convertUserInfo2AlarmTarget(UserInfoDO userInfoDO) {
-        AlarmTarget alarmTarget = new AlarmTarget();
-        BeanUtils.copyProperties(userInfoDO, alarmTarget);
-
-        alarmTarget.setName(userInfoDO.getUsername());
-        return alarmTarget;
-    }
 }

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/alarm/AlarmUtils.java

@@ -0,0 +1,35 @@
+package tech.powerjob.server.core.alarm;
+
+import org.springframework.beans.BeanUtils;
+import tech.powerjob.common.utils.CollectionUtils;
+import tech.powerjob.server.extension.alarm.AlarmTarget;
+import tech.powerjob.server.persistence.remote.model.UserInfoDO;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * AlarmUtils
+ *
+ * @author tjq
+ * @since 2023/7/31
+ */
+public class AlarmUtils {
+
+    public static List<AlarmTarget> convertUserInfoList2AlarmTargetList(List<UserInfoDO> userInfoDOS) {
+        if (CollectionUtils.isEmpty(userInfoDOS)) {
+            return Collections.emptyList();
+        }
+        return userInfoDOS.stream().map(AlarmUtils::convertUserInfo2AlarmTarget).collect(Collectors.toList());
+    }
+
+    public static AlarmTarget convertUserInfo2AlarmTarget(UserInfoDO userInfoDO) {
+        AlarmTarget alarmTarget = new AlarmTarget();
+        BeanUtils.copyProperties(userInfoDO, alarmTarget);
+
+        alarmTarget.setName(userInfoDO.getUsername());
+        return alarmTarget;
+    }
+
+}

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/alarm/module/JobInstanceAlarm.java

@@ -1,6 +1,7 @@
 package tech.powerjob.server.core.alarm.module;
 
 import lombok.Data;
+import lombok.experimental.Accessors;
 import tech.powerjob.server.extension.alarm.Alarm;
 
 /**
@@ -10,6 +11,7 @@ import tech.powerjob.server.extension.alarm.Alarm;
  * @since 2020/4/30
  */
 @Data
+@Accessors(chain = true)
 public class JobInstanceAlarm implements Alarm {
     /**
      * 应用ID

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/instance/InstanceManager.java

@@ -15,6 +15,7 @@ import tech.powerjob.remote.framework.base.URL;
 import tech.powerjob.server.common.module.WorkerInfo;
 import tech.powerjob.server.common.timewheel.holder.HashedWheelTimerHolder;
 import tech.powerjob.server.common.utils.SpringUtils;
+import tech.powerjob.server.core.alarm.AlarmUtils;
 import tech.powerjob.server.core.service.UserService;
 import tech.powerjob.server.core.workflow.WorkflowInstanceManager;
 import tech.powerjob.server.core.alarm.AlarmCenter;
@@ -239,7 +240,7 @@ public class InstanceManager implements TransportServiceAware {
         if (!StringUtils.isEmpty(alertContent)) {
             content.setResult(alertContent);
         }
-        alarmCenter.alarmFailed(content, userList);
+        alarmCenter.alarmFailed(content, AlarmUtils.convertUserInfoList2AlarmTargetList(userList));
     }
 
     @Override

powerjob-server/powerjob-server-core/src/main/java/tech/powerjob/server/core/workflow/WorkflowInstanceManager.java

@@ -20,6 +20,7 @@ import tech.powerjob.common.serialize.JsonUtils;
 import tech.powerjob.common.utils.CommonUtils;
 import tech.powerjob.server.common.constants.SwitchableStatus;
 import tech.powerjob.server.common.utils.SpringUtils;
+import tech.powerjob.server.core.alarm.AlarmUtils;
 import tech.powerjob.server.core.helper.StatusMappingHelper;
 import tech.powerjob.server.core.lock.UseCacheLock;
 import tech.powerjob.server.core.service.UserService;
@@ -458,7 +459,7 @@ public class WorkflowInstanceManager {
                     content.setResult(result);
 
                     List<UserInfoDO> userList = userService.fetchNotifyUserList(wfInfo.getNotifyUserIds());
-                    alarmCenter.alarmFailed(content, userList);
+                    alarmCenter.alarmFailed(content, AlarmUtils.convertUserInfoList2AlarmTargetList(userList));
                 });
             } catch (Exception ignore) {
                 // ignore

powerjob-server/powerjob-server-extension/src/main/java/tech/powerjob/server/extension/alarm/AlarmTarget.java

@@ -1,6 +1,7 @@
 package tech.powerjob.server.extension.alarm;
 
 import lombok.Data;
+import lombok.experimental.Accessors;
 
 import java.io.Serializable;
 import java.util.Map;
@@ -13,6 +14,7 @@ import java.util.Objects;
  * @since 2023/7/16
  */
 @Data
+@Accessors(chain = true)
 public class AlarmTarget implements Serializable {
 
     private String name;

powerjob-server/powerjob-server-persistence/src/main/java/tech/powerjob/server/persistence/storage/impl/GridFsService.java

@@ -163,7 +163,7 @@ public class GridFsService extends AbstractDFsService {
     public static class GridFsCondition extends PropertyAndOneBeanCondition {
         @Override
         protected List<String> anyConfigKey() {
-            return Lists.newArrayList("spring.data.mongodb.uri", "oms.storage.dfs.mongo.uri");
+            return Lists.newArrayList("spring.data.mongodb.uri", "oms.storage.dfs.mongodb.uri");
         }
 
         @Override

powerjob-server/powerjob-server-persistence/src/test/java/tech/powerjob/server/persistence/storage/impl/GridFsServiceTest.java

@@ -1,13 +1,9 @@
 package tech.powerjob.server.persistence.storage.impl;
 
-import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.io.FileUtils;
+import tech.powerjob.server.common.utils.TestUtils;
-import org.apache.commons.lang3.StringUtils;
 import tech.powerjob.server.extension.dfs.DFsService;
 
-import java.io.File;
-import java.nio.charset.StandardCharsets;
 import java.util.Optional;
 
 /**
@@ -22,23 +18,15 @@ class GridFsServiceTest extends AbstractDfsServiceTest {
     @Override
     protected Optional<DFsService> fetchService() {
 
-        try {
+        Object mongoUri = TestUtils.fetchTestConfig().get(TestUtils.KEY_MONGO_URI);
-            // 后续本地测试，密钥相关的内容统一存入 .powerjob_test 中，方便管理
-            String content = FileUtils.readFileToString(new File(System.getProperty("user.home").concat("/.powerjob_test")), StandardCharsets.UTF_8);
-            if (StringUtils.isNotEmpty(content)) {
-                JSONObject jsonObject = JSONObject.parseObject(content);
-                Object mongoUri = jsonObject.get("mongoUri");
-                if (mongoUri != null) {
-                    GridFsService gridFsService = new GridFsService();
-                    gridFsService.initMongo(String.valueOf(mongoUri));
 
-                    return Optional.of(gridFsService);
+        if (mongoUri == null) {
-                }
+            log.info("[GridFsServiceTest] mongoUri is null, skip load!");
-            }
+            return Optional.empty();
-        } catch (Exception e) {
-            e.printStackTrace();
-            log.warn("[GridFsServiceTest] fetch mongo config failed, skip!");
         }
-        return Optional.empty();
+
+        GridFsService gridFsService = new GridFsService();
+        gridFsService.initMongo(String.valueOf(mongoUri));
+        return Optional.of(gridFsService);
     }
 }

powerjob-server/powerjob-server-starter/src/main/java/tech/powerjob/server/web/controller/TestController.java

@@ -0,0 +1,75 @@
+package tech.powerjob.server.web.controller;
+
+import com.google.common.collect.Lists;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.MapUtils;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import tech.powerjob.common.serialize.JsonUtils;
+import tech.powerjob.common.utils.CollectionUtils;
+import tech.powerjob.server.common.utils.TestUtils;
+import tech.powerjob.server.core.alarm.AlarmCenter;
+import tech.powerjob.server.core.alarm.module.JobInstanceAlarm;
+import tech.powerjob.server.extension.alarm.AlarmTarget;
+
+import javax.annotation.Resource;
+import java.util.Map;
+
+/**
+ * 开发团队专用（或者 PRO 用户用来做自检也可以 lol）
+ * 测试某些强依赖运行时环境的组件，如 Mail 告警等
+ *
+ * @author tjq
+ * @since 2023/7/31
+ */
+@Slf4j
+@RestController
+@RequestMapping("/test")
+public class TestController {
+
+    @Value("${server.port}")
+    private int port;
+
+    @Resource
+    private AlarmCenter alarmCenter;
+
+    @RequestMapping("/io")
+    public Map<String, Object> io(@RequestBody Map<String, Object> input) {
+        log.info("[TestController] input: {}", JsonUtils.toJSONString(input));
+        return input;
+    }
+
+    @GetMapping("/check")
+    public void check() {
+        Map<String, Object> testConfig = TestUtils.fetchTestConfig();
+        if (CollectionUtils.isEmpty(testConfig)) {
+            log.info("[TestController] testConfig not exist, skip check!");
+            return;
+        }
+
+        log.info("[TestController] testConfig: {}", JsonUtils.toJSONString(testConfig));
+
+        testAlarmCenter();
+    }
+
+    void testAlarmCenter() {
+        JobInstanceAlarm jobInstanceAlarm = new JobInstanceAlarm().setAppId(277).setJobId(1).setInstanceId(2)
+                .setJobName("test-alarm").setJobParams("jobParams").setInstanceParams("instanceParams")
+                .setExecuteType(1).setFinishedTime(System.currentTimeMillis());
+
+        AlarmTarget target = new AlarmTarget().setName("ald").setPhone("208140").setExtra("extra")
+                .setPhone(MapUtils.getString(TestUtils.fetchTestConfig(), TestUtils.KEY_PHONE_NUMBER))
+                .setEmail("tjq@zju.edu.cn")
+                .setWebHook(localUrlPath().concat("/test/io"));
+
+        log.info("[TestController] start to testAlarmCenter, target: {}", target);
+        alarmCenter.alarmFailed(jobInstanceAlarm, Lists.newArrayList(target));
+    }
+
+    private String localUrlPath() {
+        return String.format("http://127.0.0.1:%d", port);
+    }
+}

powerjob-server/powerjob-server-starter/src/main/resources/application-daily.properties

@@ -11,8 +11,7 @@ spring.datasource.core.minimum-idle=5
 
 ####### MongoDB properties(Non-core configuration properties)  #######
 ####### delete mongodb config to disable mongodb #######
-#oms.mongodb.enable=true
+#oms.storage.dfs.mongodb.uri=mongodb+srv://zqq:No1Bug2Please3!@cluster0.wie54.gcp.mongodb.net/powerjob_daily?retryWrites=true&w=majority
-#spring.data.mongodb.uri=mongodb+srv://zqq:No1Bug2Please3!@cluster0.wie54.gcp.mongodb.net/powerjob_daily?retryWrites=true&w=majority
 
 ####### Email properties(Non-core configuration properties) #######
 ####### Delete the following code to disable the mail #######

powerjob-server/powerjob-server-starter/src/main/resources/application-pre.properties

@@ -11,8 +11,7 @@ spring.datasource.core.minimum-idle=5
 
 ####### MongoDB properties(Non-core configuration properties)  #######
 ####### delete mongodb config to disable mongodb #######
-oms.mongodb.enable=true
+oms.storage.dfs.mongodb.uri=mongodb://remotehost:27017/powerjob-pre
-spring.data.mongodb.uri=mongodb://remotehost:27017/powerjob-pre
 
 ####### Email properties(Non-core configuration properties) #######
 ####### Delete the following code to disable the mail #######

powerjob-server/powerjob-server-starter/src/main/resources/application-product.properties

@@ -11,8 +11,7 @@ spring.datasource.core.minimum-idle=5
 
 ####### MongoDB properties(Non-core configuration properties)  #######
 ####### delete mongodb config to disable mongodb #######
-oms.mongodb.enable=true
+oms.storage.dfs.mongodb.uri=mongodb://localhost:27017/powerjob-product
-spring.data.mongodb.uri=mongodb://localhost:27017/powerjob-product
 
 ####### Email properties(Non-core configuration properties) #######
 ####### Delete the following code to disable the mail #######