This is an automated email from the ASF dual-hosted git repository.
lidongdai pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 0d065c0 [Feature-5062][*] Create/Edit/Delete Worker Group in Web UI
(#5113)
0d065c0 is described below
commit 0d065c07125d66f9232cd59611c5f531683bdf82
Author: Shiwen Cheng <[email protected]>
AuthorDate: Tue Mar 23 23:47:41 2021 +0800
[Feature-5062][*] Create/Edit/Delete Worker Group in Web UI (#5113)
* [Improvement][Server] Rename MASTER_PREFIX to MASTER_TYPE and
WORKER_PREFIX to WORKER_TYPE
* [Feature][Api&Dao] Support api and dao of worker group
* [Feature][UI] Support to create, edit and delete worker group
* [Feature][API&UI] Support work groups from database and zookeeper to
display together in worker group management
* [Feature][API&Dao] Support long ip list and check request parameters when
saving worker group
* [Feature][*] Refactor ipList to addrList
* [Feature][Server] Support host manager to select worker host by worker
group in database
* [Improvement][Server] Improve the config of master and worker
* [Improvement][*] Rename zkRegistered to systemDefault
* [Improvement][Test] Fix unit test and improve ui
* [Improvement][Server] Improve getServerMaps
Co-authored-by: dailidong <[email protected]>
---
.../conf/dolphinscheduler/master.properties.tpl | 2 +-
.../conf/dolphinscheduler/worker.properties.tpl | 10 +-
docker/kubernetes/dolphinscheduler/README.md | 6 +-
.../api/controller/WorkerGroupController.java | 74 ++++++--
.../apache/dolphinscheduler/api/enums/Status.java | 4 +
.../api/service/WorkerGroupService.java | 18 ++
.../api/service/impl/MonitorServiceImpl.java | 2 +-
.../api/service/impl/WorkerGroupServiceImpl.java | 199 ++++++++++++++++++---
.../api/utils/ZookeeperMonitor.java | 143 +++++++--------
.../api/service/WorkerGroupServiceTest.java | 91 +++++++++-
.../apache/dolphinscheduler/common/Constants.java | 4 +-
.../dolphinscheduler/common/enums/ZKNodeType.java | 3 +-
.../dolphinscheduler/dao/entity/WorkerGroup.java | 69 +++++--
.../dao/mapper/ProcessInstanceMapper.java | 15 +-
.../dao/mapper/WorkerGroupMapper.java | 33 ++--
.../dao/mapper/ProcessInstanceMapper.xml | 12 +-
.../dao/mapper/WorkerGroupMapper.xml | 31 ++++
.../master/dispatch/host/CommonHostManager.java | 92 +++++++---
.../master/dispatch/host/HostManagerConfig.java | 2 +-
.../dispatch/host/LowerWeightHostManager.java | 64 +++++--
.../server/master/registry/MasterRegistry.java | 4 +-
.../server/registry/HeartBeatTask.java | 2 +-
.../server/registry/ZookeeperRegistryCenter.java | 6 +-
.../server/worker/config/WorkerConfig.java | 10 +-
.../server/worker/registry/WorkerRegistry.java | 4 +-
.../dolphinscheduler/server/zk/ZKMasterClient.java | 8 +-
.../src/main/resources/master.properties | 2 +-
.../src/main/resources/worker.properties | 10 +-
.../consumer/TaskPriorityQueueConsumerTest.java | 10 +-
.../service/zk/AbstractZKClient.java | 52 ++++--
.../service/zk/RegisterOperator.java | 8 +-
.../service/zk/RegisterOperatorTest.java | 8 +-
.../home/pages/dag/_source/formModel/tasks/sql.vue | 4 +-
.../conf/home/pages/security/pages/queue/index.vue | 2 -
.../home/pages/security/pages/tenement/index.vue | 2 -
.../pages/security/pages/users/_source/list.vue | 2 +-
.../conf/home/pages/security/pages/users/index.vue | 3 -
.../security/pages/warningGroups/_source/list.vue | 2 +-
.../pages/security/pages/warningGroups/index.vue | 3 -
.../pages/security/pages/warningInstance/index.vue | 3 -
.../pages/workerGroups/_source/createWorker.vue | 67 ++++---
.../security/pages/workerGroups/_source/list.vue | 23 ++-
.../pages/security/pages/workerGroups/index.vue | 31 +++-
.../src/js/conf/home/store/security/actions.js | 2 +-
.../src/js/module/i18n/locale/en_US.js | 16 +-
.../src/js/module/i18n/locale/zh_CN.js | 14 +-
sql/dolphinscheduler_mysql.sql | 16 ++
sql/dolphinscheduler_postgre.sql | 2 +-
.../{postgresql => mysql}/dolphinscheduler_ddl.sql | 24 ++-
.../dolphinscheduler_dml.sql} | 4 +-
.../postgresql/dolphinscheduler_ddl.sql | 24 ++-
51 files changed, 896 insertions(+), 346 deletions(-)
diff --git a/docker/build/conf/dolphinscheduler/master.properties.tpl
b/docker/build/conf/dolphinscheduler/master.properties.tpl
index 7b07dd7..3d7910f 100644
--- a/docker/build/conf/dolphinscheduler/master.properties.tpl
+++ b/docker/build/conf/dolphinscheduler/master.properties.tpl
@@ -15,7 +15,7 @@
# limitations under the License.
#
-# master execute thread num
+# master execute thread number
master.exec.threads=${MASTER_EXEC_THREADS}
# master execute task number in parallel
diff --git a/docker/build/conf/dolphinscheduler/worker.properties.tpl
b/docker/build/conf/dolphinscheduler/worker.properties.tpl
index 8f5907e..7dd6935 100644
--- a/docker/build/conf/dolphinscheduler/worker.properties.tpl
+++ b/docker/build/conf/dolphinscheduler/worker.properties.tpl
@@ -15,12 +15,15 @@
# limitations under the License.
#
-# worker execute thread num
+# worker execute thread number
worker.exec.threads=${WORKER_EXEC_THREADS}
# worker heartbeat interval
worker.heartbeat.interval=${WORKER_HEARTBEAT_INTERVAL}
+# worker host weight to dispatch tasks, default value 100
+worker.host.weight=${WORKER_HOST_WEIGHT}
+
# worker tenant auto create
worker.tenant.auto.create=true
@@ -33,11 +36,8 @@ worker.reserved.memory=${WORKER_RESERVED_MEMORY}
# worker listener port
worker.listen.port=${WORKER_LISTEN_PORT}
-# default worker groups
+# default worker groups, if this worker belongs different groups, you can
config the following like that 'worker.groups=default,test'
worker.groups=${WORKER_GROUPS}
-# default worker host weight
-worker.host.weight=${WORKER_HOST_WEIGHT}
-
# alert server listener host
alert.listen.host=${ALERT_LISTEN_HOST}
diff --git a/docker/kubernetes/dolphinscheduler/README.md
b/docker/kubernetes/dolphinscheduler/README.md
index a6f474a..f481cf9 100644
--- a/docker/kubernetes/dolphinscheduler/README.md
+++ b/docker/kubernetes/dolphinscheduler/README.md
@@ -134,7 +134,7 @@ The Configuration file is `values.yaml`, and the following
tables lists the conf
| `master.tolerations`
| If specified, the pod's tolerations
| `{}`
|
| `master.resources`
| The `resource` limit and request config for master server
| `{}`
|
| `master.configmap.DOLPHINSCHEDULER_OPTS`
| The java options for master server
| `""`
|
-| `master.configmap.MASTER_EXEC_THREADS`
| Master execute thread num
| `100`
|
+| `master.configmap.MASTER_EXEC_THREADS`
| Master execute thread number
| `100`
|
| `master.configmap.MASTER_EXEC_TASK_NUM`
| Master execute task number in parallel
| `20`
|
| `master.configmap.MASTER_HEARTBEAT_INTERVAL`
| Master heartbeat interval
| `10`
|
| `master.configmap.MASTER_TASK_COMMIT_RETRYTIMES`
| Master commit task retry times
| `5`
|
@@ -167,13 +167,13 @@ The Configuration file is `values.yaml`, and the
following tables lists the conf
| `worker.tolerations`
| If specified, the pod's tolerations
| `{}`
|
| `worker.resources`
| The `resource` limit and request config for worker server
| `{}`
|
| `worker.configmap.DOLPHINSCHEDULER_OPTS`
| The java options for worker server
| `""`
|
-| `worker.configmap.WORKER_EXEC_THREADS`
| Worker execute thread num
| `100`
|
+| `worker.configmap.WORKER_EXEC_THREADS`
| Worker execute thread number
| `100`
|
| `worker.configmap.WORKER_HEARTBEAT_INTERVAL`
| Worker heartbeat interval
| `10`
|
| `worker.configmap.WORKER_MAX_CPULOAD_AVG`
| Only less than cpu avg load, worker server can work. default value : the
number of cpu cores * 2 | `100`
|
| `worker.configmap.WORKER_RESERVED_MEMORY`
| Only larger than reserved memory, worker server can work. default value
: physical memory * 1/10, unit is G | `0.1`
|
| `worker.configmap.WORKER_LISTEN_PORT`
| Worker listen port
| `1234`
|
| `worker.configmap.WORKER_GROUPS`
| Worker groups
| `default`
|
-| `worker.configmap.WORKER_HOST_WEIGHT`
| Worker host weight
| `100`
|
+| `worker.configmap.WORKER_HOST_WEIGHT`
| Worker host weight
| `100`
|
| `worker.livenessProbe.enabled`
| Turn on and off liveness probe
| `true`
|
| `worker.livenessProbe.initialDelaySeconds`
| Delay before liveness probe is initiated
| `30`
|
| `worker.livenessProbe.periodSeconds`
| How often to perform the probe
| `30`
|
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/WorkerGroupController.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/WorkerGroupController.java
index 00418ab..c2c74ce 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/WorkerGroupController.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/controller/WorkerGroupController.java
@@ -17,10 +17,13 @@
package org.apache.dolphinscheduler.api.controller;
+import static
org.apache.dolphinscheduler.api.enums.Status.DELETE_WORKER_GROUP_FAIL;
import static
org.apache.dolphinscheduler.api.enums.Status.QUERY_WORKER_GROUP_FAIL;
+import static org.apache.dolphinscheduler.api.enums.Status.SAVE_ERROR;
import org.apache.dolphinscheduler.api.exceptions.ApiException;
import org.apache.dolphinscheduler.api.service.WorkerGroupService;
+import org.apache.dolphinscheduler.api.utils.RegexUtils;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.ParameterUtils;
@@ -33,6 +36,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestAttribute;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
@@ -59,6 +63,35 @@ public class WorkerGroupController extends BaseController {
WorkerGroupService workerGroupService;
/**
+ * create or update a worker group
+ *
+ * @param loginUser login user
+ * @param id worker group id
+ * @param name worker group name
+ * @param addrList addr list
+ * @return create or update result code
+ */
+ @ApiOperation(value = "saveWorkerGroup", notes =
"CREATE_WORKER_GROUP_NOTES")
+ @ApiImplicitParams({
+ @ApiImplicitParam(name = "id", value = "WORKER_GROUP_ID", dataType
= "Int", example = "10", defaultValue = "0"),
+ @ApiImplicitParam(name = "name", value = "WORKER_GROUP_NAME",
required = true, dataType = "String"),
+ @ApiImplicitParam(name = "addrList", value = "WORKER_ADDR_LIST",
required = true, dataType = "String")
+ })
+ @PostMapping(value = "/save")
+ @ResponseStatus(HttpStatus.OK)
+ @ApiException(SAVE_ERROR)
+ public Result saveWorkerGroup(@ApiIgnore @RequestAttribute(value =
Constants.SESSION_USER) User loginUser,
+ @RequestParam(value = "id", required =
false, defaultValue = "0") int id,
+ @RequestParam(value = "name") String name,
+ @RequestParam(value = "addrList") String
addrList
+ ) {
+ logger.info("save worker group: login user {}, id:{}, name: {},
addrList: {} ",
+ RegexUtils.escapeNRT(loginUser.getUserName()), id,
RegexUtils.escapeNRT(name), RegexUtils.escapeNRT(addrList));
+ Map<String, Object> result =
workerGroupService.saveWorkerGroup(loginUser, id, name, addrList);
+ return returnDataList(result);
+ }
+
+ /**
* query worker groups paging
*
* @param loginUser login user
@@ -69,21 +102,20 @@ public class WorkerGroupController extends BaseController {
*/
@ApiOperation(value = "queryAllWorkerGroupsPaging", notes =
"QUERY_WORKER_GROUP_PAGING_NOTES")
@ApiImplicitParams({
- @ApiImplicitParam(name = "id", value = "WORKER_GROUP_ID", dataType
= "Int", example = "10", defaultValue = "0"),
- @ApiImplicitParam(name = "name", value = "WORKER_GROUP_NAME",
required = true, dataType = "String"),
- @ApiImplicitParam(name = "ipList", value = "WORKER_IP_LIST",
required = true, dataType = "String")
+ @ApiImplicitParam(name = "pageNo", value = "PAGE_NO", dataType =
"Int", example = "1"),
+ @ApiImplicitParam(name = "pageSize", value = "PAGE_SIZE", dataType
= "Int", example = "20"),
+ @ApiImplicitParam(name = "searchVal", value = "SEARCH_VAL",
dataType = "String")
})
@GetMapping(value = "/list-paging")
@ResponseStatus(HttpStatus.OK)
@ApiException(QUERY_WORKER_GROUP_FAIL)
public Result queryAllWorkerGroupsPaging(@ApiIgnore
@RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@RequestParam("pageNo") Integer
pageNo,
- @RequestParam(value =
"searchVal", required = false) String searchVal,
- @RequestParam("pageSize") Integer
pageSize
+ @RequestParam("pageSize") Integer
pageSize,
+ @RequestParam(value =
"searchVal", required = false) String searchVal
) {
logger.info("query all worker group paging: login user {}, pageNo:{},
pageSize:{}, searchVal:{}",
- loginUser.getUserName(), pageNo, pageSize, searchVal);
-
+ RegexUtils.escapeNRT(loginUser.getUserName()), pageNo,
pageSize, searchVal);
searchVal = ParameterUtils.handleEscapes(searchVal);
Map<String, Object> result =
workerGroupService.queryAllGroupPaging(loginUser, pageNo, pageSize, searchVal);
return returnDataListPaging(result);
@@ -99,14 +131,32 @@ public class WorkerGroupController extends BaseController {
@GetMapping(value = "/all-groups")
@ResponseStatus(HttpStatus.OK)
@ApiException(QUERY_WORKER_GROUP_FAIL)
- public Result queryAllWorkerGroups(@ApiIgnore @RequestAttribute(value =
Constants.SESSION_USER) User loginUser
- ) {
- logger.info("query all worker group: login user {}",
- loginUser.getUserName());
-
+ public Result queryAllWorkerGroups(@ApiIgnore @RequestAttribute(value =
Constants.SESSION_USER) User loginUser) {
+ logger.info("query all worker group: login user {}",
RegexUtils.escapeNRT(loginUser.getUserName()));
Map<String, Object> result = workerGroupService.queryAllGroup();
return returnDataList(result);
}
+ /**
+ * delete worker group by id
+ *
+ * @param loginUser login user
+ * @param id group id
+ * @return delete result code
+ */
+ @ApiOperation(value = "deleteById", notes =
"DELETE_WORKER_GROUP_BY_ID_NOTES")
+ @ApiImplicitParams({
+ @ApiImplicitParam(name = "id", value = "WORKER_GROUP_ID", required
= true, dataType = "Int", example = "10"),
+ })
+ @PostMapping(value = "/delete-by-id")
+ @ResponseStatus(HttpStatus.OK)
+ @ApiException(DELETE_WORKER_GROUP_FAIL)
+ public Result deleteById(@ApiIgnore @RequestAttribute(value =
Constants.SESSION_USER) User loginUser,
+ @RequestParam("id") Integer id
+ ) {
+ logger.info("delete worker group: login user {}, id:{} ",
RegexUtils.escapeNRT(loginUser.getUserName()), id);
+ Map<String, Object> result =
workerGroupService.deleteWorkerGroupById(loginUser, id);
+ return returnDataList(result);
+ }
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
index f77be9f..062b61e 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
@@ -205,6 +205,10 @@ public enum Status {
DATASOURCE_OTHER_PARAMS_ILLEGAL(10171, "datasource other params illegal",
"数据源其他参数不合法"),
DATASOURCE_NAME_ILLEGAL(10172, "datasource name illegal", "数据源名称不合法"),
DATASOURCE_HOST_ILLEGAL(10173, "datasource host illegal", "数据源HOST不合法"),
+ DELETE_WORKER_GROUP_NOT_EXIST(10174, "delete worker group not exist ",
"删除worker分组不存在"),
+ CREATE_WORKER_GROUP_FORBIDDEN_IN_DOCKER(10175, "create worker group
forbidden in docker ", "创建worker分组在docker中禁止"),
+ DELETE_WORKER_GROUP_FORBIDDEN_IN_DOCKER(10176, "delete worker group
forbidden in docker ", "删除worker分组在docker中禁止"),
+ WORKER_ADDRESS_INVALID(10177, "worker address {0} invalid",
"worker地址[{0}]无效"),
UDF_FUNCTION_NOT_EXIST(20001, "UDF function not found", "UDF函数不存在"),
UDF_FUNCTION_EXISTS(20002, "UDF function already exists", "UDF函数已存在"),
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java
index 2f1ea2c..a8b11f4 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java
@@ -27,6 +27,17 @@ import java.util.Map;
public interface WorkerGroupService {
/**
+ * create or update a worker group
+ *
+ * @param loginUser login user
+ * @param id worker group id
+ * @param name worker group name
+ * @param addrList addr list
+ * @return create or update result code
+ */
+ Map<String, Object> saveWorkerGroup(User loginUser, int id, String name,
String addrList);
+
+ /**
* query worker group paging
*
* @param loginUser login user
@@ -44,4 +55,11 @@ public interface WorkerGroupService {
*/
Map<String, Object> queryAllGroup();
+ /**
+ * delete worker group by id
+ * @param id worker group id
+ * @return delete result code
+ */
+ Map<String, Object> deleteWorkerGroupById(User loginUser, Integer id);
+
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MonitorServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MonitorServiceImpl.java
index 7f54e3e..3cdf1d1 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MonitorServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/MonitorServiceImpl.java
@@ -159,7 +159,7 @@ public class MonitorServiceImpl extends BaseServiceImpl
implements MonitorServic
checkNotNull(zookeeperMonitor);
ZKNodeType zkNodeType = isMaster ? ZKNodeType.MASTER :
ZKNodeType.WORKER;
- return zookeeperMonitor.getServersList(zkNodeType);
+ return zookeeperMonitor.getServerList(zkNodeType);
}
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java
index e0282de..381d1e8 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkerGroupServiceImpl.java
@@ -17,34 +17,34 @@
package org.apache.dolphinscheduler.api.service.impl;
-import static
org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
-import static org.apache.dolphinscheduler.common.Constants.SLASH;
-
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.WorkerGroupService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
+import org.apache.dolphinscheduler.api.utils.ZookeeperMonitor;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.ZKNodeType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
+import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
-import org.apache.dolphinscheduler.remote.utils.Host;
+import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
import java.util.ArrayList;
+import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-
+import org.springframework.transaction.annotation.Transactional;
/**
* worker group service impl
@@ -55,12 +55,115 @@ public class WorkerGroupServiceImpl extends
BaseServiceImpl implements WorkerGro
private static final Logger logger =
LoggerFactory.getLogger(WorkerGroupServiceImpl.class);
@Autowired
+ WorkerGroupMapper workerGroupMapper;
+
+ @Autowired
protected ZookeeperCachedOperator zookeeperCachedOperator;
@Autowired
+ private ZookeeperMonitor zookeeperMonitor;
+
+ @Autowired
ProcessInstanceMapper processInstanceMapper;
/**
+ * create or update a worker group
+ *
+ * @param loginUser login user
+ * @param id worker group id
+ * @param name worker group name
+ * @param addrList addr list
+ * @return create or update result code
+ */
+ @Override
+ public Map<String, Object> saveWorkerGroup(User loginUser, int id, String
name, String addrList) {
+ Map<String, Object> result = new HashMap<>();
+ if (isNotAdmin(loginUser, result)) {
+ return result;
+ }
+ if (Constants.DOCKER_MODE && !Constants.KUBERNETES_MODE) {
+ putMsg(result, Status.CREATE_WORKER_GROUP_FORBIDDEN_IN_DOCKER);
+ return result;
+ }
+ if (StringUtils.isEmpty(name)) {
+ putMsg(result, Status.NAME_NULL);
+ return result;
+ }
+ Date now = new Date();
+ WorkerGroup workerGroup;
+ if (id != 0) {
+ workerGroup = workerGroupMapper.selectById(id);
+ // check exist
+ if (workerGroup == null) {
+ workerGroup = new WorkerGroup();
+ workerGroup.setCreateTime(now);
+ }
+ } else {
+ workerGroup = new WorkerGroup();
+ workerGroup.setCreateTime(now);
+ }
+ workerGroup.setName(name);
+ workerGroup.setAddrList(addrList);
+ workerGroup.setUpdateTime(now);
+
+ if (checkWorkerGroupNameExists(workerGroup)) {
+ putMsg(result, Status.NAME_EXIST, workerGroup.getName());
+ return result;
+ }
+ String invalidAddr = checkWorkerGroupAddrList(workerGroup);
+ if (invalidAddr != null) {
+ putMsg(result, Status.WORKER_ADDRESS_INVALID, invalidAddr);
+ return result;
+ }
+ if (workerGroup.getId() != 0) {
+ workerGroupMapper.updateById(workerGroup);
+ } else {
+ workerGroupMapper.insert(workerGroup);
+ }
+ putMsg(result, Status.SUCCESS);
+ return result;
+ }
+
+ /**
+ * check worker group name exists
+ * @param workerGroup worker group
+ * @return boolean
+ */
+ private boolean checkWorkerGroupNameExists(WorkerGroup workerGroup) {
+ List<WorkerGroup> workerGroupList =
workerGroupMapper.queryWorkerGroupByName(workerGroup.getName());
+ if (CollectionUtils.isNotEmpty(workerGroupList)) {
+ // new group has same name
+ if (workerGroup.getId() == 0) {
+ return true;
+ }
+ // check group id
+ for (WorkerGroup group : workerGroupList) {
+ if (group.getId() != workerGroup.getId()) {
+ return true;
+ }
+ }
+ }
+ // check zookeeper
+ String workerGroupPath =
zookeeperCachedOperator.getZookeeperConfig().getDsRoot() +
Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS + Constants.SLASH +
workerGroup.getName();
+ return zookeeperCachedOperator.isExisted(workerGroupPath);
+ }
+
+ /**
+ * check worker group addr list
+ * @param workerGroup worker group
+ * @return boolean
+ */
+ private String checkWorkerGroupAddrList(WorkerGroup workerGroup) {
+ Map<String, String> serverMaps =
zookeeperMonitor.getServerMaps(ZKNodeType.WORKER, true);
+ for (String addr : workerGroup.getAddrList().split(Constants.COMMA)) {
+ if (!serverMaps.containsKey(addr)) {
+ return addr;
+ }
+ }
+ return null;
+ }
+
+ /**
* query worker group paging
*
* @param loginUser login user
@@ -82,7 +185,6 @@ public class WorkerGroupServiceImpl extends BaseServiceImpl
implements WorkerGro
}
List<WorkerGroup> workerGroups = getWorkerGroups(true);
-
List<WorkerGroup> resultDataList = new ArrayList<>();
if (CollectionUtils.isNotEmpty(workerGroups)) {
@@ -98,10 +200,12 @@ public class WorkerGroupServiceImpl extends
BaseServiceImpl implements WorkerGro
searchValDataList = workerGroups;
}
- if (searchValDataList.size() < pageSize) {
- toIndex = (pageNo - 1) * pageSize + searchValDataList.size();
+ if (fromIndex < searchValDataList.size()) {
+ if (toIndex > searchValDataList.size()) {
+ toIndex = searchValDataList.size();
+ }
+ resultDataList = searchValDataList.subList(fromIndex, toIndex);
}
- resultDataList = searchValDataList.subList(fromIndex, toIndex);
}
PageInfo<WorkerGroup> pageInfo = new PageInfo<>(pageNo, pageSize);
@@ -121,13 +225,16 @@ public class WorkerGroupServiceImpl extends
BaseServiceImpl implements WorkerGro
@Override
public Map<String, Object> queryAllGroup() {
Map<String, Object> result = new HashMap<>();
-
List<WorkerGroup> workerGroups = getWorkerGroups(false);
-
- Set<String> availableWorkerGroupSet = workerGroups.stream()
+ List<String> availableWorkerGroupList = workerGroups.stream()
.map(WorkerGroup::getName)
- .collect(Collectors.toSet());
- result.put(Constants.DATA_LIST, availableWorkerGroupSet);
+ .collect(Collectors.toList());
+ int index =
availableWorkerGroupList.indexOf(Constants.DEFAULT_WORKER_GROUP);
+ if (index > -1) {
+ availableWorkerGroupList.remove(index);
+ availableWorkerGroupList.add(0, Constants.DEFAULT_WORKER_GROUP);
+ }
+ result.put(Constants.DATA_LIST, availableWorkerGroupList);
putMsg(result, Status.SUCCESS);
return result;
}
@@ -139,8 +246,10 @@ public class WorkerGroupServiceImpl extends
BaseServiceImpl implements WorkerGro
* @return WorkerGroup list
*/
private List<WorkerGroup> getWorkerGroups(boolean isPaging) {
+ // worker groups from database
+ List<WorkerGroup> workerGroups =
workerGroupMapper.queryAllWorkerGroup();
+ // worker groups from zookeeper
String workerPath =
zookeeperCachedOperator.getZookeeperConfig().getDsRoot() +
Constants.ZOOKEEPER_DOLPHINSCHEDULER_WORKERS;
- List<WorkerGroup> workerGroups = new ArrayList<>();
List<String> workerGroupList = null;
try {
workerGroupList =
zookeeperCachedOperator.getChildrenKeys(workerPath);
@@ -148,32 +257,70 @@ public class WorkerGroupServiceImpl extends
BaseServiceImpl implements WorkerGro
logger.error("getWorkerGroups exception: {}, workerPath: {},
isPaging: {}", e.getMessage(), workerPath, isPaging);
}
- if (workerGroupList == null || workerGroupList.isEmpty()) {
- if (!isPaging) {
+ if (CollectionUtils.isEmpty(workerGroupList)) {
+ if (CollectionUtils.isEmpty(workerGroups) && !isPaging) {
WorkerGroup wg = new WorkerGroup();
- wg.setName(DEFAULT_WORKER_GROUP);
+ wg.setName(Constants.DEFAULT_WORKER_GROUP);
workerGroups.add(wg);
}
return workerGroups;
}
for (String workerGroup : workerGroupList) {
- String workerGroupPath = workerPath + SLASH + workerGroup;
- List<String> childrenNodes =
zookeeperCachedOperator.getChildrenKeys(workerGroupPath);
- if (CollectionUtils.isEmpty(childrenNodes)) {
+ String workerGroupPath = workerPath + Constants.SLASH +
workerGroup;
+ List<String> childrenNodes = null;
+ try {
+ childrenNodes =
zookeeperCachedOperator.getChildrenKeys(workerGroupPath);
+ } catch (Exception e) {
+ logger.error("getChildrenNodes exception: {}, workerGroupPath:
{}", e.getMessage(), workerGroupPath);
+ }
+ if (childrenNodes == null || childrenNodes.isEmpty()) {
continue;
}
WorkerGroup wg = new WorkerGroup();
wg.setName(workerGroup);
if (isPaging) {
- wg.setIpList(childrenNodes.stream().map(node ->
Host.of(node).getIp()).collect(Collectors.toList()));
- String registeredValue =
zookeeperCachedOperator.get(workerGroupPath + SLASH + childrenNodes.get(0));
-
wg.setCreateTime(DateUtils.stringToDate(registeredValue.split(",")[6]));
-
wg.setUpdateTime(DateUtils.stringToDate(registeredValue.split(",")[7]));
+ wg.setAddrList(String.join(Constants.COMMA, childrenNodes));
+ String registeredValue =
zookeeperCachedOperator.get(workerGroupPath + Constants.SLASH +
childrenNodes.get(0));
+
wg.setCreateTime(DateUtils.stringToDate(registeredValue.split(Constants.COMMA)[6]));
+
wg.setUpdateTime(DateUtils.stringToDate(registeredValue.split(Constants.COMMA)[7]));
+ wg.setSystemDefault(true);
}
workerGroups.add(wg);
}
return workerGroups;
}
+ /**
+ * delete worker group by id
+ * @param id worker group id
+ * @return delete result code
+ */
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public Map<String, Object> deleteWorkerGroupById(User loginUser, Integer
id) {
+ Map<String, Object> result = new HashMap<>();
+ if (isNotAdmin(loginUser, result)) {
+ return result;
+ }
+ if (Constants.DOCKER_MODE && !Constants.KUBERNETES_MODE) {
+ putMsg(result, Status.DELETE_WORKER_GROUP_FORBIDDEN_IN_DOCKER);
+ return result;
+ }
+ WorkerGroup workerGroup = workerGroupMapper.selectById(id);
+ if (workerGroup == null) {
+ putMsg(result, Status.DELETE_WORKER_GROUP_NOT_EXIST);
+ return result;
+ }
+ List<ProcessInstance> processInstances =
processInstanceMapper.queryByWorkerGroupNameAndStatus(workerGroup.getName(),
Constants.NOT_TERMINATED_STATES);
+ if (CollectionUtils.isNotEmpty(processInstances)) {
+ putMsg(result, Status.DELETE_WORKER_GROUP_BY_ID_FAIL,
processInstances.size());
+ return result;
+ }
+ workerGroupMapper.deleteById(id);
+
processInstanceMapper.updateProcessInstanceByWorkerGroupName(workerGroup.getName(),
"");
+ putMsg(result, Status.SUCCESS);
+ return result;
+ }
+
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/ZookeeperMonitor.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/ZookeeperMonitor.java
index be47545..b599695 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/ZookeeperMonitor.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/ZookeeperMonitor.java
@@ -14,92 +14,93 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.api.utils;
import org.apache.dolphinscheduler.common.enums.ZKNodeType;
-import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.common.model.Server;
+import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.dao.entity.ZookeeperRecord;
import org.apache.dolphinscheduler.service.zk.AbstractZKClient;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
/**
- * monitor zookeeper info
+ * monitor zookeeper info
*/
@Component
public class ZookeeperMonitor extends AbstractZKClient {
- private static final Logger LOG =
LoggerFactory.getLogger(ZookeeperMonitor.class);
-
- /**
- *
- * @return zookeeper info list
- */
- public List<ZookeeperRecord> zookeeperInfoList(){
- String zookeeperServers =
getZookeeperQuorum().replaceAll("[\\t\\n\\x0B\\f\\r]", "");
- try{
- return zookeeperInfoList(zookeeperServers);
- }catch(Exception e){
- LOG.error(e.getMessage(),e);
- }
- return null;
- }
-
- /**
- * get master servers
- * @return master server information
- */
- public List<Server> getMasterServers(){
- return getServersList(ZKNodeType.MASTER);
- }
-
- /**
- * master construct is the same with worker, use the master instead
- * @return worker server informations
- */
- public List<Server> getWorkerServers(){
- return getServersList(ZKNodeType.WORKER);
- }
-
- private static List<ZookeeperRecord> zookeeperInfoList(String
zookeeperServers) {
-
- List<ZookeeperRecord> list = new ArrayList<>(5);
-
- if(StringUtils.isNotBlank(zookeeperServers)){
- String[] zookeeperServersArray =
zookeeperServers.split(",");
-
- for (String zookeeperServer : zookeeperServersArray) {
- ZooKeeperState state = new
ZooKeeperState(zookeeperServer);
- boolean ok = state.ruok();
- if(ok){
- state.getZookeeperInfo();
- }
-
- int connections = state.getConnections();
- int watches = state.getWatches();
- long sent = state.getSent();
- long received = state.getReceived();
- String mode = state.getMode();
- float minLatency = state.getMinLatency();
- float avgLatency = state.getAvgLatency();
- float maxLatency = state.getMaxLatency();
- int nodeCount = state.getNodeCount();
- int status = ok ? 1 : 0;
- Date date = new Date();
-
- ZookeeperRecord zookeeperRecord = new
ZookeeperRecord(zookeeperServer,connections,watches,sent,received,mode,minLatency,avgLatency,maxLatency,nodeCount,status,date);
- list.add(zookeeperRecord);
-
- }
- }
-
- return list;
- }
+ private static final Logger LOG =
LoggerFactory.getLogger(ZookeeperMonitor.class);
+
+ /**
+ *
+ * @return zookeeper info list
+ */
+ public List<ZookeeperRecord> zookeeperInfoList() {
+ String zookeeperServers =
getZookeeperQuorum().replaceAll("[\\t\\n\\x0B\\f\\r]", "");
+ try {
+ return zookeeperInfoList(zookeeperServers);
+ } catch (Exception e) {
+ LOG.error(e.getMessage(),e);
+ }
+ return null;
+ }
+
+ /**
+ * get master servers
+ * @return master server information
+ */
+ public List<Server> getMasterServers() {
+ return getServerList(ZKNodeType.MASTER);
+ }
+
+ /**
+ * master construct is the same with worker, use the master instead
+ * @return worker server informations
+ */
+ public List<Server> getWorkerServers() {
+ return getServerList(ZKNodeType.WORKER);
+ }
+
+ private static List<ZookeeperRecord> zookeeperInfoList(String
zookeeperServers) {
+
+ List<ZookeeperRecord> list = new ArrayList<>(5);
+
+ if (StringUtils.isNotBlank(zookeeperServers)) {
+ String[] zookeeperServersArray = zookeeperServers.split(",");
+
+ for (String zookeeperServer : zookeeperServersArray) {
+ ZooKeeperState state = new ZooKeeperState(zookeeperServer);
+ boolean ok = state.ruok();
+ if (ok) {
+ state.getZookeeperInfo();
+ }
+
+ int connections = state.getConnections();
+ int watches = state.getWatches();
+ long sent = state.getSent();
+ long received = state.getReceived();
+ String mode = state.getMode();
+ float minLatency = state.getMinLatency();
+ float avgLatency = state.getAvgLatency();
+ float maxLatency = state.getMaxLatency();
+ int nodeCount = state.getNodeCount();
+ int status = ok ? 1 : 0;
+ Date date = new Date();
+
+ ZookeeperRecord zookeeperRecord = new
ZookeeperRecord(zookeeperServer,connections,watches,sent,received,mode,minLatency,avgLatency,maxLatency,nodeCount,status,date);
+ list.add(zookeeperRecord);
+
+ }
+ }
+
+ return list;
+ }
}
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java
index ce9f239..0930a54 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkerGroupServiceTest.java
@@ -17,21 +17,25 @@
package org.apache.dolphinscheduler.api.service;
+import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.impl.WorkerGroupServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
+import org.apache.dolphinscheduler.api.utils.ZookeeperMonitor;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.UserType;
+import org.apache.dolphinscheduler.common.enums.ZKNodeType;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
+import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import org.junit.Assert;
import org.junit.Before;
@@ -52,11 +56,19 @@ public class WorkerGroupServiceTest {
private WorkerGroupServiceImpl workerGroupService;
@Mock
+ private WorkerGroupMapper workerGroupMapper;
+
+ @Mock
private ProcessInstanceMapper processInstanceMapper;
@Mock
private ZookeeperCachedOperator zookeeperCachedOperator;
+ @Mock
+ private ZookeeperMonitor zookeeperMonitor;
+
+ private String groupName = "groupName000001";
+
@Before
public void init() {
ZookeeperConfig zookeeperConfig = new ZookeeperConfig();
@@ -80,6 +92,33 @@ public class WorkerGroupServiceTest {
}
/**
+ * create or update a worker group
+ */
+ @Test
+ public void testSaveWorkerGroup() {
+ // worker server maps
+ Map<String, String> serverMaps = new HashMap<>();
+ serverMaps.put("127.0.0.1:1234",
"0.3,0.07,4.4,7.42,16.0,0.3,2021-03-19 20:17:58,2021-03-19 20:25:29,0,79214");
+ Mockito.when(zookeeperMonitor.getServerMaps(ZKNodeType.WORKER,
true)).thenReturn(serverMaps);
+
+ User user = new User();
+ // general user add
+ user.setUserType(UserType.GENERAL_USER);
+ Map<String, Object> result = workerGroupService.saveWorkerGroup(user,
0, groupName, "127.0.0.1:1234");
+ Assert.assertEquals(Status.USER_NO_OPERATION_PERM.getMsg(),
result.get(Constants.MSG));
+
+ // success
+ user.setUserType(UserType.ADMIN_USER);
+ result = workerGroupService.saveWorkerGroup(user, 0, groupName,
"127.0.0.1:1234");
+ Assert.assertEquals(Status.SUCCESS.getMsg(),
result.get(Constants.MSG));
+ // group name exist
+
Mockito.when(workerGroupMapper.selectById(2)).thenReturn(getWorkerGroup(2));
+
Mockito.when(workerGroupMapper.queryWorkerGroupByName(groupName)).thenReturn(getList());
+ result = workerGroupService.saveWorkerGroup(user, 2, groupName,
"127.0.0.1:1234");
+ Assert.assertEquals(Status.NAME_EXIST, result.get(Constants.STATUS));
+ }
+
+ /**
* query worker group paging
*/
@Test
@@ -93,17 +132,38 @@ public class WorkerGroupServiceTest {
}
@Test
- public void testQueryAllGroup() throws Exception {
+ public void testQueryAllGroup() {
Map<String, Object> result = workerGroupService.queryAllGroup();
- Set<String> workerGroups = (Set<String>)
result.get(Constants.DATA_LIST);
+ List<String> workerGroups = (List<String>)
result.get(Constants.DATA_LIST);
Assert.assertEquals(workerGroups.size(), 1);
}
/**
+ * delete group by id
+ */
+ @Test
+ public void testDeleteWorkerGroupById() {
+ User user = new User();
+ user.setUserType(UserType.ADMIN_USER);
+ WorkerGroup wg2 = getWorkerGroup(2);
+ Mockito.when(workerGroupMapper.selectById(2)).thenReturn(wg2);
+
Mockito.when(processInstanceMapper.queryByWorkerGroupNameAndStatus(wg2.getName(),
Constants.NOT_TERMINATED_STATES)).thenReturn(getProcessInstanceList());
+ Map<String, Object> result =
workerGroupService.deleteWorkerGroupById(user, 1);
+ Assert.assertEquals(Status.DELETE_WORKER_GROUP_NOT_EXIST.getCode(),
((Status) result.get(Constants.STATUS)).getCode());
+ result = workerGroupService.deleteWorkerGroupById(user, 2);
+ Assert.assertEquals(Status.DELETE_WORKER_GROUP_BY_ID_FAIL.getCode(),
((Status) result.get(Constants.STATUS)).getCode());
+ // correct
+ WorkerGroup wg3 = getWorkerGroup(3);
+ Mockito.when(workerGroupMapper.selectById(3)).thenReturn(wg3);
+
Mockito.when(processInstanceMapper.queryByWorkerGroupNameAndStatus(wg3.getName(),
Constants.NOT_TERMINATED_STATES)).thenReturn(new ArrayList<>());
+ result = workerGroupService.deleteWorkerGroupById(user, 3);
+ Assert.assertEquals(Status.SUCCESS.getMsg(),
result.get(Constants.MSG));
+ }
+
+ /**
* get processInstances
*/
private List<ProcessInstance> getProcessInstanceList() {
-
List<ProcessInstance> processInstances = new ArrayList<>();
processInstances.add(new ProcessInstance());
return processInstances;
@@ -112,9 +172,30 @@ public class WorkerGroupServiceTest {
@Test
public void testQueryAllGroupWithDefault() {
Map<String, Object> result = workerGroupService.queryAllGroup();
- Set<String> workerGroups = (Set<String>)
result.get(Constants.DATA_LIST);
+ List<String> workerGroups = (List<String>)
result.get(Constants.DATA_LIST);
Assert.assertEquals(1, workerGroups.size());
Assert.assertEquals("default", workerGroups.toArray()[0]);
}
+ /**
+ * get Group
+ * @return
+ */
+ private WorkerGroup getWorkerGroup(int id) {
+ WorkerGroup workerGroup = new WorkerGroup();
+ workerGroup.setName(groupName);
+ workerGroup.setId(id);
+ return workerGroup;
+ }
+
+ private WorkerGroup getWorkerGroup() {
+ return getWorkerGroup(1);
+ }
+
+ private List<WorkerGroup> getList() {
+ List<WorkerGroup> list = new ArrayList<>();
+ list.add(getWorkerGroup());
+ return list;
+ }
+
}
diff --git
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
index 7925ee0..c4b6243 100644
---
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
+++
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
@@ -785,8 +785,8 @@ public final class Constants {
/**
* master/worker server use for zk
*/
- public static final String MASTER_PREFIX = "master";
- public static final String WORKER_PREFIX = "worker";
+ public static final String MASTER_TYPE = "master";
+ public static final String WORKER_TYPE = "worker";
public static final String DELETE_ZK_OP = "delete";
public static final String ADD_ZK_OP = "add";
public static final String ALIAS = "alias";
diff --git
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ZKNodeType.java
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ZKNodeType.java
index b4b3c59..034f880 100644
---
a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ZKNodeType.java
+++
b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/ZKNodeType.java
@@ -25,7 +25,6 @@ public enum ZKNodeType {
* 0 master node;
* 1 worker node;
* 2 dead_server node;
- * 3 task_queue node;
*/
- MASTER, WORKER, DEAD_SERVER, TASK_QUEUE;
+ MASTER, WORKER, DEAD_SERVER;
}
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/WorkerGroup.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/WorkerGroup.java
index b850219..b6dc13a 100644
---
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/WorkerGroup.java
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/WorkerGroup.java
@@ -14,30 +14,62 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.dolphinscheduler.dao.entity;
+import java.util.Date;
+
import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.annotation.JsonFormat;
-import java.util.Date;
-import java.util.List;
-
/**
* worker group
*/
+@TableName("t_ds_worker_group")
public class WorkerGroup {
+ @TableId(value = "id", type = IdType.AUTO)
+ private int id;
+
private String name;
- private List<String> ipList;
+ private String addrList;
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
+ @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date createTime;
- @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
+
+ @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date updateTime;
+ @TableField(exist = false)
+ private boolean systemDefault;
+
+ public int getId() {
+ return id;
+ }
+
+ public void setId(int id) {
+ this.id = id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getAddrList() {
+ return addrList;
+ }
+
+ public void setAddrList(String addrList) {
+ this.addrList = addrList;
+ }
public Date getCreateTime() {
return createTime;
@@ -55,21 +87,24 @@ public class WorkerGroup {
this.updateTime = updateTime;
}
- public String getName() {
- return name;
+ public boolean getSystemDefault() {
+ return systemDefault;
}
- public void setName(String name) {
- this.name = name;
+ public void setSystemDefault(boolean systemDefault) {
+ this.systemDefault = systemDefault;
}
- public List<String> getIpList() {
- return ipList;
+ @Override
+ public String toString() {
+ return "WorkerGroup{"
+ + "id= " + id
+ + ", name= " + name
+ + ", addrList= " + addrList
+ + ", createTime= " + createTime
+ + ", updateTime= " + updateTime
+ + ", systemDefault= " + systemDefault
+ + "}";
}
- public void setIpList(List<String> ipList) {
- this.ipList = ipList;
- }
-
-
}
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
index c110df0..08d1740 100644
---
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java
@@ -62,12 +62,12 @@ public interface ProcessInstanceMapper extends
BaseMapper<ProcessInstance> {
/**
* query process instance by worker group and stateArray
- * @param workerGroupId workerGroupId
+ * @param workerGroupName workerGroupName
* @param states states array
* @return process instance list
*/
- List<ProcessInstance>
queryByWorkerGroupIdAndStatus(@Param("workerGroupId") int workerGroupId,
- @Param("states") int[]
states);
+ List<ProcessInstance>
queryByWorkerGroupNameAndStatus(@Param("workerGroupName") String
workerGroupName,
+ @Param("states")
int[] states);
/**
* process instance page
@@ -134,12 +134,13 @@ public interface ProcessInstanceMapper extends
BaseMapper<ProcessInstance> {
@Param("destTenantId") int
destTenantId);
/**
- * update process instance by worker groupId
- * @param originWorkerGroupId originWorkerGroupId
- * @param destWorkerGroupId destWorkerGroupId
+ * update process instance by worker group name
+ * @param originWorkerGroupName originWorkerGroupName
+ * @param destWorkerGroupName destWorkerGroupName
* @return update result
*/
- int updateProcessInstanceByWorkerGroupId(@Param("originWorkerGroupId") int
originWorkerGroupId, @Param("destWorkerGroupId") int destWorkerGroupId);
+ int updateProcessInstanceByWorkerGroupName(@Param("originWorkerGroupName")
String originWorkerGroupName,
+ @Param("destWorkerGroupName")
String destWorkerGroupName);
/**
* count process instance state by user
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkerGroupMapper.java
similarity index 59%
copy from
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java
copy to
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkerGroupMapper.java
index 2f1ea2c..21af4b3 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkerGroupService.java
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/WorkerGroupMapper.java
@@ -15,33 +15,32 @@
* limitations under the License.
*/
-package org.apache.dolphinscheduler.api.service;
+package org.apache.dolphinscheduler.dao.mapper;
-import org.apache.dolphinscheduler.dao.entity.User;
+import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
-import java.util.Map;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
/**
- * worker group service
+ * worker group mapper interface
*/
-public interface WorkerGroupService {
+public interface WorkerGroupMapper extends BaseMapper<WorkerGroup> {
/**
- * query worker group paging
- *
- * @param loginUser login user
- * @param pageNo page number
- * @param searchVal search value
- * @param pageSize page size
- * @return worker group list page
+ * query all worker group
+ * @return worker group list
*/
- Map<String, Object> queryAllGroupPaging(User loginUser, Integer pageNo,
Integer pageSize, String searchVal);
+ List<WorkerGroup> queryAllWorkerGroup();
/**
- * query all worker group
- *
- * @return all worker group list
+ * query worer grouop by name
+ * @param name name
+ * @return worker group list
*/
- Map<String, Object> queryAllGroup();
+ List<WorkerGroup> queryWorkerGroupByName(@Param("name") String name);
}
diff --git
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
index c402fb3..dc3b831 100644
---
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
+++
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml
@@ -72,13 +72,13 @@
order by id asc
</select>
- <select id="queryByWorkerGroupIdAndStatus"
resultType="org.apache.dolphinscheduler.dao.entity.ProcessInstance">
+ <select id="queryByWorkerGroupNameAndStatus"
resultType="org.apache.dolphinscheduler.dao.entity.ProcessInstance">
select
<include refid="baseSql"/>
from t_ds_process_instance
where 1=1
- <if test="workerGroupId != -1">
- and worker_group_id =#{workerGroupId}
+ <if test="workerGroupName != ''">
+ and worker_group =#{workerGroupName}
</if>
and state in
<foreach collection="states" item="i" open="(" close=")" separator=",">
@@ -138,10 +138,10 @@
where tenant_id = #{originTenantId}
</update>
- <update id="updateProcessInstanceByWorkerGroupId">
+ <update id="updateProcessInstanceByWorkerGroupName">
update t_ds_process_instance
- set worker_group_id = #{destWorkerGroupId}
- where worker_group_id = #{originWorkerGroupId}
+ set worker_group = #{destWorkerGroupName}
+ where worker_group = #{originWorkerGroupName}
</update>
<select id="countInstanceStateByUser"
resultType="org.apache.dolphinscheduler.dao.entity.ExecuteStatusCount">
diff --git
a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkerGroupMapper.xml
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkerGroupMapper.xml
new file mode 100644
index 0000000..2665f12
--- /dev/null
+++
b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/WorkerGroupMapper.xml
@@ -0,0 +1,31 @@
+<?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.
+ -->
+
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd"; >
+<mapper namespace="org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper">
+ <select id="queryAllWorkerGroup"
resultType="org.apache.dolphinscheduler.dao.entity.WorkerGroup">
+ select *
+ from t_ds_worker_group
+ order by update_time desc
+ </select>
+ <select id="queryWorkerGroupByName"
resultType="org.apache.dolphinscheduler.dao.entity.WorkerGroup">
+ select *
+ from t_ds_worker_group
+ where name = #{name}
+ </select>
+</mapper>
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/CommonHostManager.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/CommonHostManager.java
index 7184b9e..8e1ead3 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/CommonHostManager.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/CommonHostManager.java
@@ -18,18 +18,24 @@
package org.apache.dolphinscheduler.server.master.dispatch.host;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.ZKNodeType;
+import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.ResInfo;
import org.apache.dolphinscheduler.common.utils.StringUtils;
+import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
+import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
import org.apache.dolphinscheduler.remote.utils.Host;
import
org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext;
import org.apache.dolphinscheduler.server.master.dispatch.enums.ExecutorType;
import
org.apache.dolphinscheduler.server.master.dispatch.host.assign.HostWorker;
import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
+import org.apache.dolphinscheduler.server.zk.ZKMasterClient;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
@@ -45,53 +51,96 @@ public abstract class CommonHostManager implements
HostManager {
protected ZookeeperRegistryCenter registryCenter;
/**
- * zookeeperNodeManager
+ * zookeeper node manager
*/
@Autowired
protected ZookeeperNodeManager zookeeperNodeManager;
/**
+ * zk master client
+ */
+ @Autowired
+ protected ZKMasterClient zkMasterClient;
+
+ /**
+ * worker group mapper
+ */
+ @Autowired
+ protected WorkerGroupMapper workerGroupMapper;
+
+ /**
* select host
* @param context context
* @return host
*/
@Override
public Host select(ExecutionContext context) {
- Host host = new Host();
- Collection<String> nodes = null;
+ List<HostWorker> candidates = null;
String workerGroup = context.getWorkerGroup();
- // executor type
ExecutorType executorType = context.getExecutorType();
switch (executorType) {
case WORKER:
- nodes = zookeeperNodeManager.getWorkerGroupNodes(workerGroup);
+ candidates = getHostWorkersFromDatabase(workerGroup);
+ if (candidates.isEmpty()) {
+ candidates = getHostWorkersFromZookeeper(workerGroup);
+ }
break;
case CLIENT:
break;
default:
throw new IllegalArgumentException("invalid executorType : " +
executorType);
-
}
- if (nodes == null || nodes.isEmpty()) {
- return host;
+
+ if (CollectionUtils.isEmpty(candidates)) {
+ return new Host();
}
- List<HostWorker> candidateHosts = new ArrayList<>();
- nodes.forEach(node -> {
- String workerGroupPath =
registryCenter.getWorkerGroupPath(workerGroup);
- String heartbeat =
registryCenter.getRegisterOperator().get(workerGroupPath + "/" + node);
- int hostWeight = Constants.DEFAULT_WORKER_HOST_WEIGHT;
- if (StringUtils.isNotEmpty(heartbeat)) {
- String[] parts = heartbeat.split(Constants.COMMA);
- if (ResInfo.isNewHeartbeatWithWeight(parts)) {
- hostWeight = Integer.parseInt(parts[10]);
+ return select(candidates);
+ }
+
+ protected abstract HostWorker select(Collection<HostWorker> nodes);
+
+ protected List<HostWorker> getHostWorkersFromDatabase(String workerGroup) {
+ List<HostWorker> hostWorkers = new ArrayList<>();
+ List<WorkerGroup> workerGroups =
workerGroupMapper.queryWorkerGroupByName(workerGroup);
+ if (CollectionUtils.isNotEmpty(workerGroups)) {
+ Map<String, String> serverMaps =
zkMasterClient.getServerMaps(ZKNodeType.WORKER, true);
+ for (WorkerGroup wg : workerGroups) {
+ for (String addr : wg.getAddrList().split(Constants.COMMA)) {
+ if (serverMaps.containsKey(addr)) {
+ String heartbeat = serverMaps.get(addr);
+ int hostWeight =
getWorkerHostWeightFromHeartbeat(heartbeat);
+ hostWorkers.add(HostWorker.of(addr, hostWeight,
workerGroup));
+ }
}
}
- candidateHosts.add(HostWorker.of(node, hostWeight, workerGroup));
- });
- return select(candidateHosts);
+ }
+ return hostWorkers;
}
- protected abstract HostWorker select(Collection<HostWorker> nodes);
+ protected List<HostWorker> getHostWorkersFromZookeeper(String workerGroup)
{
+ List<HostWorker> hostWorkers = new ArrayList<>();
+ Collection<String> nodes =
zookeeperNodeManager.getWorkerGroupNodes(workerGroup);
+ if (CollectionUtils.isNotEmpty(nodes)) {
+ for (String node : nodes) {
+ String workerGroupPath =
registryCenter.getWorkerGroupPath(workerGroup);
+ String heartbeat =
registryCenter.getRegisterOperator().get(workerGroupPath + "/" + node);
+ int hostWeight = getWorkerHostWeightFromHeartbeat(heartbeat);
+ hostWorkers.add(HostWorker.of(node, hostWeight, workerGroup));
+ }
+ }
+ return hostWorkers;
+ }
+
+ protected int getWorkerHostWeightFromHeartbeat(String heartbeat) {
+ int hostWeight = Constants.DEFAULT_WORKER_HOST_WEIGHT;
+ if (StringUtils.isNotEmpty(heartbeat)) {
+ String[] parts = heartbeat.split(Constants.COMMA);
+ if (ResInfo.isNewHeartbeatWithWeight(parts)) {
+ hostWeight = Integer.parseInt(parts[10]);
+ }
+ }
+ return hostWeight;
+ }
public void setZookeeperNodeManager(ZookeeperNodeManager
zookeeperNodeManager) {
this.zookeeperNodeManager = zookeeperNodeManager;
@@ -100,4 +149,5 @@ public abstract class CommonHostManager implements
HostManager {
public ZookeeperNodeManager getZookeeperNodeManager() {
return zookeeperNodeManager;
}
+
}
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/HostManagerConfig.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/HostManagerConfig.java
index 458a1ee..bce1d4f 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/HostManagerConfig.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/HostManagerConfig.java
@@ -45,7 +45,7 @@ public class HostManagerConfig {
String hostSelector = masterConfig.getHostSelector();
HostSelector selector = HostSelector.of(hostSelector);
HostManager hostManager;
- switch (selector){
+ switch (selector) {
case RANDOM:
hostManager = new RandomHostManager();
break;
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java
index ef249a0..e0b585a 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java
@@ -18,9 +18,11 @@
package org.apache.dolphinscheduler.server.master.dispatch.host;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.ZKNodeType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.ResInfo;
+import org.apache.dolphinscheduler.dao.entity.WorkerGroup;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory;
import
org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext;
@@ -28,9 +30,11 @@ import
org.apache.dolphinscheduler.server.master.dispatch.host.assign.HostWeight
import
org.apache.dolphinscheduler.server.master.dispatch.host.assign.HostWorker;
import
org.apache.dolphinscheduler.server.master.dispatch.host.assign.LowerWeightRoundRobin;
+import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -90,7 +94,7 @@ public class LowerWeightHostManager extends CommonHostManager
{
}
@PreDestroy
- public void close(){
+ public void close() {
this.executorService.shutdownNow();
}
@@ -137,9 +141,30 @@ public class LowerWeightHostManager extends
CommonHostManager {
@Override
public void run() {
try {
+ Map<String, Set<HostWeight>> workerHostWeights = new
HashMap<>();
+ // from database
+ List<WorkerGroup> workerGroups =
workerGroupMapper.queryAllWorkerGroup();
+ if (CollectionUtils.isNotEmpty(workerGroups)) {
+ Map<String, String> serverMaps =
zkMasterClient.getServerMaps(ZKNodeType.WORKER, true);
+ for (WorkerGroup wg : workerGroups) {
+ String workerGroup = wg.getName();
+ List<String> addrs =
Arrays.asList(wg.getAddrList().split(Constants.COMMA));
+ Set<HostWeight> hostWeights = new
HashSet<>(addrs.size());
+ for (String addr : addrs) {
+ if (serverMaps.containsKey(addr)) {
+ String heartbeat = serverMaps.get(addr);
+ HostWeight hostWeight = getHostWeight(addr,
workerGroup, heartbeat);
+ if (hostWeight != null) {
+ hostWeights.add(hostWeight);
+ }
+ }
+ }
+ workerHostWeights.put(workerGroup, hostWeights);
+ }
+ }
+ // from zookeeper
Map<String, Set<String>> workerGroupNodes =
zookeeperNodeManager.getWorkerGroupNodes();
Set<Map.Entry<String, Set<String>>> entries =
workerGroupNodes.entrySet();
- Map<String, Set<HostWeight>> workerHostWeights = new
HashMap<>();
for (Map.Entry<String, Set<String>> entry : entries) {
String workerGroup = entry.getKey();
Set<String> nodes = entry.getValue();
@@ -147,20 +172,8 @@ public class LowerWeightHostManager extends
CommonHostManager {
Set<HostWeight> hostWeights = new HashSet<>(nodes.size());
for (String node : nodes) {
String heartbeat =
registryCenter.getRegisterOperator().get(workerGroupPath + "/" + node);
- if (ResInfo.isValidHeartbeatForZKInfo(heartbeat)) {
- String[] parts = heartbeat.split(Constants.COMMA);
- int status = Integer.parseInt(parts[8]);
- if (status == Constants.ABNORMAL_NODE_STATUS) {
- logger.warn("load is too high or
availablePhysicalMemorySize(G) is too low, it's
availablePhysicalMemorySize(G):{},loadAvg:{}",
- Double.parseDouble(parts[3]) ,
Double.parseDouble(parts[2]));
- continue;
- }
- double cpu = Double.parseDouble(parts[0]);
- double memory = Double.parseDouble(parts[1]);
- double loadAverage = Double.parseDouble(parts[2]);
- long startTime =
DateUtils.stringToDate(parts[6]).getTime();
- int weight =
ResInfo.isNewHeartbeatWithWeight(parts) ? Integer.parseInt(parts[10]) :
Constants.DEFAULT_WORKER_HOST_WEIGHT;
- HostWeight hostWeight = new
HostWeight(HostWorker.of(node, weight, workerGroup), cpu, memory, loadAverage,
startTime);
+ HostWeight hostWeight = getHostWeight(node,
workerGroup, heartbeat);
+ if (hostWeight != null) {
hostWeights.add(hostWeight);
}
}
@@ -171,6 +184,25 @@ public class LowerWeightHostManager extends
CommonHostManager {
logger.error("RefreshResourceTask error", ex);
}
}
+
+ public HostWeight getHostWeight(String addr, String workerGroup,
String heartbeat) {
+ if (ResInfo.isValidHeartbeatForZKInfo(heartbeat)) {
+ String[] parts = heartbeat.split(Constants.COMMA);
+ int status = Integer.parseInt(parts[8]);
+ if (status == Constants.ABNORMAL_NODE_STATUS) {
+ logger.warn("load is too high or
availablePhysicalMemorySize(G) is too low, it's
availablePhysicalMemorySize(G):{},loadAvg:{}",
+ Double.parseDouble(parts[3]),
Double.parseDouble(parts[2]));
+ return null;
+ }
+ double cpu = Double.parseDouble(parts[0]);
+ double memory = Double.parseDouble(parts[1]);
+ double loadAverage = Double.parseDouble(parts[2]);
+ long startTime = DateUtils.stringToDate(parts[6]).getTime();
+ int weight = getWorkerHostWeightFromHeartbeat(heartbeat);
+ return new HostWeight(HostWorker.of(addr, weight,
workerGroup), cpu, memory, loadAverage, startTime);
+ }
+ return null;
+ }
}
}
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java
index f77f00d..09e9e22 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java
@@ -81,7 +81,7 @@ public class MasterRegistry {
* registry
*/
public void registry() {
- String address = NetUtils.getHost();
+ String address = NetUtils.getAddr(masterConfig.getListenPort());
String localNodePath = getMasterPath();
zookeeperRegistryCenter.getRegisterOperator().persistEphemeral(localNodePath,
"");
zookeeperRegistryCenter.getRegisterOperator().getZkClient().getConnectionStateListenable().addListener(
@@ -101,7 +101,7 @@ public class MasterRegistry {
masterConfig.getMasterMaxCpuloadAvg(),
masterConfig.getMasterReservedMemory(),
Sets.newHashSet(getMasterPath()),
- Constants.MASTER_PREFIX,
+ Constants.MASTER_TYPE,
zookeeperRegistryCenter);
this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask,
masterHeartbeatInterval, masterHeartbeatInterval, TimeUnit.SECONDS);
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java
index e4c3d6a..460cc9f 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java
@@ -113,7 +113,7 @@ public class HeartBeatTask implements Runnable {
// save process id
builder.append(OSUtils.getProcessID());
// worker host weight
- if (Constants.WORKER_PREFIX.equals(serverType)) {
+ if (Constants.WORKER_TYPE.equals(serverType)) {
builder.append(Constants.COMMA).append(hostWeight);
}
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperRegistryCenter.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperRegistryCenter.java
index 591333b..fdbcb8f 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperRegistryCenter.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/ZookeeperRegistryCenter.java
@@ -17,10 +17,8 @@
package org.apache.dolphinscheduler.server.registry;
-import static org.apache.dolphinscheduler.common.Constants.MASTER_PREFIX;
import static org.apache.dolphinscheduler.common.Constants.SINGLE_SLASH;
import static org.apache.dolphinscheduler.common.Constants.UNDERLINE;
-import static org.apache.dolphinscheduler.common.Constants.WORKER_PREFIX;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.IStoppable;
@@ -230,9 +228,7 @@ public class ZookeeperRegistryCenter implements
InitializingBean {
// ip_sequence_no
String[] zNodesPath = zNode.split("\\/");
String ipSeqNo = zNodesPath[zNodesPath.length - 1];
-
- String type = serverType.equals(MASTER_PREFIX) ? MASTER_PREFIX :
WORKER_PREFIX;
- String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH + type
+ UNDERLINE + ipSeqNo;
+ String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH +
serverType + UNDERLINE + ipSeqNo;
return !registerOperator.isExisted(zNode) ||
registerOperator.isExisted(deadServerPath);
}
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
index 1492899..22c19af 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
@@ -35,6 +35,9 @@ public class WorkerConfig {
@Value("${worker.heartbeat.interval:10}")
private int workerHeartbeatInterval;
+ @Value("${worker.host.weight:100}")
+ private int hostWeight;
+
@Value("${worker.tenant.auto.create:false}")
private boolean workerTenantAutoCreate;
@@ -44,14 +47,11 @@ public class WorkerConfig {
@Value("${worker.reserved.memory:0.3}")
private double workerReservedMemory;
- @Value("#{'${worker.groups:default}'.split(',')}")
- private Set<String> workerGroups;
-
@Value("${worker.listen.port:1234}")
private int listenPort;
- @Value("${worker.host.weight:100}")
- private int hostWeight;
+ @Value("#{'${worker.groups:default}'.split(',')}")
+ private Set<String> workerGroups;
@Value("${alert.listen.host:localhost}")
private String alertListenHost;
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java
index 994fb58..0ef9956 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java
@@ -99,7 +99,7 @@ public class WorkerRegistry {
* registry
*/
public void registry() {
- String address = NetUtils.getHost();
+ String address = NetUtils.getAddr(workerConfig.getListenPort());
Set<String> workerZkPaths = getWorkerZkPaths();
int workerHeartbeatInterval =
workerConfig.getWorkerHeartbeatInterval();
@@ -125,7 +125,7 @@ public class WorkerRegistry {
workerConfig.getWorkerReservedMemory(),
workerConfig.getHostWeight(),
workerZkPaths,
- Constants.WORKER_PREFIX,
+ Constants.WORKER_TYPE,
zookeeperRegistryCenter);
this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask,
workerHeartbeatInterval, workerHeartbeatInterval, TimeUnit.SECONDS);
diff --git
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/zk/ZKMasterClient.java
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/zk/ZKMasterClient.java
index 0f30a5c..f19bfa2 100644
---
a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/zk/ZKMasterClient.java
+++
b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/zk/ZKMasterClient.java
@@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.server.zk;
import static org.apache.dolphinscheduler.common.Constants.SLEEP_TIME_MILLIS;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.IStoppable;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.ZKNodeType;
import org.apache.dolphinscheduler.common.model.Server;
@@ -30,7 +31,6 @@ import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.server.builder.TaskExecutionContextBuilder;
import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
-import org.apache.dolphinscheduler.server.master.MasterServer;
import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
import org.apache.dolphinscheduler.server.utils.ProcessUtils;
import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -73,7 +73,7 @@ public class ZKMasterClient extends AbstractZKClient {
@Autowired
private MasterRegistry masterRegistry;
- public void start(MasterServer masterServer) {
+ public void start(IStoppable stoppable) {
InterProcessMutex mutex = null;
try {
// create distributed lock with the root node path of the lock
space as /dolphinscheduler/lock/failover/startup-masters
@@ -83,7 +83,7 @@ public class ZKMasterClient extends AbstractZKClient {
// master registry
masterRegistry.registry();
-
masterRegistry.getZookeeperRegistryCenter().setStoppable(masterServer);
+
masterRegistry.getZookeeperRegistryCenter().setStoppable(stoppable);
String registryPath = this.masterRegistry.getMasterPath();
masterRegistry.getZookeeperRegistryCenter().getRegisterOperator().handleDeadServer(registryPath,
ZKNodeType.MASTER, Constants.DELETE_ZK_OP);
@@ -282,7 +282,7 @@ public class ZKMasterClient extends AbstractZKClient {
return false;
}
Date workerServerStartDate = null;
- List<Server> workerServers = getServersList(ZKNodeType.WORKER);
+ List<Server> workerServers = getServerList(ZKNodeType.WORKER);
for (Server workerServer : workerServers) {
if (taskInstance.getHost().equals(workerServer.getHost() +
Constants.COLON + workerServer.getPort())) {
workerServerStartDate = workerServer.getCreateTime();
diff --git a/dolphinscheduler-server/src/main/resources/master.properties
b/dolphinscheduler-server/src/main/resources/master.properties
index 01d1189..bb9877c 100644
--- a/dolphinscheduler-server/src/main/resources/master.properties
+++ b/dolphinscheduler-server/src/main/resources/master.properties
@@ -15,7 +15,7 @@
# limitations under the License.
#
-# master execute thread num
+# master execute thread number
#master.exec.threads=100
# master execute task number in parallel
diff --git a/dolphinscheduler-server/src/main/resources/worker.properties
b/dolphinscheduler-server/src/main/resources/worker.properties
index 88c15a6..c5c379c 100644
--- a/dolphinscheduler-server/src/main/resources/worker.properties
+++ b/dolphinscheduler-server/src/main/resources/worker.properties
@@ -15,12 +15,15 @@
# limitations under the License.
#
-# worker execute thread num
+# worker execute thread number
#worker.exec.threads=100
# worker heartbeat interval
#worker.heartbeat.interval=10
+# worker host weight to dispatch tasks, default value 100
+#worker.host.weight=100
+
# worker tenant auto create
#worker.tenant.auto.create=false
@@ -33,11 +36,8 @@
# worker listener port
#worker.listen.port=1234
-# default worker groups
+# default worker groups, if this worker belongs different groups, you can
config the following like that 'worker.groups=default,test'
#worker.groups=default
-# default worker host weight
-#worker.host.weight=100
-
# alert server listener host
alert.listen.host=localhost
diff --git
a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
index 74cd2da..14b1c61 100644
---
a/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
+++
b/dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/consumer/TaskPriorityQueueConsumerTest.java
@@ -25,6 +25,7 @@ import org.apache.dolphinscheduler.common.enums.ResourceType;
import org.apache.dolphinscheduler.common.model.TaskNode;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.dao.datasource.SpringConnectionFactory;
import org.apache.dolphinscheduler.dao.entity.DataSource;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
@@ -36,10 +37,12 @@ import
org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.dispatch.ExecutorDispatcher;
import
org.apache.dolphinscheduler.server.master.dispatch.executor.NettyExecutorManager;
+import org.apache.dolphinscheduler.server.master.registry.MasterRegistry;
import org.apache.dolphinscheduler.server.registry.DependencyConfig;
import org.apache.dolphinscheduler.server.registry.ZookeeperNodeManager;
import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
import org.apache.dolphinscheduler.server.zk.SpringZKServer;
+import org.apache.dolphinscheduler.server.zk.ZKMasterClient;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.queue.TaskPriority;
@@ -66,12 +69,11 @@ import
org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = {DependencyConfig.class,
SpringApplicationContext.class, SpringZKServer.class,
CuratorZookeeperClient.class,
- NettyExecutorManager.class, ExecutorDispatcher.class,
ZookeeperRegistryCenter.class, TaskPriorityQueueConsumer.class,
- ZookeeperNodeManager.class, RegisterOperator.class,
ZookeeperConfig.class, MasterConfig.class,
- CuratorZookeeperClient.class})
+ NettyExecutorManager.class, ExecutorDispatcher.class,
ZookeeperRegistryCenter.class, ZKMasterClient.class,
TaskPriorityQueueConsumer.class,
+ ZookeeperNodeManager.class, RegisterOperator.class,
ZookeeperConfig.class, MasterConfig.class, MasterRegistry.class,
+ CuratorZookeeperClient.class, SpringConnectionFactory.class})
public class TaskPriorityQueueConsumerTest {
-
@Autowired
private TaskPriorityQueue<TaskPriority> taskPriorityQueue;
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java
index 7cdf680..029dfac 100644
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java
+++
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java
@@ -76,38 +76,38 @@ public abstract class AbstractZKClient extends
RegisterOperator {
* @param zkNodeType zookeeper node type
* @return server list
*/
- public List<Server> getServersList(ZKNodeType zkNodeType) {
- Map<String, String> masterMap = getServerMaps(zkNodeType);
+ public List<Server> getServerList(ZKNodeType zkNodeType) {
+ Map<String, String> serverMaps = getServerMaps(zkNodeType);
String parentPath = getZNodeParentPath(zkNodeType);
- List<Server> masterServers = new ArrayList<>();
- for (Map.Entry<String, String> entry : masterMap.entrySet()) {
- Server masterServer =
ResInfo.parseHeartbeatForZKInfo(entry.getValue());
- if (masterServer == null) {
+ List<Server> serverList = new ArrayList<>();
+ for (Map.Entry<String, String> entry : serverMaps.entrySet()) {
+ Server server = ResInfo.parseHeartbeatForZKInfo(entry.getValue());
+ if (server == null) {
continue;
}
String key = entry.getKey();
- masterServer.setZkDirectory(parentPath + "/" + key);
- //set host and port
+ server.setZkDirectory(parentPath + "/" + key);
+ // set host and port
String[] hostAndPort = key.split(COLON);
String[] hosts = hostAndPort[0].split(DIVISION_STRING);
// fetch the last one
- masterServer.setHost(hosts[hosts.length - 1]);
- masterServer.setPort(Integer.parseInt(hostAndPort[1]));
- masterServers.add(masterServer);
+ server.setHost(hosts[hosts.length - 1]);
+ server.setPort(Integer.parseInt(hostAndPort[1]));
+ serverList.add(server);
}
- return masterServers;
+ return serverList;
}
/**
- * get master server list map.
+ * get server list map.
*
* @param zkNodeType zookeeper node type
+ * @param hostOnly host only
* @return result : {host : resource info}
*/
- public Map<String, String> getServerMaps(ZKNodeType zkNodeType) {
-
- Map<String, String> masterMap = new HashMap<>();
+ public Map<String, String> getServerMaps(ZKNodeType zkNodeType, boolean
hostOnly) {
+ Map<String, String> serverMap = new HashMap<>();
try {
String path = getZNodeParentPath(zkNodeType);
List<String> serverList = super.getChildrenKeys(path);
@@ -122,13 +122,27 @@ public abstract class AbstractZKClient extends
RegisterOperator {
serverList = workerList;
}
for (String server : serverList) {
- masterMap.putIfAbsent(server, super.get(path + Constants.SLASH
+ server));
+ String host = server;
+ if (zkNodeType == ZKNodeType.WORKER && hostOnly) {
+ host = server.split(Constants.SLASH)[1];
+ }
+ serverMap.putIfAbsent(host, super.get(path + Constants.SLASH +
server));
}
} catch (Exception e) {
logger.error("get server list failed", e);
}
- return masterMap;
+ return serverMap;
+ }
+
+ /**
+ * get server list map.
+ *
+ * @param zkNodeType zookeeper node type
+ * @return result : {host : resource info}
+ */
+ public Map<String, String> getServerMaps(ZKNodeType zkNodeType) {
+ return getServerMaps(zkNodeType, false);
}
/**
@@ -145,7 +159,7 @@ public abstract class AbstractZKClient extends
RegisterOperator {
host, zkNodeType);
return false;
}
- Map<String, String> serverMaps = getServerMaps(zkNodeType);
+ Map<String, String> serverMaps = getServerMaps(zkNodeType, true);
for (String hostKey : serverMaps.keySet()) {
if (hostKey.contains(host)) {
return true;
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/RegisterOperator.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/RegisterOperator.java
index 0c6ac6d..185f9ff 100644
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/RegisterOperator.java
+++
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/RegisterOperator.java
@@ -19,10 +19,10 @@ package org.apache.dolphinscheduler.service.zk;
import static org.apache.dolphinscheduler.common.Constants.ADD_ZK_OP;
import static org.apache.dolphinscheduler.common.Constants.DELETE_ZK_OP;
-import static org.apache.dolphinscheduler.common.Constants.MASTER_PREFIX;
+import static org.apache.dolphinscheduler.common.Constants.MASTER_TYPE;
import static org.apache.dolphinscheduler.common.Constants.SINGLE_SLASH;
import static org.apache.dolphinscheduler.common.Constants.UNDERLINE;
-import static org.apache.dolphinscheduler.common.Constants.WORKER_PREFIX;
+import static org.apache.dolphinscheduler.common.Constants.WORKER_TYPE;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ZKNodeType;
@@ -99,7 +99,7 @@ public class RegisterOperator extends ZookeeperCachedOperator
{
*/
public void handleDeadServer(String zNode, ZKNodeType zkNodeType, String
opType) throws Exception {
String host = getHostByEventDataPath(zNode);
- String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_PREFIX :
WORKER_PREFIX;
+ String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_TYPE :
WORKER_TYPE;
//check server restart, if restart , dead server path in zk should be
delete
if (opType.equals(DELETE_ZK_OP)) {
@@ -130,7 +130,7 @@ public class RegisterOperator extends
ZookeeperCachedOperator {
*/
public void handleDeadServer(Set<String> zNodeSet, ZKNodeType zkNodeType,
String opType) throws Exception {
- String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_PREFIX :
WORKER_PREFIX;
+ String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_TYPE :
WORKER_TYPE;
for (String zNode : zNodeSet) {
String host = getHostByEventDataPath(zNode);
//check server restart, if restart , dead server path in zk should
be delete
diff --git
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/zk/RegisterOperatorTest.java
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/zk/RegisterOperatorTest.java
index 7823c9b..cf77080 100644
---
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/zk/RegisterOperatorTest.java
+++
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/zk/RegisterOperatorTest.java
@@ -97,7 +97,7 @@ public class RegisterOperatorTest {
testAfterPropertiesSet();
registerOperator.handleDeadServer(MASTER_NODE,
ZKNodeType.MASTER,Constants.ADD_ZK_OP);
String path = registerOperator.getDeadZNodeParentPath();
-
Assert.assertTrue(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_PREFIX,MASTER_NODE)));
+
Assert.assertTrue(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_TYPE,MASTER_NODE)));
}
@@ -107,10 +107,10 @@ public class RegisterOperatorTest {
String path = registerOperator.getDeadZNodeParentPath();
registerOperator.handleDeadServer(MASTER_NODE,
ZKNodeType.MASTER,Constants.ADD_ZK_OP);
-
Assert.assertTrue(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_PREFIX,MASTER_NODE)));
+
Assert.assertTrue(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_TYPE,MASTER_NODE)));
-
registerOperator.removeDeadServerByHost(MASTER_NODE,Constants.MASTER_PREFIX);
-
Assert.assertFalse(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_PREFIX,MASTER_NODE)));
+
registerOperator.removeDeadServerByHost(MASTER_NODE,Constants.MASTER_TYPE);
+
Assert.assertFalse(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_TYPE,MASTER_NODE)));
}
@Test
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/sql.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/sql.vue
index d016c30..3df9ee6 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/sql.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/sql.vue
@@ -37,9 +37,9 @@
<el-switch size="small" v-model="sendEmail"></el-switch>
</div>
<div style="display: inline-block;" v-if="sqlType === '0'">
- <span class="text-b">{{$t('Display query result')}}</span>
+ <span class="text-b">{{$t('Log display')}}</span>
<m-select-input v-model="displayRows" :list="[1,10,25,50,100]"
style="width: 70px;"></m-select-input>
- <span>({{$t('Rows')}})</span>
+ <span>{{$t('rows of result')}}</span>
</div>
</div>
</m-list-box>
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/queue/index.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/queue/index.vue
index 2f7bea6..353018e 100644
--- a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/queue/index.vue
+++ b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/queue/index.vue
@@ -117,11 +117,9 @@
this._debounceGET('false')
this.createQueueDialog = false
},
-
close () {
this.createQueueDialog = false
},
-
_getList (flag) {
if (sessionStorage.getItem('isLeft') === 0) {
this.isLeft = false
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/tenement/index.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/tenement/index.vue
index 89ed41e..5c82073 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/tenement/index.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/tenement/index.vue
@@ -119,11 +119,9 @@
this._debounceGET('false')
this.createTenementDialog = false
},
-
close () {
this.createTenementDialog = false
},
-
_getList (flag) {
if (sessionStorage.getItem('isLeft') === 0) {
this.isLeft = false
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/_source/list.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/_source/list.vue
index 7ecf124..615cf97 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/_source/list.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/_source/list.vue
@@ -48,7 +48,7 @@
<span>{{scope.row.updateTime | formatDate}}</span>
</template>
</el-table-column>
- <el-table-column :label="$t('Operation')" width="120" fixed="right">
+ <el-table-column :label="$t('Operation')" width="100" fixed="right">
<template slot-scope="scope">
<el-tooltip :content="$t('Authorize')" placement="top">
<el-dropdown trigger="click">
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/index.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/index.vue
index 46e91ad..220346b 100644
--- a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/index.vue
+++ b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/users/index.vue
@@ -113,16 +113,13 @@
this.item = item
this.createUserDialog = true
},
-
onUpdate () {
this._debounceGET('false')
this.createUserDialog = false
},
-
close () {
this.createUserDialog = false
},
-
_getList (flag) {
if (sessionStorage.getItem('isLeft') === 0) {
this.isLeft = false
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningGroups/_source/list.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningGroups/_source/list.vue
index 90561f7..0946c12 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningGroups/_source/list.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningGroups/_source/list.vue
@@ -35,7 +35,7 @@
<span>{{scope.row.updateTime | formatDate}}</span>
</template>
</el-table-column>
- <el-table-column :label="$t('Operation')" width="130">
+ <el-table-column :label="$t('Operation')" width="100">
<template slot-scope="scope">
<el-tooltip :content="$t('Edit')" placement="top">
<span><el-button type="primary" size="mini"
icon="el-icon-edit-outline" @click="_edit(scope.row)" circle></el-button></span>
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningGroups/index.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningGroups/index.vue
index cfe9bac..50704a3 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningGroups/index.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningGroups/index.vue
@@ -121,16 +121,13 @@
this.item = item
this.createWarningDialog = true
},
-
onUpdate () {
this._debounceGET('false')
this.createWarningDialog = false
},
-
close () {
this.createWarningDialog = false
},
-
_getList (flag) {
if (sessionStorage.getItem('isLeft') === 0) {
this.isLeft = false
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningInstance/index.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningInstance/index.vue
index deb0c00..3b676e8 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningInstance/index.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/warningInstance/index.vue
@@ -121,16 +121,13 @@
this.item = item
this.createWarningDialog = true
},
-
onUpdate () {
this._debounceGET('false')
this.createWarningDialog = false
},
-
close () {
this.createWarningDialog = false
},
-
_getList (flag) {
if (sessionStorage.getItem('isLeft') === 0) {
this.isLeft = false
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/createWorker.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/createWorker.vue
index bc2d948..b799c1a 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/createWorker.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/createWorker.vue
@@ -15,11 +15,11 @@
* limitations under the License.
*/
<template>
- <m-popup
- ref="popup"
+ <m-popover
+ ref="popover"
:ok-text="item ? $t('Edit') : $t('Submit')"
- :nameText="item ? $t('Edit worker group') : $t('Create worker
group')"
- @ok="_ok">
+ @ok="_ok"
+ @close="close">
<template slot="content">
<div class="create-worker-model">
<m-list-box-f>
@@ -35,28 +35,28 @@
</template>
</m-list-box-f>
<m-list-box-f>
- <template slot="name"><strong>*</strong>IP</template>
+ <template slot="name"><strong>*</strong>{{$t('Worker
Addresses')}}</template>
<template slot="content">
<el-input
:autosize="{ minRows: 4, maxRows: 6 }"
type="textarea"
size="mini"
- v-model="ipList"
- :placeholder="$t('Please enter the IP address separated by
commas')">
+ v-model.trim="addrList"
+ :placeholder="$t('Please enter the worker addresses
separated by commas')">
</el-input>
- <div class="ipt-tip">
- <span>{{$t('Note: Multiple IP addresses have been comma
separated')}}</span>
+ <div class="cwm-tip">
+ <span>{{$t('Note: Multiple worker addresses have been comma
separated')}}</span>
</div>
</template>
</m-list-box-f>
</div>
</template>
- </m-popup>
+ </m-popover>
</template>
<script>
import i18n from '@/module/i18n'
import store from '@/conf/home/store'
- import mPopup from '@/module/components/popup/popup'
+ import mPopover from '@/module/components/popup/popover'
import mListBoxF from '@/module/components/listBoxF/listBoxF'
export default {
@@ -66,7 +66,7 @@
store,
id: 0,
name: '',
- ipList: ''
+ addrList: ''
}
},
props: {
@@ -79,15 +79,13 @@
this._submit()
}
},
- checkIsIps (ips) {
- let reg =
/^(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])$/
- let valdata = ips.split(',')
- for (let i = 0; i < valdata.length; i++) {
- if (reg.test(valdata[i]) === false) {
- return false
- }
- }
- return true
+ checkIpAndPorts (addrs) {
+ let reg =
/^(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5])\.(\d{1,2}|1\d\d|2[0-4]\d|25[0-5]):\d{1,5}$/
+ return addrs.split(',').every(item => reg.test(item))
+ },
+ checkFqdnAndPorts (addrs) {
+ let reg = /^([\w-]+\.)*[\w-]+:\d{1,5}$/i
+ return addrs.split(',').every(item => reg.test(item))
},
_verification () {
// group name
@@ -95,12 +93,12 @@
this.$message.warning(`${i18n.$t('Please enter group name')}`)
return false
}
- if (!this.ipList) {
- this.$message.warning(`${i18n.$t('IP address cannot be empty')}`)
+ if (!this.addrList) {
+ this.$message.warning(`${i18n.$t('Worker addresses cannot be
empty')}`)
return false
}
- if (!this.checkIsIps(this.ipList)) {
- this.$message.warning(`${i18n.$t('Please enter the correct IP')}`)
+ if (!this.checkIpAndPorts(this.addrList) &&
!this.checkFqdnAndPorts(this.addrList)) {
+ this.$message.warning(`${i18n.$t('Please enter the correct worker
addresses')}`)
return false
}
return true
@@ -109,22 +107,23 @@
let param = {
id: this.id,
name: this.name,
- ipList: this.ipList
+ addrList: this.addrList
}
if (this.item) {
param.id = this.item.id
}
- this.$refs.popup.spinnerLoading = true
+ this.$refs.popover.spinnerLoading = true
this.store.dispatch('security/saveWorkerGroups', param).then(res => {
+ this.$refs.popover.spinnerLoading = false
this.$emit('onUpdate')
this.$message.success(res.msg)
- setTimeout(() => {
- this.$refs.popup.spinnerLoading = false
- }, 800)
}).catch(e => {
this.$message.error(e.msg || '')
- this.$refs.popup.spinnerLoading = false
+ this.$refs.popover.spinnerLoading = false
})
+ },
+ close () {
+ this.$emit('close')
}
},
watch: {},
@@ -132,17 +131,17 @@
if (this.item) {
this.id = this.item.id
this.name = this.item.name
- this.ipList = this.item.ipList
+ this.addrList = this.item.addrList
}
},
mounted () {
},
- components: { mPopup, mListBoxF }
+ components: { mPopover, mListBoxF }
}
</script>
<style lang="scss" rel="stylesheet/scss">
.create-worker-model {
- .ipt-tip {
+ .cwm-tip {
color: #999;
padding-top: 4px;
display: block;
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/list.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/list.vue
index 0c5150e..7a0b126 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/list.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/_source/list.vue
@@ -20,9 +20,9 @@
<el-table :data="list" size="mini" style="width: 100%">
<el-table-column type="index" :label="$t('#')"
width="50"></el-table-column>
<el-table-column prop="name" :label="$t('Group')"></el-table-column>
- <el-table-column label="IPList" min-width="300">
+ <el-table-column :label="$t('Addresses')" min-width="300">
<template slot-scope="scope">
- <span>{{scope.row.ipList.join(',')}}</span>
+ <span style="display: inline-block; margin-right:
10px">{{scope.row.addrList}}</span>
</template>
</el-table-column>
<el-table-column :label="$t('Create Time')" min-width="120">
@@ -35,6 +35,25 @@
<span>{{scope.row.updateTime | formatDate}}</span>
</template>
</el-table-column>
+ <el-table-column :label="$t('Operation')" width="100">
+ <template slot-scope="scope">
+ <el-tooltip :content="$t('Edit')" placement="top"
v-if="!scope.row.systemDefault">
+ <el-button type="primary" size="mini"
icon="el-icon-edit-outline" @click="_edit(scope.row)" circle></el-button>
+ </el-tooltip>
+ <el-tooltip :content="$t('Delete')" placement="top"
v-if="!scope.row.systemDefault">
+ <el-popconfirm
+ :confirmButtonText="$t('Confirm')"
+ :cancelButtonText="$t('Cancel')"
+ icon="el-icon-info"
+ iconColor="red"
+ :title="$t('Delete?')"
+ @onConfirm="_delete(scope.row,scope.row.id)"
+ >
+ <el-button type="danger" size="mini" icon="el-icon-delete"
circle slot="reference"></el-button>
+ </el-popconfirm>
+ </el-tooltip>
+ </template>
+ </el-table-column>
</el-table>
</div>
</div>
diff --git
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/index.vue
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/index.vue
index 7f8bc6c..144c36c 100644
---
a/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/index.vue
+++
b/dolphinscheduler-ui/src/js/conf/home/pages/security/pages/workerGroups/index.vue
@@ -17,7 +17,18 @@
<template>
<m-list-construction :title="$t('Worker group manage')">
<template slot="conditions">
- <m-conditions @on-conditions="_onConditions"></m-conditions>
+ <m-conditions @on-conditions="_onConditions">
+ <template slot="button-group" v-if="isADMIN">
+ <el-button size="mini" @click="_create('')">{{$t('Create worker
group')}}</el-button>
+ <el-dialog
+ :title="item ? $t('Edit worker group') : $t('Create worker group')"
+ v-if="createWorkerGroupDialog"
+ :visible.sync="createWorkerGroupDialog"
+ width="auto">
+ <m-create-worker :item="item" @onUpdate="onUpdate"
@close="close"></m-create-worker>
+ </el-dialog>
+ </template>
+ </m-conditions>
</template>
<template slot="content">
<template v-if="workerGroupList.length || total>0">
@@ -57,6 +68,7 @@
import listUrlParamHandle from '@/module/mixin/listUrlParamHandle'
import mConditions from '@/module/components/conditions/conditions'
import mListConstruction from
'@/module/components/listConstruction/listConstruction'
+ import mCreateWorker from './_source/createWorker'
export default {
name: 'worker-groups-index',
@@ -70,7 +82,9 @@
pageNo: 1,
searchVal: ''
},
- isADMIN: store.state.user.userInfo.userType === 'ADMIN_USER'
+ isADMIN: store.state.user.userInfo.userType === 'ADMIN_USER',
+ createWorkerGroupDialog: false,
+ item: {}
}
},
mixins: [listUrlParamHandle],
@@ -96,6 +110,17 @@
_onEdit (item) {
this._create(item)
},
+ _create (item) {
+ this.createWorkerGroupDialog = true
+ this.item = item
+ },
+ onUpdate () {
+ this._debounceGET('false')
+ this.createWorkerGroupDialog = false
+ },
+ close () {
+ this.createWorkerGroupDialog = false
+ },
_getList (flag) {
this.isLoading = !flag
this.getWorkerGroups(this.searchParams).then(res => {
@@ -122,6 +147,6 @@
created () {},
mounted () {
},
- components: { mList, mListConstruction, mConditions, mSpin, mNoData }
+ components: { mList, mListConstruction, mConditions, mSpin, mNoData,
mCreateWorker }
}
</script>
diff --git a/dolphinscheduler-ui/src/js/conf/home/store/security/actions.js
b/dolphinscheduler-ui/src/js/conf/home/store/security/actions.js
index 97df05d..72dcd07 100644
--- a/dolphinscheduler-ui/src/js/conf/home/store/security/actions.js
+++ b/dolphinscheduler-ui/src/js/conf/home/store/security/actions.js
@@ -624,7 +624,7 @@ export default {
},
deleteWorkerGroups ({ state }, payload) {
return new Promise((resolve, reject) => {
- io.get('worker-group/delete-by-id', payload, res => {
+ io.post('worker-group/delete-by-id', payload, res => {
resolve(res)
}).catch(e => {
reject(e)
diff --git a/dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
b/dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
index 58fa7e2..a504112 100755
--- a/dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
+++ b/dolphinscheduler-ui/src/js/module/i18n/locale/en_US.js
@@ -131,8 +131,8 @@ export default {
'Please enter app name(optional)': 'Please enter app name(optional)',
'SQL Type': 'SQL Type',
'Send Email': 'Send Email',
- 'Display query result': 'Display query result',
- Rows: 'Rows',
+ 'Log display': 'Log display',
+ 'rows of result': 'rows of result',
Title: 'Title',
'Please enter the title of email': 'Please enter the title of email',
Table: 'Table',
@@ -371,7 +371,7 @@ export default {
'Drag the file into the current upload window': 'Drag the file into the
current upload window',
'Drag area upload': 'Drag area upload',
Upload: 'Upload',
- 'ReUpload File': 'Re-upload file',
+ 'ReUpload File': 'ReUpload File',
'Please enter file name': 'Please enter file name',
'Please select the file to upload': 'Please select the file to upload',
'Resources manage': 'Resources',
@@ -502,8 +502,10 @@ export default {
'Token manage': 'Token manage',
'Create token': 'Create token',
'Edit token': 'Edit token',
- 'Please enter the IP address separated by commas': 'Please enter the IP
address separated by commas',
- 'Note: Multiple IP addresses have been comma separated': 'Note: Multiple IP
addresses have been comma separated',
+ Addresses: 'Addresses',
+ 'Worker Addresses': 'Worker Addresses',
+ 'Please enter the worker addresses separated by commas': 'Please enter the
worker addresses separated by commas',
+ 'Note: Multiple worker addresses have been comma separated': 'Note: Multiple
worker addresses have been comma separated',
'Failure time': 'Failure time',
'Expiration time': 'Expiration time',
User: 'User',
@@ -568,8 +570,8 @@ export default {
'Please Enter Http Url': 'Please Enter Http Url(required)',
'Please Enter Http Condition': 'Please Enter Http Condition',
'There is no data for this period of time': 'There is no data for this
period of time',
- 'IP address cannot be empty': 'IP address cannot be empty',
- 'Please enter the correct IP': 'Please enter the correct IP',
+ 'Worker addresses cannot be empty': 'Worker addresses cannot be empty',
+ 'Please enter the correct worker addresses': 'Please enter the correct
worker addresses',
'Please generate token': 'Please generate token',
'Spark Version': 'Spark Version',
TargetDataBase: 'target database',
diff --git a/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
b/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
index 4d7260b..e9a3603 100755
--- a/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
+++ b/dolphinscheduler-ui/src/js/module/i18n/locale/zh_CN.js
@@ -131,8 +131,8 @@ export default {
'Please enter app name(optional)': '请输入任务名称(选填)',
'SQL Type': 'sql类型',
'Send Email': '发送邮件',
- 'Display query result': '展示查询结果',
- Rows: '行',
+ 'Log display': '日志显示',
+ 'rows of result': '行查询结果',
Title: '主题',
'Please enter the title of email': '请输入邮件主题',
Table: '表名',
@@ -502,8 +502,10 @@ export default {
'Token manage': '令牌管理',
'Create token': '创建令牌',
'Edit token': '编辑令牌',
- 'Please enter the IP address separated by commas': '请输入IP地址多个用英文逗号隔开',
- 'Note: Multiple IP addresses have been comma separated': '注意：多个IP地址以英文逗号分割',
+ Addresses: '地址',
+ 'Worker Addresses': 'Worker地址',
+ 'Please enter the worker addresses separated by commas':
'请输入Worker地址，多个用英文逗号隔开',
+ 'Note: Multiple worker addresses have been comma separated':
'注意：多个Worker地址以英文逗号分割',
'Failure time': '失效时间',
'Expiration time': '失效时间',
User: '用户',
@@ -568,8 +570,8 @@ export default {
'Please Enter Http Url': '请填写请求地址(必填)',
'Please Enter Http Condition': '请填写校验内容',
'There is no data for this period of time': '该时间段无数据',
- 'IP address cannot be empty': 'IP地址不能为空',
- 'Please enter the correct IP': '请输入正确的IP',
+ 'Worker addresses cannot be empty': 'Worker地址不能为空',
+ 'Please enter the correct worker addresses': '请输入正确的Worker地址',
'Please generate token': '请生成Token',
'Spark Version': 'Spark版本',
TargetDataBase: '目标库',
diff --git a/sql/dolphinscheduler_mysql.sql b/sql/dolphinscheduler_mysql.sql
index 1da5c8a..5e71b4a 100644
--- a/sql/dolphinscheduler_mysql.sql
+++ b/sql/dolphinscheduler_mysql.sql
@@ -789,6 +789,22 @@ CREATE TABLE `t_ds_user` (
-- Records of t_ds_user
-- ----------------------------
+-- ----------------------------
+-- Table structure for t_ds_worker_group
+-- ----------------------------
+DROP TABLE IF EXISTS `t_ds_worker_group`;
+CREATE TABLE `t_ds_worker_group` (
+ `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT 'id',
+ `name` varchar(256) NULL DEFAULT NULL COMMENT 'worker group name',
+ `addr_list` text NULL DEFAULT NULL COMMENT 'worker addr list. split by [,]',
+ `create_time` datetime NULL DEFAULT NULL COMMENT 'create time',
+ `update_time` datetime NULL DEFAULT NULL COMMENT 'update time',
+ PRIMARY KEY (`id`)
+) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
+
+-- ----------------------------
+-- Records of t_ds_worker_group
+-- ----------------------------
-- ----------------------------
-- Table structure for t_ds_version
diff --git a/sql/dolphinscheduler_postgre.sql b/sql/dolphinscheduler_postgre.sql
index a3853fb..17bfc70 100644
--- a/sql/dolphinscheduler_postgre.sql
+++ b/sql/dolphinscheduler_postgre.sql
@@ -659,7 +659,7 @@ DROP TABLE IF EXISTS t_ds_worker_group;
CREATE TABLE t_ds_worker_group (
id bigint NOT NULL ,
name varchar(256) DEFAULT NULL ,
- ip_list varchar(256) DEFAULT NULL ,
+ addr_list text DEFAULT NULL ,
create_time timestamp DEFAULT NULL ,
update_time timestamp DEFAULT NULL ,
PRIMARY KEY (id)
diff --git a/sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
b/sql/upgrade/1.3.6_schema/mysql/dolphinscheduler_ddl.sql
similarity index 54%
copy from sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
copy to sql/upgrade/1.3.6_schema/mysql/dolphinscheduler_ddl.sql
index 752e684..1e492a6 100644
--- a/sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
+++ b/sql/upgrade/1.3.6_schema/mysql/dolphinscheduler_ddl.sql
@@ -14,5 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
--- Add foreign key constraints for t_ds_task_instance --
-ALTER TABLE t_ds_task_instance ADD CONSTRAINT foreign_key_instance_id FOREIGN
KEY(process_instance_id) REFERENCES t_ds_process_instance(id) ON DELETE CASCADE;
\ No newline at end of file
+
+SET sql_mode=(SELECT REPLACE(@@sql_mode,'ONLY_FULL_GROUP_BY',''));
+
+-- uc_dolphin_T_t_ds_worker_group_R_ip_list
+drop PROCEDURE if EXISTS uc_dolphin_T_t_ds_worker_group_R_ip_list;
+delimiter d//
+CREATE PROCEDURE uc_dolphin_T_t_ds_worker_group_R_ip_list()
+BEGIN
+ IF EXISTS (SELECT 1 FROM information_schema.COLUMNS
+ WHERE TABLE_NAME='t_ds_worker_group'
+ AND TABLE_SCHEMA=(SELECT DATABASE())
+ AND COLUMN_NAME ='ip_list')
+ THEN
+ ALTER TABLE t_ds_worker_group CHANGE COLUMN `ip_list` `addr_list` text;
+ END IF;
+END;
+
+d//
+
+delimiter ;
+CALL uc_dolphin_T_t_ds_worker_group_R_ip_list;
+DROP PROCEDURE uc_dolphin_T_t_ds_worker_group_R_ip_list;
diff --git a/sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
b/sql/upgrade/1.3.6_schema/mysql/dolphinscheduler_dml.sql
similarity index 78%
copy from sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
copy to sql/upgrade/1.3.6_schema/mysql/dolphinscheduler_dml.sql
index 752e684..38964cc 100644
--- a/sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
+++ b/sql/upgrade/1.3.6_schema/mysql/dolphinscheduler_dml.sql
@@ -13,6 +13,4 @@
* 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.
-*/
--- Add foreign key constraints for t_ds_task_instance --
-ALTER TABLE t_ds_task_instance ADD CONSTRAINT foreign_key_instance_id FOREIGN
KEY(process_instance_id) REFERENCES t_ds_process_instance(id) ON DELETE CASCADE;
\ No newline at end of file
+*/
\ No newline at end of file
diff --git a/sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
b/sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
index 752e684..d39a032 100644
--- a/sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
+++ b/sql/upgrade/1.3.6_schema/postgresql/dolphinscheduler_ddl.sql
@@ -14,5 +14,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
+
+-- uc_dolphin_T_t_ds_worker_group_A_ip_list
+delimiter d//
+CREATE OR REPLACE FUNCTION uc_dolphin_T_t_ds_worker_group_A_ip_list() RETURNS
void AS $$
+BEGIN
+ IF EXISTS (SELECT 1 FROM information_schema.COLUMNS
+ WHERE TABLE_NAME='t_ds_worker_group'
+ AND COLUMN_NAME ='ip_list')
+ THEN
+ ALTER TABLE t_ds_worker_group rename ip_list TO addr_list;
+ ALTER TABLE t_ds_worker_group ALTER column addr_list type text;
+ END IF;
+END;
+$$ LANGUAGE plpgsql;
+d//
+
+delimiter ;
+SELECT uc_dolphin_T_t_ds_worker_group_A_ip_list();
+DROP FUNCTION IF EXISTS uc_dolphin_T_t_ds_worker_group_A_ip_list();
+
-- Add foreign key constraints for t_ds_task_instance --
-ALTER TABLE t_ds_task_instance ADD CONSTRAINT foreign_key_instance_id FOREIGN
KEY(process_instance_id) REFERENCES t_ds_process_instance(id) ON DELETE CASCADE;
\ No newline at end of file
+ALTER TABLE t_ds_task_instance ADD CONSTRAINT foreign_key_instance_id FOREIGN
KEY(process_instance_id) REFERENCES t_ds_process_instance(id) ON DELETE CASCADE;
+
