Merge remote-tracking branch 'remotes/upstream/dev-1.1.0' into dev-1.1.0

README.md

@@ -50,7 +50,7 @@ Easy Scheduler
 
 ### 近期研发计划
 
-EasyScheduler的工作计划：<a href="https://github.com/analysys/EasyScheduler/projects/1" target="_blank">研发计划</a> ，其中 In Develop卡片下是1.0.2版本的功能，TODO卡片是待做事项(包括 feature ideas)
+EasyScheduler的工作计划：<a href="https://github.com/analysys/EasyScheduler/projects/1" target="_blank">研发计划</a> ，其中 In Develop卡片下是1.1.0版本的功能，TODO卡片是待做事项(包括 feature ideas)
 
 ### 贡献代码
 

escheduler-api/src/main/java/cn/escheduler/api/controller/ExecutorController.java

@@ -24,7 +24,7 @@ import cn.escheduler.api.utils.Constants;
 import cn.escheduler.api.utils.Result;
 import cn.escheduler.common.enums.*;
 import cn.escheduler.dao.model.User;
-import io.swagger.annotations.Api;
+import io.swagger.annotations.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -38,9 +38,9 @@ import static cn.escheduler.api.enums.Status.*;
 
 
 /**
- * execute task controller
+ * execute process controller
  */
-@ApiIgnore
+@Api(tags = "PROCESS_INSTANCE_EXECUTOR_TAG", position = 1)
 @RestController
 @RequestMapping("projects/{projectName}/executors")
 public class ExecutorController extends BaseController {
@@ -53,10 +53,27 @@ public class ExecutorController extends BaseController {
     /**
      * execute process instance
      */
+    @ApiOperation(value = "startProcessInstance", notes= "RUN_PROCESS_INSTANCE_NOTES")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "processDefinitionId", value = "PROCESS_DEFINITION_ID", required = true, dataType = "Int", example = "100"),
+            @ApiImplicitParam(name = "scheduleTime", value = "SCHEDULE_TIME", required = true, dataType = "String"),
+            @ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", required = true, dataType ="FailureStrategy"),
+            @ApiImplicitParam(name = "startNodeList", value = "START_NODE_LIST", dataType ="String"),
+            @ApiImplicitParam(name = "taskDependType", value = "TASK_DEPEND_TYPE", dataType ="TaskDependType"),
+            @ApiImplicitParam(name = "execType", value = "COMMAND_TYPE", dataType ="CommandType"),
+            @ApiImplicitParam(name = "warningType", value = "WARNING_TYPE",required = true, dataType ="WarningType"),
+            @ApiImplicitParam(name = "warningGroupId", value = "WARNING_GROUP_ID",required = true, dataType ="Int", example = "100"),
+            @ApiImplicitParam(name = "receivers", value = "RECEIVERS",dataType ="String" ),
+            @ApiImplicitParam(name = "receiversCc", value = "RECEIVERS_CC",dataType ="String" ),
+            @ApiImplicitParam(name = "runMode", value = "RUN_MODE",dataType ="RunMode" ),
+            @ApiImplicitParam(name = "processInstancePriority", value = "PROCESS_INSTANCE_PRIORITY", required = true, dataType = "Priority" ),
+            @ApiImplicitParam(name = "workerGroupId", value = "WORKER_GROUP_ID", dataType = "Int",example = "100"),
+            @ApiImplicitParam(name = "timeout", value = "TIMEOUT", dataType = "Int",example = "100"),
+    })
     @PostMapping(value = "start-process-instance")
     @ResponseStatus(HttpStatus.OK)
-    public Result startProcessInstance(@RequestAttribute(value = Constants.SESSION_USER) User loginUser,
-                                       @PathVariable String projectName,
+    public Result startProcessInstance(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+                                       @ApiParam(name = "projectName", value = "PROJECT_NAME", required = true) @PathVariable String projectName,
                                        @RequestParam(value = "processDefinitionId") int processDefinitionId,
                                        @RequestParam(value = "scheduleTime", required = false) String scheduleTime,
                                        @RequestParam(value = "failureStrategy", required = true) FailureStrategy failureStrategy,
@@ -102,10 +119,15 @@ public class ExecutorController extends BaseController {
      * @param processInstanceId
      * @return
      */
+    @ApiOperation(value = "execute", notes= "EXECUTE_ACTION_TO_PROCESS_INSTANCE_NOTES")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "processInstanceId", value = "PROCESS_INSTANCE_ID", required = true, dataType = "Int", example = "100"),
+            @ApiImplicitParam(name = "executeType", value = "EXECUTE_TYPE", required = true, dataType = "ExecuteType")
+    })
     @PostMapping(value = "/execute")
     @ResponseStatus(HttpStatus.OK)
-    public Result execute(@RequestAttribute(value = Constants.SESSION_USER) User loginUser,
-                                                  @PathVariable String projectName,
+    public Result execute(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+                          @ApiParam(name = "projectName", value = "PROJECT_NAME", required = true) @PathVariable String projectName,
                                                   @RequestParam("processInstanceId") Integer processInstanceId,
                                                   @RequestParam("executeType") ExecuteType executeType
     ) {
@@ -127,9 +149,13 @@ public class ExecutorController extends BaseController {
      * @param processDefinitionId
      * @return
      */
+    @ApiOperation(value = "startCheckProcessDefinition", notes= "START_CHECK_PROCESS_DEFINITION_NOTES")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "processDefinitionId", value = "PROCESS_DEFINITION_ID", required = true, dataType = "Int", example = "100")
+    })
     @PostMapping(value = "/start-check")
     @ResponseStatus(HttpStatus.OK)
-    public Result startCheckProcessDefinition(@RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+    public Result startCheckProcessDefinition(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
                              @RequestParam(value = "processDefinitionId") int processDefinitionId) {
         logger.info("login user {}, check process definition", loginUser.getUserName(), processDefinitionId);
         try {
@@ -149,9 +175,16 @@ public class ExecutorController extends BaseController {
      * @param processDefinitionId
      * @return
      */
+    @ApiIgnore
+    @ApiOperation(value = "getReceiverCc", notes= "GET_RECEIVER_CC_NOTES")
+    @ApiImplicitParams({
+            @ApiImplicitParam(name = "processDefinitionId", value = "PROCESS_DEFINITION_ID", required = true, dataType = "Int", example = "100"),
+            @ApiImplicitParam(name = "processInstanceId", value = "PROCESS_INSTANCE_ID", required = true, dataType = "Int", example = "100")
+
+    })
     @GetMapping(value = "/get-receiver-cc")
     @ResponseStatus(HttpStatus.OK)
-    public Result getReceiverCc(@RequestAttribute(value = Constants.SESSION_USER) User loginUser,
+    public Result getReceiverCc(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
                                 @RequestParam(value = "processDefinitionId",required = false) Integer processDefinitionId,
                                 @RequestParam(value = "processInstanceId",required = false) Integer processInstanceId) {
         logger.info("login user {}, get process definition receiver and cc", loginUser.getUserName());

escheduler-api/src/main/java/cn/escheduler/api/service/ExecutorService.java

@@ -191,6 +191,16 @@ public class ExecutorService extends BaseService{
             return checkResult;
         }
 
+        // checkTenantExists();
+        Tenant tenant = processDao.getTenantForProcess(processDefinition.getTenantId(),
+                processDefinition.getUserId());
+        if(tenant == null){
+            logger.error("there is not any vaild tenant for the process definition: id:{},name:{}, ",
+                    processDefinition.getId(), processDefinition.getName());
+            putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, processInstanceId);
+            return result;
+        }
+
         switch (executeType) {
             case REPEAT_RUNNING:
                 result = insertCommand(loginUser, processInstanceId, processDefinition.getId(), CommandType.REPEAT_RUNNING);
@@ -260,7 +270,7 @@ public class ExecutorService extends BaseService{
                 }
                 break;
             case RECOVER_SUSPENDED_PROCESS:
-                if (executionStatus.typeIsPause()) {
+                if (executionStatus.typeIsPause()|| executionStatus.typeIsCancel()) {
                     checkResult = true;
                 }
             default:

escheduler-api/src/main/java/cn/escheduler/api/service/ProcessInstanceService.java

@@ -38,10 +38,7 @@ import cn.escheduler.common.utils.JSONUtils;
 import cn.escheduler.common.utils.ParameterUtils;
 import cn.escheduler.common.utils.placeholder.BusinessTimeUtils;
 import cn.escheduler.dao.ProcessDao;
-import cn.escheduler.dao.mapper.ProcessDefinitionMapper;
-import cn.escheduler.dao.mapper.ProcessInstanceMapper;
-import cn.escheduler.dao.mapper.ProjectMapper;
-import cn.escheduler.dao.mapper.TaskInstanceMapper;
+import cn.escheduler.dao.mapper.*;
 import cn.escheduler.dao.model.*;
 import com.alibaba.fastjson.JSON;
 import org.apache.commons.lang3.StringUtils;
@@ -97,6 +94,9 @@ public class ProcessInstanceService extends BaseDAGService {
     @Autowired
     LoggerService loggerService;
 
+    @Autowired
+    WorkerGroupMapper workerGroupMapper;
+
     /**
      * query process instance by id
      *
@@ -115,6 +115,15 @@ public class ProcessInstanceService extends BaseDAGService {
             return checkResult;
         }
         ProcessInstance processInstance = processDao.findProcessInstanceDetailById(processId);
+        if(processInstance.getWorkerGroupId() == -1){
+            processInstance.setWorkerGroupName(DEFAULT);
+        }else{
+            WorkerGroup workerGroup = workerGroupMapper.queryById(processInstance.getWorkerGroupId());
+            processInstance.setWorkerGroupName(workerGroup.getName());
+        }
+        ProcessDefinition processDefinition = processDao.findProcessDefineById(processInstance.getProcessDefinitionId());
+        processInstance.setReceivers(processDefinition.getReceivers());
+        processInstance.setReceiversCc(processDefinition.getReceiversCc());
         result.put(Constants.DATA_LIST, processInstance);
         putMsg(result, Status.SUCCESS);
 

escheduler-common/src/main/java/cn/escheduler/common/Constants.java

@@ -219,6 +219,11 @@ public final class Constants {
      */
     public static final String SEMICOLON = ";";
 
+    /**
+     * DOT .
+     */
+    public static final String DOT = ".";
+
     /**
      * ZOOKEEPER_SESSION_TIMEOUT
      */
@@ -483,6 +488,8 @@ public final class Constants {
 
     public static final String TASK_RECORD_PWD = "task.record.datasource.password";
 
+    public static final String DEFAULT = "Default";
+
     public static  String TASK_RECORD_TABLE_HIVE_LOG = "eamp_hive_log_hd";
 
     public static  String TASK_RECORD_TABLE_HISTORY_HIVE_LOG = "eamp_hive_hist_log_hd";
@@ -883,6 +890,11 @@ public final class Constants {
      */
     public static final String LOGIN_USER_KEY_TAB_USERNAME = "login.user.keytab.username";
 
+    /**
+     * default worker group id
+     */
+    public static final int DEFAULT_WORKER_ID = -1;
+
     /**
      * loginUserFromKeytab path
      */

escheduler-common/src/main/java/cn/escheduler/common/queue/ITaskQueue.java

@@ -24,20 +24,17 @@ public interface ITaskQueue {
     /**
      * take out all the elements
      *
-     * this method has deprecated
-     * use checkTaskExists instead
      *
      * @param key
      * @return
      */
-    @Deprecated
     List<String> getAllTasks(String key);
 
     /**
      * check task exists in the task queue or not
      *
      * @param key queue name
-     * @param task ${priority}_${processInstanceId}_${taskId}
+     * @param task ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}
      * @return true if exists in the queue
      */
     boolean checkTaskExists(String key, String task);
@@ -54,10 +51,10 @@ public interface ITaskQueue {
      * an element pops out of the queue
      *
      * @param key  queue name
-     * @param remove  whether remove the element
+     * @param n    how many elements to poll
      * @return
      */
-    String poll(String key, boolean remove);
+    List<String> poll(String key, int n);
 
     /**
      * remove a element from queue

escheduler-common/src/main/java/cn/escheduler/common/queue/TaskQueueFactory.java

@@ -42,7 +42,7 @@ public class TaskQueueFactory {
   public static ITaskQueue getTaskQueueInstance() {
     String queueImplValue = CommonUtils.getQueueImplValue();
     if (StringUtils.isNotBlank(queueImplValue)) {
-//      queueImplValue = StringUtils.trim(queueImplValue);
+//      queueImplValue = IpUtils.trim(queueImplValue);
 
 //      if (SCHEDULER_QUEUE_REDIS_IMPL.equals(queueImplValue)) {
 //        logger.info("task queue impl use reids ");

escheduler-common/src/main/java/cn/escheduler/common/queue/TaskQueueZkImpl.java

@@ -19,6 +19,8 @@ package cn.escheduler.common.queue;
 
 import cn.escheduler.common.Constants;
 import cn.escheduler.common.utils.Bytes;
+import cn.escheduler.common.utils.IpUtils;
+import cn.escheduler.common.utils.OSUtils;
 import cn.escheduler.common.zk.AbstractZKClient;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.zookeeper.CreateMode;
@@ -26,10 +28,7 @@ import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 
 /**
  * A singleton of a task queue implemented with zookeeper
@@ -62,7 +61,6 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
      * @param key   task queue name
      * @return
      */
-    @Deprecated
     @Override
     public List<String> getAllTasks(String key) {
         try {
@@ -80,7 +78,7 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
      * check task exists in the task queue or not
      *
      * @param key       queue name
-     * @param task      ${priority}_${processInstanceId}_${taskId}
+     * @param task      ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}
      * @return true if exists in the queue
      */
     @Override
@@ -110,7 +108,7 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
      * add task to tasks queue
      *
      * @param key      task queue name
-     * @param value    ${priority}_${processInstanceId}_${taskId}
+     * @param value    ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}_host1,host2,...
      */
     @Override
     public void add(String key, String value) {
@@ -118,9 +116,6 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
             String taskIdPath = getTasksPath(key) + Constants.SINGLE_SLASH + value;
             String result = getZkClient().create().withMode(CreateMode.PERSISTENT).forPath(taskIdPath, Bytes.toBytes(value));
 
-//            String path = conf.getString(Constants.ZOOKEEPER_SCHEDULER_ROOT) + Constants.SINGLE_SLASH + Constants.SCHEDULER_TASKS_QUEUE + "_add" + Constants.SINGLE_SLASH + value;
-//            getZkClient().create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path,
-//                    Bytes.toBytes(value));
             logger.info("add task : {} to tasks queue , result success",result);
         } catch (Exception e) {
             logger.error("add task to tasks queue exception",e);
@@ -132,16 +127,16 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
     /**
      * An element pops out of the queue <p>
      * note:
-     *   ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}
+     *   ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}_host1,host2,...
      *   The tasks with the highest priority are selected by comparing the priorities of the above four levels from high to low.
      *
-     *   流程实例优先级_流程实例id_任务优先级_任务id       high <- low
+     *   流程优先级_流程实例id_任务优先级_任务id_任务执行的机器id1,任务执行的机器id2,...       high <- low
      * @param  key  task queue name
-     * @param  remove  whether remove the element
-     * @return the task id  to be executed
+     * @param  tasksNum    how many elements to poll
+     * @return the task ids  to be executed
      */
     @Override
-    public String poll(String key, boolean remove) {
+    public List<String> poll(String key, int tasksNum) {
         try{
             CuratorFramework zk = getZkClient();
             String tasksQueuePath = getTasksPath(key) + Constants.SINGLE_SLASH;
@@ -149,55 +144,83 @@ public class TaskQueueZkImpl extends AbstractZKClient implements ITaskQueue {
 
             if(list != null && list.size() > 0){
 
+                String workerIp = OSUtils.getHost();
+                String workerIpLongStr = String.valueOf(IpUtils.ipToLong(workerIp));
+
                 int size = list.size();
 
-                String formatTargetTask = null;
-                String targetTaskKey = null;
+
+                Set<String> taskTreeSet = new TreeSet<>();
+
                 for (int i = 0; i < size; i++) {
+
                     String taskDetail = list.get(i);
                     String[] taskDetailArrs = taskDetail.split(Constants.UNDERLINE);
 
-                    if(taskDetailArrs.length == 4){
+                    //向前版本兼容
+                    if(taskDetailArrs.length >= 4){
+
                         //format ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskId}
                         String formatTask = String.format("%s_%010d_%s_%010d", taskDetailArrs[0], Long.parseLong(taskDetailArrs[1]), taskDetailArrs[2], Long.parseLong(taskDetailArrs[3]));
-                        if(i > 0){
-                            int result = formatTask.compareTo(formatTargetTask);
-                            if(result < 0){
-                                formatTargetTask = formatTask;
-                                targetTaskKey = taskDetail;
+                        if(taskDetailArrs.length > 4){
+                            String taskHosts = taskDetailArrs[4];
+
+                            //task can assign to any worker host if equals default ip value of worker server
+                            if(!taskHosts.equals(String.valueOf(Constants.DEFAULT_WORKER_ID))){
+                                String[] taskHostsArr = taskHosts.split(Constants.COMMA);
+                                if(!Arrays.asList(taskHostsArr).contains(workerIpLongStr)){
+                                    continue;
+                                }
                             }
-                        }else{
-                            formatTargetTask = formatTask;
-                            targetTaskKey = taskDetail;
                         }
-                    }else{
-                        logger.error("task queue poll error, task detail :{} , please check!", taskDetail);
+
+                        taskTreeSet.add(formatTask);
+
                     }
-                }
 
-                if(formatTargetTask != null){
-                    String taskIdPath = tasksQueuePath + targetTaskKey;
+                }
 
-                    logger.info("consume task {}", taskIdPath);
+                List<String> taskslist = getTasksListFromTreeSet(tasksNum, taskTreeSet);
 
-                    String[] vals = targetTaskKey.split(Constants.UNDERLINE);
+                logger.info("consume tasks: {},there still have {} tasks need to be executed", Arrays.toString(taskslist.toArray()), size - taskslist.size());
 
-                    if(remove){
-                        removeNode(key, targetTaskKey);
-                    }
-                    logger.info("consume task: {},there still have {} tasks need to be executed", vals[vals.length - 1], size - 1);
-                    return targetTaskKey;
-                }else{
-                    logger.error("should not go here, task queue poll error, please check!");
-                }
+                return taskslist;
+            }else{
+                Thread.sleep(Constants.SLEEP_TIME_MILLIS);
             }
 
         } catch (Exception e) {
             logger.error("add task to tasks queue exception",e);
         }
-        return null;
+        return new ArrayList<String>();
     }
 
+
+    /**
+     * get task list from tree set
+     *
+     * @param tasksNum
+     * @param taskTreeSet
+     */
+    public List<String> getTasksListFromTreeSet(int tasksNum, Set<String> taskTreeSet) {
+        Iterator<String> iterator = taskTreeSet.iterator();
+        int j = 0;
+        List<String> taskslist = new ArrayList<>(tasksNum);
+        while(iterator.hasNext()){
+            if(j++ < tasksNum){
+                String task = iterator.next();
+                String[] taskArray = task.split(Constants.UNDERLINE);
+                int processInstanceId = Integer.parseInt(taskArray[1]);
+                int taskId = Integer.parseInt(taskArray[3]);
+                String destTask = taskArray[0]+Constants.UNDERLINE + processInstanceId + Constants.UNDERLINE
+                        + taskArray[2] + Constants.UNDERLINE + taskId;
+                taskslist.add(destTask);
+            }
+        }
+        return taskslist;
+    }
+
+
     @Override
     public void removeNode(String key, String nodeValue){
 

escheduler-common/src/main/java/cn/escheduler/common/utils/IpUtils.java

+/*
+ * 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 cn.escheduler.common.utils;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * http utils
+ */
+public class IpUtils {
+
+  private static final Logger logger = LoggerFactory.getLogger(IpUtils.class);
+  public static final String DOT = ".";
+
+  /**
+   * ip str to long <p>
+   *
+   * @param ipStr ip string
+   */
+  public static Long ipToLong(String ipStr) {
+    String[] ipSet = ipStr.split("\\" + DOT);
+
+    return Long.parseLong(ipSet[0]) << 24 | Long.parseLong(ipSet[1]) << 16 | Long.parseLong(ipSet[2]) << 8 | Long.parseLong(ipSet[3]);
+  }
+
+  /**
+   * long to ip
+   * @param ipLong the long number converted from IP
+   * @return String
+   */
+  public static String longToIp(long ipLong) {
+    long[] ipNumbers = new long[4];
+    long tmp = 0xFF;
+    ipNumbers[0] = ipLong >> 24 & tmp;
+    ipNumbers[1] = ipLong >> 16 & tmp;
+    ipNumbers[2] = ipLong >> 8 & tmp;
+    ipNumbers[3] = ipLong & tmp;
+
+    StringBuilder sb = new StringBuilder(16);
+    sb.append(ipNumbers[0]).append(DOT)
+            .append(ipNumbers[1]).append(DOT)
+            .append(ipNumbers[2]).append(DOT)
+            .append(ipNumbers[3]);
+    return sb.toString();
+  }
+
+
+
+  public static void main(String[] args){
+    long ipLong = ipToLong("11.3.4.5");
+    logger.info(longToIp(ipLong));
+  }
+}

escheduler-common/src/main/java/cn/escheduler/common/zk/AbstractZKClient.java

@@ -312,7 +312,11 @@ public abstract class AbstractZKClient {
 				childrenList = zkClient.getChildren().forPath(masterZNodeParentPath);
 			}
 		} catch (Exception e) {
-			logger.warn(e.getMessage(),e);
+//			logger.warn(e.getMessage());
+			if(!e.getMessage().contains("java.lang.IllegalStateException: instance must be started")){
+				logger.warn(e.getMessage(),e);
+			}
+
 			return childrenList.size();
 		}
 		return childrenList.size();

escheduler-common/src/test/java/cn/escheduler/common/os/OSUtilsTest.java

@@ -37,6 +37,12 @@ public class OSUtilsTest {
 //    static HardwareAbstractionLayer hal = si.getHardware();
 
 
+    @Test
+    public void getHost(){
+        logger.info(OSUtils.getHost());
+    }
+
+
     @Test
     public void memoryUsage() {
         logger.info("memoryUsage : {}", OSUtils.memoryUsage());// 0.3361799418926239

escheduler-common/src/test/java/cn/escheduler/common/queue/TaskQueueImplTest.java

@@ -17,12 +17,15 @@
 package cn.escheduler.common.queue;
 
 import cn.escheduler.common.Constants;
+import org.junit.After;
 import org.junit.Assert;
+import org.junit.Before;
 import org.junit.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Arrays;
+import java.util.List;
 import java.util.Random;
 
 import static org.junit.Assert.assertEquals;
@@ -34,59 +37,62 @@ public class TaskQueueImplTest {
 
     private static final Logger logger = LoggerFactory.getLogger(TaskQueueImplTest.class);
 
+    ITaskQueue tasksQueue = null;
 
-    @Test
-    public void testTaskQueue(){
+    @Before
+    public void before(){
+        tasksQueue = TaskQueueFactory.getTaskQueueInstance();
+        //clear all data
+        tasksQueue.delete();
 
-        ITaskQueue tasksQueue = TaskQueueFactory.getTaskQueueInstance();
+    }
+
+
+    @After
+    public void after(){
         //clear all data
         tasksQueue.delete();
+    }
+
+
+    @Test
+    public void testAdd(){
 
         //add
-        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"1");
-        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"2");
-        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"3");
-        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"4");
+        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"1_1_1_1_2130706433,3232236775");
+        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"0_1_1_1_2130706433,3232236775");
+        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"1_1_0_1_2130706433,3232236775");
+        tasksQueue.add(Constants.SCHEDULER_TASKS_QUEUE,"1_2_1_1_2130706433,3232236775");
+
+        List<String> tasks = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, 1);
+
+        if(tasks.size() < 0){
+            return;
+        }
 
         //pop
-        String node1 = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, false);
-        assertEquals(node1,"1");
-        String node2 = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, false);
-        assertEquals(node2,"2");
-
-        //sadd
-        String task1 = "1.1.1.1-1-mr";
-        String task2 = "1.1.1.2-2-mr";
-        String task3 = "1.1.1.3-3-mr";
-        String task4 = "1.1.1.4-4-mr";
-        String task5 = "1.1.1.5-5-mr";
-
-        tasksQueue.sadd(Constants.SCHEDULER_TASKS_KILL,task1);
-        tasksQueue.sadd(Constants.SCHEDULER_TASKS_KILL,task2);
-        tasksQueue.sadd(Constants.SCHEDULER_TASKS_KILL,task3);
-        tasksQueue.sadd(Constants.SCHEDULER_TASKS_KILL,task4);
-        tasksQueue.sadd(Constants.SCHEDULER_TASKS_KILL,task5);
-        tasksQueue.sadd(Constants.SCHEDULER_TASKS_KILL,task5); //repeat task
-
-        Assert.assertEquals(tasksQueue.smembers(Constants.SCHEDULER_TASKS_KILL).size(),5);
-        logger.info(Arrays.toString(tasksQueue.smembers(Constants.SCHEDULER_TASKS_KILL).toArray()));
-        //srem
-        tasksQueue.srem(Constants.SCHEDULER_TASKS_KILL,task5);
-        //smembers
-        Assert.assertEquals(tasksQueue.smembers(Constants.SCHEDULER_TASKS_KILL).size(),4);
-        logger.info(Arrays.toString(tasksQueue.smembers(Constants.SCHEDULER_TASKS_KILL).toArray()));
+        String node1 = tasks.get(0);
 
+        assertEquals(node1,"0_0000000001_1_0000000001");
+
+        tasks = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, 1);
+
+        if(tasks.size() < 0){
+            return;
+        }
+
+        String node2 = tasks.get(0);
+        assertEquals(node2,"0_0000000001_1_0000000001");
 
     }
 
+
+
     /**
      * test one million data from zookeeper queue
      */
     @Test
     public void extremeTest(){
-        ITaskQueue tasksQueue = TaskQueueFactory.getTaskQueueInstance();
-        //clear all data
-        tasksQueue.delete();
         int total = 30 * 10000;
 
         for(int i = 0; i < total; i++)
@@ -99,14 +105,9 @@ public class TaskQueueImplTest {
             }
         }
 
-        String node1 = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, false);
+        String node1 = tasksQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, 1).get(0);
         assertEquals(node1,"0");
 
-        //clear all data
-        tasksQueue.delete();
-
-
-
     }
 
 }

escheduler-dao/src/main/java/cn/escheduler/dao/ProcessDao.java

@@ -25,6 +25,7 @@ import cn.escheduler.common.queue.ITaskQueue;
 import cn.escheduler.common.queue.TaskQueueFactory;
 import cn.escheduler.common.task.subprocess.SubProcessParameters;
 import cn.escheduler.common.utils.DateUtils;
+import cn.escheduler.common.utils.IpUtils;
 import cn.escheduler.common.utils.JSONUtils;
 import cn.escheduler.common.utils.ParameterUtils;
 import cn.escheduler.dao.mapper.*;
@@ -117,7 +118,7 @@ public class ProcessDao extends AbstractBaseDao {
      */
     @Override
     protected void init() {
-        userMapper=getMapper(UserMapper.class);
+        userMapper = getMapper(UserMapper.class);
         processDefineMapper = getMapper(ProcessDefinitionMapper.class);
         processInstanceMapper = getMapper(ProcessInstanceMapper.class);
         dataSourceMapper = getMapper(DataSourceMapper.class);
@@ -1015,11 +1016,58 @@ public class ProcessDao extends AbstractBaseDao {
      *
      * 流程实例优先级_流程实例id_任务优先级_任务id       high <- low
      *
-     * @param task
+     * @param taskInstance
      * @return
      */
-    private String taskZkInfo(TaskInstance task) {
-        return String.valueOf(task.getProcessInstancePriority().ordinal()) + Constants.UNDERLINE + task.getProcessInstanceId() + Constants.UNDERLINE + task.getTaskInstancePriority().ordinal() + Constants.UNDERLINE + task.getId();
+    private String taskZkInfo(TaskInstance taskInstance) {
+
+        int taskWorkerGroupId = getTaskWorkerGroupId(taskInstance);
+
+        StringBuilder sb = new StringBuilder(100);
+
+        sb.append(taskInstance.getProcessInstancePriority().ordinal()).append(Constants.UNDERLINE)
+                .append(taskInstance.getProcessInstanceId()).append(Constants.UNDERLINE)
+                .append(taskInstance.getTaskInstancePriority().ordinal()).append(Constants.UNDERLINE)
+                .append(taskInstance.getId()).append(Constants.UNDERLINE);
+
+        if(taskWorkerGroupId > 0){
+            //not to find data from db
+            WorkerGroup workerGroup = queryWorkerGroupById(taskWorkerGroupId);
+            if(workerGroup == null ){
+                logger.info("task {} cannot find the worker group, use all worker instead.", taskInstance.getId());
+
+                sb.append(Constants.DEFAULT_WORKER_ID);
+                return sb.toString();
+            }
+
+            String ips = workerGroup.getIpList();
+
+            if(StringUtils.isBlank(ips)){
+                logger.error("task:{} worker group:{} parameters(ip_list) is null, this task would be running on all workers",
+                        taskInstance.getId(), workerGroup.getId());
+                sb.append(Constants.DEFAULT_WORKER_ID);
+                return sb.toString();
+            }
+
+            StringBuilder ipSb = new StringBuilder(100);
+            String[] ipArray = ips.split(COMMA);
+
+            for (String ip : ipArray) {
+               long ipLong = IpUtils.ipToLong(ip);
+                ipSb.append(ipLong).append(COMMA);
+            }
+
+            if(ipSb.length() > 0) {
+                ipSb.deleteCharAt(ipSb.length() - 1);
+            }
+
+            sb.append(ipSb);
+        }else{
+            sb.append(Constants.DEFAULT_WORKER_ID);
+        }
+
+
+        return  sb.toString();
     }
 
     /**
@@ -1683,5 +1731,24 @@ public class ProcessDao extends AbstractBaseDao {
     }
 
 
+    /**
+     * get task worker group id
+     *
+     * @param taskInstance
+     * @return
+     */
+    public int getTaskWorkerGroupId(TaskInstance taskInstance) {
+        int taskWorkerGroupId = taskInstance.getWorkerGroupId();
+        ProcessInstance processInstance = findProcessInstanceByTaskId(taskInstance.getId());
+        if(processInstance == null){
+            logger.error("cannot find the task:{} process instance", taskInstance.getId());
+            return Constants.DEFAULT_WORKER_ID;
+        }
+        int processWorkerGroupId = processInstance.getWorkerGroupId();
+
+        taskWorkerGroupId = (taskWorkerGroupId <= 0 ? processWorkerGroupId : taskWorkerGroupId);
+        return taskWorkerGroupId;
+    }
+
 
 }

escheduler-dao/src/main/java/cn/escheduler/dao/model/ProcessInstance.java

@@ -194,6 +194,21 @@ public class ProcessInstance {
      */
     private int tenantId;
 
+    /**
+     * worker group name. for api.
+     */
+    private String workerGroupName;
+
+    /**
+     * receivers for api
+     */
+    private String receivers;
+
+    /**
+     * receivers cc for api
+     */
+    private String receiversCc;
+
     public ProcessInstance(){
 
     }
@@ -560,4 +575,28 @@ public class ProcessInstance {
     public int getTenantId() {
         return this.tenantId ;
     }
+
+    public String getWorkerGroupName() {
+        return workerGroupName;
+    }
+
+    public void setWorkerGroupName(String workerGroupName) {
+        this.workerGroupName = workerGroupName;
+    }
+
+    public String getReceivers() {
+        return receivers;
+    }
+
+    public void setReceivers(String receivers) {
+        this.receivers = receivers;
+    }
+
+    public String getReceiversCc() {
+        return receiversCc;
+    }
+
+    public void setReceiversCc(String receiversCc) {
+        this.receiversCc = receiversCc;
+    }
 }

escheduler-server/pom.xml

@@ -89,7 +89,7 @@
 			<artifactId>escheduler-alert</artifactId>
 		</dependency>
 
-	</dependencies>
+    </dependencies>
 
 
 	<build>

escheduler-server/src/main/java/cn/escheduler/server/master/MasterServer.java

@@ -216,7 +216,7 @@ public class MasterServer implements CommandLineRunner, IStoppable {
                 if(Stopper.isRunning()) {
                     // send heartbeat to zk
                     if (StringUtils.isBlank(zkMasterClient.getMasterZNode())) {
-                        logger.error("master send heartbeat to zk failed");
+                        logger.error("master send heartbeat to zk failed: can't find zookeeper regist path of master server");
                         return;
                     }
 

escheduler-server/src/main/java/cn/escheduler/server/worker/runner/FetchTaskThread.java

@@ -25,8 +25,9 @@ import cn.escheduler.common.utils.OSUtils;
 import cn.escheduler.dao.ProcessDao;
 import cn.escheduler.dao.model.*;
 import cn.escheduler.server.zk.ZKWorkerClient;
-import com.cronutils.utils.StringUtils;
 import org.apache.commons.configuration.Configuration;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.recipes.locks.InterProcessMutex;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -98,15 +99,7 @@ public class FetchTaskThread implements Runnable{
      */
     private boolean checkWorkerGroup(TaskInstance taskInstance, String host){
 
-        int taskWorkerGroupId = taskInstance.getWorkerGroupId();
-        ProcessInstance processInstance = processDao.findProcessInstanceByTaskId(taskInstance.getId());
-        if(processInstance == null){
-            logger.error("cannot find the task:{} process instance", taskInstance.getId());
-            return false;
-        }
-        int processWorkerGroupId = processInstance.getWorkerGroupId();
-
-        taskWorkerGroupId = (taskWorkerGroupId <= 0 ? processWorkerGroupId : taskWorkerGroupId);
+        int taskWorkerGroupId = processDao.getTaskWorkerGroupId(taskInstance);
 
         if(taskWorkerGroupId <= 0){
             return true;
@@ -117,99 +110,103 @@ public class FetchTaskThread implements Runnable{
             return true;
         }
         String ips = workerGroup.getIpList();
-        if(ips == null){
+        if(StringUtils.isBlank(ips)){
             logger.error("task:{} worker group:{} parameters(ip_list) is null, this task would be running on all workers",
                     taskInstance.getId(), workerGroup.getId());
         }
-        String[] ipArray = ips.split(",");
+        String[] ipArray = ips.split(Constants.COMMA);
         List<String> ipList =  Arrays.asList(ipArray);
         return ipList.contains(host);
     }
 
 
+
+
     @Override
     public void run() {
 
         while (Stopper.isRunning()){
             InterProcessMutex mutex = null;
             try {
-                if(OSUtils.checkResource(this.conf, false)) {
 
-                    // creating distributed locks, lock path /escheduler/lock/worker
-                    String zNodeLockPath = zkWorkerClient.getWorkerLockPath();
-                    mutex = new InterProcessMutex(zkWorkerClient.getZkClient(), zNodeLockPath);
-                    mutex.acquire();
+                ThreadPoolExecutor poolExecutor = (ThreadPoolExecutor) workerExecService;
 
-                    ThreadPoolExecutor poolExecutor = (ThreadPoolExecutor) workerExecService;
+                //check memory and cpu usage and threads
+                if(OSUtils.checkResource(this.conf, false) && checkThreadCount(poolExecutor)) {
 
-                    for (int i = 0; i < taskNum; i++) {
-
-                        int activeCount = poolExecutor.getActiveCount();
-                        if (activeCount >= workerExecNums) {
-                            logger.info("thread insufficient , activeCount : {} , workerExecNums : {}",activeCount,workerExecNums);
-                            continue;
-                        }
+                    //whether have tasks, if no tasks , no need lock  //get all tasks
+                    List<String> tasksQueueList = taskQueue.getAllTasks(Constants.SCHEDULER_TASKS_QUEUE);
+                    if(tasksQueueList.size() > 0){
+                        // creating distributed locks, lock path /escheduler/lock/worker
+                        String zNodeLockPath = zkWorkerClient.getWorkerLockPath();
+                        mutex = new InterProcessMutex(zkWorkerClient.getZkClient(), zNodeLockPath);
+                        mutex.acquire();
 
                         // task instance id str
-                        String taskQueueStr = taskQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, false);
+                        List<String> taskQueueStrArr = taskQueue.poll(Constants.SCHEDULER_TASKS_QUEUE, taskNum);
 
-                        if (!StringUtils.isEmpty(taskQueueStr )) {
+                        for(String taskQueueStr : taskQueueStrArr){
+                            if (StringUtils.isNotBlank(taskQueueStr )) {
 
-                            String[] taskStringArray = taskQueueStr.split(Constants.UNDERLINE);
-                            String taskInstIdStr = taskStringArray[taskStringArray.length - 1];
-                            Date now = new Date();
-                            Integer taskId = Integer.parseInt(taskInstIdStr);
+                                if (!checkThreadCount(poolExecutor)) {
+                                    break;
+                                }
 
-                            // find task instance by task id
-                            TaskInstance taskInstance = processDao.findTaskInstanceById(taskId);
+                                String[] taskStringArray = taskQueueStr.split(Constants.UNDERLINE);
+                                String taskInstIdStr = taskStringArray[taskStringArray.length - 1];
+                                Date now = new Date();
+                                Integer taskId = Integer.parseInt(taskInstIdStr);
 
-                            logger.info("worker fetch taskId : {} from queue ", taskId);
+                                // find task instance by task id
+                                TaskInstance taskInstance = processDao.findTaskInstanceById(taskId);
 
-                            int retryTimes = 30;
-                            // mainly to wait for the master insert task to succeed
-                            while (taskInstance == null && retryTimes > 0) {
-                                Thread.sleep(Constants.SLEEP_TIME_MILLIS);
-                                taskInstance = processDao.findTaskInstanceById(taskId);
-                                retryTimes--;
-                            }
+                                logger.info("worker fetch taskId : {} from queue ", taskId);
 
-                            if (taskInstance == null ) {
-                                logger.error("task instance is null. task id : {} ", taskId);
-                                continue;
-                            }
-                            if(!checkWorkerGroup(taskInstance, OSUtils.getHost())){
-                                continue;
-                            }
-                            taskQueue.removeNode(Constants.SCHEDULER_TASKS_QUEUE, taskQueueStr);
-                            logger.info("remove task:{} from queue", taskQueueStr);
+                                int retryTimes = 30;
+                                // mainly to wait for the master insert task to succeed
+                                while (taskInstance == null && retryTimes > 0) {
+                                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
+                                    taskInstance = processDao.findTaskInstanceById(taskId);
+                                    retryTimes--;
+                                }
+
+                                if (taskInstance == null ) {
+                                    logger.error("task instance is null. task id : {} ", taskId);
+                                    continue;
+                                }
 
-                            // set execute task worker host
-                            taskInstance.setHost(OSUtils.getHost());
-                            taskInstance.setStartTime(now);
+                                if(!checkWorkerGroup(taskInstance, OSUtils.getHost())){
+                                    continue;
+                                }
+                                taskQueue.removeNode(Constants.SCHEDULER_TASKS_QUEUE, taskQueueStr);
+                                logger.info("remove task:{} from queue", taskQueueStr);
 
+                                // set execute task worker host
+                                taskInstance.setHost(OSUtils.getHost());
+                                taskInstance.setStartTime(now);
 
-                            // get process instance
-                            ProcessInstance processInstance = processDao.findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
 
+                                // get process instance
+                                ProcessInstance processInstance = processDao.findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
 
-                            // get process define
-                            ProcessDefinition processDefine = processDao.findProcessDefineById(taskInstance.getProcessDefinitionId());
+                                // get process define
+                                ProcessDefinition processDefine = processDao.findProcessDefineById(taskInstance.getProcessDefinitionId());
 
 
-                            taskInstance.setProcessInstance(processInstance);
-                            taskInstance.setProcessDefine(processDefine);
+                                taskInstance.setProcessInstance(processInstance);
+                                taskInstance.setProcessDefine(processDefine);
 
 
-                            // get local execute path
-                            String execLocalPath = FileUtils.getProcessExecDir(processDefine.getProjectId(),
-                                    processDefine.getId(),
-                                    processInstance.getId(),
-                                    taskInstance.getId());
-                            logger.info("task instance  local execute path : {} ", execLocalPath);
+                                // get local execute path
+                                String execLocalPath = FileUtils.getProcessExecDir(processDefine.getProjectId(),
+                                        processDefine.getId(),
+                                        processInstance.getId(),
+                                        taskInstance.getId());
+                                logger.info("task instance  local execute path : {} ", execLocalPath);
 
 
-                            // set task execute path
-                            taskInstance.setExecutePath(execLocalPath);
+                                // set task execute path
+                                taskInstance.setExecutePath(execLocalPath);
 
                             Tenant tenant = processDao.getTenantForProcess(processInstance.getTenantId(),
                                     processDefine.getUserId());
@@ -218,21 +215,22 @@ public class FetchTaskThread implements Runnable{
                             FileUtils.createWorkDirAndUserIfAbsent(execLocalPath,
                                     tenant.getTenantCode(), logger);
 
-                            logger.info("task : {} ready to submit to task scheduler thread",taskId);
-                            // submit task
-                            workerExecService.submit(new TaskScheduleThread(taskInstance, processDao));
+                                logger.info("task : {} ready to submit to task scheduler thread",taskId);
+                                // submit task
+                                workerExecService.submit(new TaskScheduleThread(taskInstance, processDao));
 
+                            }
                         }
 
                     }
+
                 }
 
                 Thread.sleep(Constants.SLEEP_TIME_MILLIS);
 
             }catch (Exception e){
                 logger.error("fetch task thread exception : " + e.getMessage(),e);
-            }
-            finally {
+            }finally {
                 if (mutex != null){
                     try {
                         mutex.release();
@@ -247,4 +245,18 @@ public class FetchTaskThread implements Runnable{
             }
         }
     }
+
+    /**
+     *
+     * @param poolExecutor
+     * @return
+     */
+    private boolean checkThreadCount(ThreadPoolExecutor poolExecutor) {
+        int activeCount = poolExecutor.getActiveCount();
+        if (activeCount >= workerExecNums) {
+            logger.info("thread insufficient , activeCount : {} , workerExecNums : {}, will sleep : {} millis for thread resource", activeCount, workerExecNums, Constants.SLEEP_TIME_MILLIS);
+            return false;
+        }
+        return true;
+    }
 }
\ No newline at end of file

escheduler-server/src/main/java/cn/escheduler/server/worker/task/sql/SqlTask.java

@@ -387,7 +387,7 @@ public class SqlTask extends AbstractTask {
         String showTypeName = sqlParameters.getShowType().replace(Constants.COMMA,"").trim();
         if(EnumUtils.isValidEnum(ShowType.class,showTypeName)){
             Map<String, Object> mailResult = MailUtils.sendMails(receviersList, receviersCcList, title, content, ShowType.valueOf(showTypeName));
-            if(!(Boolean) mailResult.get(Constants.STATUS)){
+            if(!(Boolean) mailResult.get(cn.escheduler.common.Constants.STATUS)){
                 throw new RuntimeException("send mail failed!");
             }
         }else{

escheduler-ui/src/js/conf/home/pages/dag/_source/startingParam/index.vue

@@ -85,8 +85,6 @@
         deep: true,
         handler () {
           this.isActive = false
-          this.notifyGroupList = null
-          this.workerGroupList = null
           this.$nextTick(() => (this.isActive = true))
         }
       }

escheduler-ui/src/js/conf/home/pages/monitor/pages/servers/master.vue

@@ -6,7 +6,7 @@
           <div class="row-title">
             <div class="left">
               <span class="sp">IP: {{item.host}}</span>
-              <span class="sp">{{$t('Port')}}: {{item.port}}</span>
+              <span class="sp">{{$t('Process Pid')}}: {{item.port}}</span>
               <span class="sp">{{$t('Zk registration directory')}}: {{item.zkDirectory}}</span>
             </div>
             <div class="right">
@@ -93,4 +93,4 @@
 </script>
 <style lang="scss" rel="stylesheet/scss">
   @import "./servers";
-</style>
+</style>
\ No newline at end of file

escheduler-ui/src/js/conf/home/pages/monitor/pages/servers/worker.vue

@@ -6,7 +6,7 @@
           <div class="row-title">
             <div class="left">
               <span class="sp">IP: {{item.host}}</span>
-              <span class="sp">{{$t('Port')}}: {{item.port}}</span>
+              <span class="sp">{{$t('Process Pid')}}: {{item.port}}</span>
               <span class="sp">{{$t('Zk registration directory')}}: {{item.zkDirectory}}</span>
             </div>
             <div class="right">
@@ -94,4 +94,4 @@
 </script>
 <style lang="scss" rel="stylesheet/scss">
   @import "./servers";
-</style>
+</style>
\ No newline at end of file

escheduler-ui/src/js/conf/home/pages/projects/pages/definition/pages/list/_source/util.js

@@ -37,7 +37,7 @@ let warningTypeList = [
 ]
 
 const isEmial = (val) => {
-  let regEmail = /^([a-zA-Z0-9]+[_|\_|\.]?)*[a-zA-Z0-9]+@([a-zA-Z0-9]+[_|\_|\.]?)*[a-zA-Z0-9]+\.[a-zA-Z]{2,3}$/ // eslint-disable-line
+  let regEmail = /^([a-zA-Z0-9]+[_|\-|\.]?)*[a-zA-Z0-9]+@([a-zA-Z0-9]+[_|\-|\.]?)*[a-zA-Z0-9]+\.[a-zA-Z]{2,3}$/ // eslint-disable-line
   return regEmail.test(val)
 }
 

escheduler-ui/src/js/conf/home/pages/projects/pages/instance/pages/list/_source/list.vue

@@ -95,17 +95,17 @@
                         shape="circle"
                         size="xsmall"
                         data-toggle="tooltip"
-                        :title="$t('Stop')"
-                        @click="_stop(item)"
-                        icon="iconfont icon-zanting1"
-                        :disabled="item.state !== 'RUNNING_EXEUTION'"></x-button>
+                        :title="item.state === 'STOP' ? $t('Recovery Suspend') : $t('Stop')"
+                        @click="_stop(item,$index)"
+                        :icon="item.state === 'STOP' ? 'iconfont icon-ai06' : 'iconfont icon-zanting'"
+                        :disabled="item.state !== 'RUNNING_EXEUTION' && item.state != 'STOP'"></x-button>
               <x-button type="warning"
                         shape="circle"
                         size="xsmall"
                         data-toggle="tooltip"
                         :title="item.state === 'PAUSE' ? $t('Recovery Suspend') : $t('Pause')"
                         @click="_suspend(item,$index)"
-                        :icon="item.state === 'PAUSE' ? 'iconfont icon-ai06' : 'iconfont icon-zanting'"
+                        :icon="item.state === 'PAUSE' ? 'iconfont icon-ai06' : 'iconfont icon-zanting1'"
                         :disabled="item.state !== 'RUNNING_EXEUTION' && item.state !== 'PAUSE'"></x-button>
               <x-poptip
                       :ref="'poptip-delete-' + $index"
@@ -155,7 +155,7 @@
                       shape="circle"
                       size="xsmall"
                       disabled="true">
-                {{item.count}}s
+                {{item.count}}
               </x-button>
               <x-button
                       v-show="buttonType !== 'run'"
@@ -173,7 +173,7 @@
                       shape="circle"
                       size="xsmall"
                       disabled="true">
-                {{item.count}}s
+                {{item.count}}
               </x-button>
               <x-button
                       v-show="buttonType !== 'store'"
@@ -185,26 +185,26 @@
               </x-button>
 
               <!--Stop-->
-              <x-button
-                      type="error"
-                      shape="circle"
-                      size="xsmall"
-                      icon="iconfont icon-zanting1"
-                      disabled="true">
-              </x-button>
+              <!--<x-button-->
+                      <!--type="error"-->
+                      <!--shape="circle"-->
+                      <!--size="xsmall"-->
+                      <!--icon="iconfont icon-zanting1"-->
+                      <!--disabled="true">-->
+              <!--</x-button>-->
 
               <!--倒计时 => Recovery Suspend/Pause-->
               <x-button
-                      v-show="item.state === 'PAUSE' && buttonType === 'suspend'"
+                      v-show="(item.state === 'PAUSE' || item.state == 'STOP') && buttonType === 'suspend'"
                       type="warning"
                       shape="circle"
                       size="xsmall"
                       disabled="true">
-                {{item.count}}s
+                {{item.count}}
               </x-button>
               <!--Recovery Suspend-->
               <x-button
-                      v-show="item.state === 'PAUSE' && buttonType !== 'suspend'"
+                      v-show="(item.state === 'PAUSE' || item.state == 'STOP') && buttonType !== 'suspend'"
                       type="warning"
                       shape="circle"
                       size="xsmall"
@@ -217,6 +217,15 @@
                       type="warning"
                       shape="circle"
                       size="xsmall"
+                      icon="iconfont icon-zanting1"
+                      disabled="true">
+              </x-button>
+            <!--Stop-->
+              <x-button
+                      v-show="item.state !== 'STOP'"
+                      type="warning"
+                      shape="circle"
+                      size="xsmall"
                       icon="iconfont icon-zanting"
                       disabled="true">
               </x-button>
@@ -362,11 +371,20 @@
        * stop
        * @param STOP
        */
-      _stop (item) {
-        this._upExecutorsState({
-          processInstanceId: item.id,
-          executeType: 'STOP'
-        })
+      _stop (item, index) {
+        if(item.state == 'STOP') {
+          this._countDownFn({
+            id: item.id,
+            executeType: 'RECOVER_SUSPENDED_PROCESS',
+            index: index,
+            buttonType: 'suspend'
+          })
+        } else {
+          this._upExecutorsState({
+            processInstanceId: item.id,
+            executeType: 'STOP'
+          })
+        }
       },
       /**
        * pause
@@ -383,7 +401,7 @@
         } else {
           this._upExecutorsState({
             processInstanceId: item.id,
-            executeType: item.state === 'PAUSE' ? 'RECOVER_SUSPENDED_PROCESS' : 'PAUSE'
+            executeType: 'PAUSE'
           })
         }
       },
@@ -435,7 +453,7 @@
         if (data.length) {
           _.map(data, v => {
             v.disabled = true
-            v.count = 10
+            v.count = 9
           })
         }
         return data

escheduler-ui/src/js/conf/home/pages/security/pages/users/_source/createUser.vue

@@ -131,7 +131,7 @@
         }
       },
       _verification () {
-        let regEmail = /^([a-zA-Z0-9]+[_|\_|\.]?)*[a-zA-Z0-9]+@([a-zA-Z0-9]+[_|\_|\.]?)*[a-zA-Z0-9]+\.[a-zA-Z]{2,3}$/ // eslint-disable-line
+        let regEmail = /^([a-zA-Z0-9]+[_|\-|\.]?)*[a-zA-Z0-9]+@([a-zA-Z0-9]+[_|\-|\.]?)*[a-zA-Z0-9]+\.[a-zA-Z]{2,3}$/ // eslint-disable-line
         // Mobile phone number regular
         let regPhone = /^1(3|4|5|6|7|8)\d{9}$/; // eslint-disable-line
 

escheduler-ui/src/js/module/i18n/locale/zh_CN.js

@@ -237,7 +237,7 @@ export default {
   'Recovery Failed': '恢复失败',
   'Stop': '停止',
   'Pause': '暂停',
-  'Recovery Suspend': '恢复暂停',
+  'Recovery Suspend': '恢复运行',
   'Gantt': '甘特图',
   'Name': '名称',
   'Node Type': '节点类型',
@@ -282,7 +282,7 @@ export default {
   'Start Process': '启动工作流',
   'Execute from the current node': '从当前节点开始执行',
   'Recover tolerance fault process': '恢复被容错的工作流',
-  'Resume the suspension process': '恢复暂停流程',
+  'Resume the suspension process': '恢复运行流程',
   'Execute from the failed nodes': '从失败节点开始执行',
   'Complement Data': '补数',
   'Scheduling execution': '调度执行',

install.sh

@@ -330,7 +330,7 @@ sed -i ${txt} "s#master.exec.task.number.*#master.exec.task.number=${masterExecT
 sed -i ${txt} "s#master.heartbeat.interval.*#master.heartbeat.interval=${masterHeartbeatInterval}#g" conf/master.properties
 sed -i ${txt} "s#master.task.commit.retryTimes.*#master.task.commit.retryTimes=${masterTaskCommitRetryTimes}#g" conf/master.properties
 sed -i ${txt} "s#master.task.commit.interval.*#master.task.commit.interval=${masterTaskCommitInterval}#g" conf/master.properties
-sed -i ${txt} "s#master.max.cpuload.avg.*#master.max.cpuload.avg=${masterMaxCpuLoadAvg}#g" conf/master.properties
+#sed -i ${txt} "s#master.max.cpuload.avg.*#master.max.cpuload.avg=${masterMaxCpuLoadAvg}#g" conf/master.properties
 sed -i ${txt} "s#master.reserved.memory.*#master.reserved.memory=${masterReservedMemory}#g" conf/master.properties