natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[Feature-2815]cherry pick commit from dev for one worker can belong different worker groups (#3277) 

* [Feature-2815]cherry pick commit from dev for one worker can belong to different worker groups

* fix 2975:download resource need find the tenant code of resource owner

* change NetUtils to OSUtils because NetUtils didn't exist in the 1.3.2-release branch

Co-authored-by: Yichao Yang <1048262223@qq.com>
This commit is contained in:
lgcareer 2020-07-22 19:39:03 +08:00 committed by GitHub
parent 98f5991f1e
commit f854c5f794
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 527 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/MonitorService.java
View File

@ -16,29 +16,33 @@
*/
package org.apache.dolphinscheduler.api.service;
import static org.apache.dolphinscheduler.common.utils.Preconditions.checkNotNull;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.utils.ZookeeperMonitor;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ZKNodeType;
import org.apache.dolphinscheduler.dao.MonitorDBDao;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.common.model.WorkerServerModel;
import org.apache.dolphinscheduler.dao.MonitorDBDao;
import org.apache.dolphinscheduler.dao.entity.MonitorRecord;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.entity.ZookeeperRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.apache.dolphinscheduler.common.utils.Preconditions.*;
import com.google.common.collect.Sets;
/**
* monitor service
*/
@Service
public class MonitorService extends BaseService{
public class MonitorService extends BaseService {
@Autowired
private ZookeeperMonitor zookeeperMonitor;
@ -108,15 +112,41 @@ public class MonitorService extends BaseService{
public Map<String,Object> queryWorker(User loginUser) {
Map<String, Object> result = new HashMap<>(5);
List<Server> masterServers = getServerListFromZK(false);
List<WorkerServerModel> workerServers = getServerListFromZK(false)
.stream()
.map((Server server) -> {
WorkerServerModel model = new WorkerServerModel();
model.setId(server.getId());
model.setHost(server.getHost());
model.setPort(server.getPort());
model.setZkDirectories(Sets.newHashSet(server.getZkDirectory()));
model.setResInfo(server.getResInfo());
model.setCreateTime(server.getCreateTime());
model.setLastHeartbeatTime(server.getLastHeartbeatTime());
return model;
})
.collect(Collectors.toList());
result.put(Constants.DATA_LIST, masterServers);
Map<String, WorkerServerModel> workerHostPortServerMapping = workerServers
.stream()
.collect(Collectors.toMap(
(WorkerServerModel worker) -> {
String[] s = worker.getZkDirectories().iterator().next().split("/");
return s[s.length - 1];
}
, Function.identity()
, (WorkerServerModel oldOne, WorkerServerModel newOne) -> {
oldOne.getZkDirectories().addAll(newOne.getZkDirectories());
return oldOne;
}));
result.put(Constants.DATA_LIST, workerHostPortServerMapping.values());
putMsg(result,Status.SUCCESS);
return result;
}
public List<Server> getServerListFromZK(boolean isMaster){
public List<Server> getServerListFromZK(boolean isMaster) {
checkNotNull(zookeeperMonitor);
ZKNodeType zkNodeType = isMaster ? ZKNodeType.MASTER : ZKNodeType.WORKER;

122
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/model/WorkerServerModel.java Normal file
View File

@ -0,0 +1,122 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.common.model;
import java.util.Date;
import java.util.Set;
import com.fasterxml.jackson.annotation.JsonFormat;
/**
* server
*/
public class WorkerServerModel {
/**
* id
*/
private int id;
/**
* host
*/
private String host;
/**
* port
*/
private int port;
/**
* worker directories in zookeeper
*/
private Set<String> zkDirectories;
/**
* resource info about CPU and memory
*/
private String resInfo;
/**
* create time
*/
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date createTime;
/**
* last heart beat time
*/
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date lastHeartbeatTime;
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public Date getCreateTime() {
return createTime;
}
public void setCreateTime(Date createTime) {
this.createTime = createTime;
}
public Set<String> getZkDirectories() {
return zkDirectories;
}
public void setZkDirectories(Set<String> zkDirectories) {
this.zkDirectories = zkDirectories;
}
public Date getLastHeartbeatTime() {
return lastHeartbeatTime;
}
public void setLastHeartbeatTime(Date lastHeartbeatTime) {
this.lastHeartbeatTime = lastHeartbeatTime;
}
public String getResInfo() {
return resInfo;
}
public void setResInfo(String resInfo) {
this.resInfo = resInfo;
}
public String getHost() {
return host;
}
public void setHost(String host) {
this.host = host;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
}

15
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistry.java
View File

@ -16,6 +16,13 @@
*/
package org.apache.dolphinscheduler.server.master.registry;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
@ -31,11 +38,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.Sets;
import static org.apache.dolphinscheduler.remote.utils.Constants.COMMA;
@ -100,7 +103,7 @@ public class MasterRegistry {
HeartBeatTask heartBeatTask = new HeartBeatTask(startTime,
masterConfig.getMasterReservedMemory(),
masterConfig.getMasterMaxCpuloadAvg(),
getMasterPath(),
Sets.newHashSet(getMasterPath()),
zookeeperRegistryCenter);
this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, masterHeartbeatInterval, masterHeartbeatInterval, TimeUnit.SECONDS);

33
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java
View File

@ -17,50 +17,50 @@
package org.apache.dolphinscheduler.server.registry;
import static org.apache.dolphinscheduler.remote.utils.Constants.COMMA;
import java.util.Date;
import java.util.Set;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Date;
import static org.apache.dolphinscheduler.remote.utils.Constants.COMMA;
public class HeartBeatTask extends Thread{
public class HeartBeatTask extends Thread {
private final Logger logger = LoggerFactory.getLogger(HeartBeatTask.class);
private String startTime;
private double reservedMemory;
private double maxCpuloadAvg;
private String heartBeatPath;
private Set<String> heartBeatPaths;
private ZookeeperRegistryCenter zookeeperRegistryCenter;
public HeartBeatTask(String startTime,
double reservedMemory,
double maxCpuloadAvg,
String heartBeatPath,
ZookeeperRegistryCenter zookeeperRegistryCenter){
Set<String> heartBeatPaths,
ZookeeperRegistryCenter zookeeperRegistryCenter) {
this.startTime = startTime;
this.reservedMemory = reservedMemory;
this.maxCpuloadAvg = maxCpuloadAvg;
this.heartBeatPath = heartBeatPath;
this.heartBeatPaths = heartBeatPaths;
this.zookeeperRegistryCenter = zookeeperRegistryCenter;
}
@Override
public void run() {
try {
double availablePhysicalMemorySize = OSUtils.availablePhysicalMemorySize();
double loadAverage = OSUtils.loadAverage();
int status = Constants.NORAML_NODE_STATUS;
if(availablePhysicalMemorySize < reservedMemory
|| loadAverage > maxCpuloadAvg){
logger.warn("load is too high or availablePhysicalMemorySize(G) is too low, it's availablePhysicalMemorySize(G):{},loadAvg:{}", availablePhysicalMemorySize , loadAverage);
if (availablePhysicalMemorySize < reservedMemory
|| loadAverage > maxCpuloadAvg) {
logger.warn("load is too high or availablePhysicalMemorySize(G) is too low, it's availablePhysicalMemorySize(G):{},loadAvg:{}", availablePhysicalMemorySize, loadAverage);
status = Constants.ABNORMAL_NODE_STATUS;
}
@ -76,8 +76,11 @@ public class HeartBeatTask extends Thread{
builder.append(status).append(COMMA);
//save process id
builder.append(OSUtils.getProcessID());
zookeeperRegistryCenter.getZookeeperCachedOperator().update(heartBeatPath, builder.toString());
} catch (Throwable ex){
for (String heartBeatPath : heartBeatPaths) {
zookeeperRegistryCenter.getZookeeperCachedOperator().update(heartBeatPath, builder.toString());
}
} catch (Throwable ex) {
logger.error("error write heartbeat info", ex);
}
}

14
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/config/WorkerConfig.java
View File

@ -17,6 +17,8 @@
*/
package org.apache.dolphinscheduler.server.worker.config;
import java.util.Set;
import org.apache.dolphinscheduler.common.Constants;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.PropertySource;
@ -41,8 +43,8 @@ public class WorkerConfig {
@Value("${worker.reserved.memory:0.3}")
private double workerReservedMemory;
@Value("${worker.group: default}")
private String workerGroup;
@Value("#{'${worker.groups:default}'.split(',')}")
private Set<String> workerGroups;
@Value("${worker.listen.port: 1234}")
private int listenPort;
@ -55,12 +57,12 @@ public class WorkerConfig {
this.listenPort = listenPort;
}
public String getWorkerGroup() {
return workerGroup;
public Set<String> getWorkerGroups() {
return workerGroups;
}
public void setWorkerGroup(String workerGroup) {
this.workerGroup = workerGroup;
public void setWorkerGroups(Set<String> workerGroups) {
this.workerGroups = workerGroups;
}
public int getWorkerExecThreads() {

113
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistry.java
View File

@ -16,6 +16,17 @@
*/
package org.apache.dolphinscheduler.server.worker.registry;
import static org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
import static org.apache.dolphinscheduler.common.Constants.SLASH;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
@ -32,15 +43,8 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import static org.apache.dolphinscheduler.common.Constants.COMMA;
import static org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
import static org.apache.dolphinscheduler.common.Constants.SLASH;
import com.google.common.collect.Sets;
import com.google.common.collect.Sets;
/**
@ -74,11 +78,11 @@ public class WorkerRegistry {
private String startTime;
private String workerGroup;
private Set<String> workerGroups;
@PostConstruct
public void init(){
this.workerGroup = workerConfig.getWorkerGroup();
this.workerGroups = workerConfig.getWorkerGroups();
this.startTime = DateUtils.dateToString(new Date());
this.heartBeatExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("HeartBeatExecutor"));
}
@ -88,31 +92,35 @@ public class WorkerRegistry {
*/
public void registry() {
String address = OSUtils.getHost();
String localNodePath = getWorkerPath();
zookeeperRegistryCenter.getZookeeperCachedOperator().persistEphemeral(localNodePath, "");
zookeeperRegistryCenter.getZookeeperCachedOperator().getZkClient().getConnectionStateListenable().addListener(new ConnectionStateListener() {
@Override
public void stateChanged(CuratorFramework client, ConnectionState newState) {
if(newState == ConnectionState.LOST){
logger.error("worker : {} connection lost from zookeeper", address);
} else if(newState == ConnectionState.RECONNECTED){
logger.info("worker : {} reconnected to zookeeper", address);
zookeeperRegistryCenter.getZookeeperCachedOperator().persistEphemeral(localNodePath, "");
} else if(newState == ConnectionState.SUSPENDED){
logger.warn("worker : {} connection SUSPENDED ", address);
}
}
});
Set<String> workerZkPaths = getWorkerZkPaths();
int workerHeartbeatInterval = workerConfig.getWorkerHeartbeatInterval();
HeartBeatTask heartBeatTask = new HeartBeatTask(startTime,
workerConfig.getWorkerReservedMemory(),
workerConfig.getWorkerMaxCpuloadAvg(),
getWorkerPath(),
zookeeperRegistryCenter);
this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, workerHeartbeatInterval, workerHeartbeatInterval, TimeUnit.SECONDS);
logger.info("worker node : {} registry to ZK successfully with heartBeatInterval : {}s", address, workerHeartbeatInterval);
for (String workerZKPath : workerZkPaths) {
zookeeperRegistryCenter.getZookeeperCachedOperator().persistEphemeral(workerZKPath, "");
zookeeperRegistryCenter.getZookeeperCachedOperator().getZkClient().getConnectionStateListenable().addListener(new ConnectionStateListener() {
@Override
public void stateChanged(CuratorFramework client, ConnectionState newState) {
if (newState == ConnectionState.LOST) {
logger.error("worker : {} connection lost from zookeeper", address);
} else if (newState == ConnectionState.RECONNECTED) {
logger.info("worker : {} reconnected to zookeeper", address);
zookeeperRegistryCenter.getZookeeperCachedOperator().persistEphemeral(workerZKPath, "");
} else if (newState == ConnectionState.SUSPENDED) {
logger.warn("worker : {} connection SUSPENDED ", address);
}
}
});
logger.info("worker node : {} registry to ZK {} successfully", address, workerZKPath);
}
HeartBeatTask heartBeatTask = new HeartBeatTask(this.startTime,
this.workerConfig.getWorkerReservedMemory(),
this.workerConfig.getWorkerMaxCpuloadAvg(),
workerZkPaths,
this.zookeeperRegistryCenter);
this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, workerHeartbeatInterval, workerHeartbeatInterval, TimeUnit.SECONDS);
logger.info("worker node : {} heartbeat interval {} s", address, workerHeartbeatInterval);
}
/**
@ -120,33 +128,40 @@ public class WorkerRegistry {
*/
public void unRegistry() {
String address = getLocalAddress();
String localNodePath = getWorkerPath();
zookeeperRegistryCenter.getZookeeperCachedOperator().remove(localNodePath);
Set<String> workerZkPaths = getWorkerZkPaths();
for (String workerZkPath : workerZkPaths) {
zookeeperRegistryCenter.getZookeeperCachedOperator().remove(workerZkPath);
logger.info("worker node : {} unRegistry from ZK {}.", address, workerZkPath);
}
this.heartBeatExecutor.shutdownNow();
logger.info("worker node : {} unRegistry to ZK.", address);
}
/**
* get worker path
* @return
*/
private String getWorkerPath() {
private Set<String> getWorkerZkPaths() {
Set<String> workerZkPaths = Sets.newHashSet();
String address = getLocalAddress();
StringBuilder builder = new StringBuilder(100);
String workerPath = this.zookeeperRegistryCenter.getWorkerPath();
builder.append(workerPath).append(SLASH);
if(StringUtils.isEmpty(workerGroup)){
workerGroup = DEFAULT_WORKER_GROUP;
String workerZkPathPrefix = this.zookeeperRegistryCenter.getWorkerPath();
for (String workGroup : this.workerGroups) {
StringBuilder workerZkPathBuilder = new StringBuilder(100);
workerZkPathBuilder.append(workerZkPathPrefix).append(SLASH);
if (StringUtils.isEmpty(workGroup)) {
workGroup = DEFAULT_WORKER_GROUP;
}
// trim and lower case is need
workerZkPathBuilder.append(workGroup.trim().toLowerCase()).append(SLASH);
workerZkPathBuilder.append(address);
workerZkPaths.add(workerZkPathBuilder.toString());
}
//trim and lower case is need
builder.append(workerGroup.trim().toLowerCase()).append(SLASH);
builder.append(address);
return builder.toString();
return workerZkPaths;
}
/**
* get local address
* @return
* get local address
* @return local address
*/
private String getLocalAddress(){
return OSUtils.getHost() + Constants.COLON + workerConfig.getListenPort();

2
dolphinscheduler-server/src/main/resources/worker.properties
View File

@ -31,4 +31,4 @@
#worker.listen.port: 1234
# default worker group
worker.group=default
#worker.groups=default

167
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistryTest.java
View File

@ -17,63 +17,154 @@
package org.apache.dolphinscheduler.server.worker.registry;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.remote.utils.Constants;
import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.zk.SpringZKServer;
import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
import org.apache.dolphinscheduler.service.zk.ZookeeperConfig;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;
import org.apache.curator.framework.imps.CuratorFrameworkImpl;
import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.service.zk.ZookeeperCachedOperator;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Sets;
import static org.apache.dolphinscheduler.common.Constants.HEARTBEAT_FOR_ZOOKEEPER_INFO_LENGTH;
/**
* worker registry test
*/
@RunWith(SpringRunner.class)
@ContextConfiguration(classes={SpringZKServer.class, WorkerRegistry.class,ZookeeperRegistryCenter.class, WorkerConfig.class, ZookeeperCachedOperator.class, ZookeeperConfig.class})
@RunWith(MockitoJUnitRunner.Silent.class)
public class WorkerRegistryTest {
@Autowired
private static final Logger LOGGER = LoggerFactory.getLogger(WorkerRegistryTest.class);
private static final String TEST_WORKER_GROUP = "test";
@InjectMocks
private WorkerRegistry workerRegistry;
@Autowired
@Mock
private ZookeeperRegistryCenter zookeeperRegistryCenter;
@Autowired
@Mock
private ZookeeperCachedOperator zookeeperCachedOperator;
@Mock
private CuratorFrameworkImpl zkClient;
@Mock
private WorkerConfig workerConfig;
@Test
public void testRegistry() throws InterruptedException {
workerRegistry.registry();
String workerPath = zookeeperRegistryCenter.getWorkerPath();
Assert.assertEquals(DEFAULT_WORKER_GROUP, workerConfig.getWorkerGroup().trim());
String instancePath = workerPath + "/" + workerConfig.getWorkerGroup().trim() + "/" + (OSUtils.getHost() + ":" + workerConfig.getListenPort());
TimeUnit.SECONDS.sleep(workerConfig.getWorkerHeartbeatInterval() + 2); //wait heartbeat info write into zk node
String heartbeat = zookeeperRegistryCenter.getZookeeperCachedOperator().get(instancePath);
Assert.assertEquals(HEARTBEAT_FOR_ZOOKEEPER_INFO_LENGTH, heartbeat.split(",").length);
@Before
public void before() {
Set<String> workerGroups = Sets.newHashSet(DEFAULT_WORKER_GROUP, TEST_WORKER_GROUP);
Mockito.when(workerConfig.getWorkerGroups()).thenReturn(workerGroups);
Mockito.when(zookeeperRegistryCenter.getWorkerPath()).thenReturn("/dolphinscheduler/nodes/worker");
Mockito.when(zookeeperRegistryCenter.getZookeeperCachedOperator()).thenReturn(zookeeperCachedOperator);
Mockito.when(zookeeperRegistryCenter.getZookeeperCachedOperator().getZkClient()).thenReturn(zkClient);
Mockito.when(zookeeperRegistryCenter.getZookeeperCachedOperator().getZkClient().getConnectionStateListenable()).thenReturn(
new Listenable<ConnectionStateListener>() {
@Override
public void addListener(ConnectionStateListener connectionStateListener) {
LOGGER.info("add listener");
}
@Override
public void addListener(ConnectionStateListener connectionStateListener, Executor executor) {
LOGGER.info("add listener executor");
}
@Override
public void removeListener(ConnectionStateListener connectionStateListener) {
LOGGER.info("remove listener");
}
});
Mockito.when(workerConfig.getWorkerHeartbeatInterval()).thenReturn(10);
Mockito.when(workerConfig.getWorkerReservedMemory()).thenReturn(1.1);
Mockito.when(workerConfig.getWorkerMaxCpuloadAvg()).thenReturn(1);
}
@Test
public void testUnRegistry() throws InterruptedException {
public void testRegistry() {
workerRegistry.init();
workerRegistry.registry();
TimeUnit.SECONDS.sleep(workerConfig.getWorkerHeartbeatInterval() + 2); //wait heartbeat info write into zk node
String workerPath = zookeeperRegistryCenter.getWorkerPath();
int i = 0;
for (String workerGroup : workerConfig.getWorkerGroups()) {
String workerZkPath = workerPath + "/" + workerGroup.trim() + "/" + (OSUtils.getHost() + ":" + workerConfig.getListenPort());
String heartbeat = zookeeperRegistryCenter.getZookeeperCachedOperator().get(workerZkPath);
if (0 == i) {
Assert.assertTrue(workerZkPath.startsWith("/dolphinscheduler/nodes/worker/test/"));
} else {
Assert.assertTrue(workerZkPath.startsWith("/dolphinscheduler/nodes/worker/default/"));
}
i++;
}
workerRegistry.unRegistry();
workerConfig.getWorkerGroups().add(StringUtils.EMPTY);
workerRegistry.init();
workerRegistry.registry();
workerRegistry.unRegistry();
// testEmptyWorkerGroupsRegistry
workerConfig.getWorkerGroups().remove(StringUtils.EMPTY);
workerConfig.getWorkerGroups().remove(TEST_WORKER_GROUP);
workerConfig.getWorkerGroups().remove(DEFAULT_WORKER_GROUP);
workerRegistry.init();
workerRegistry.registry();
List<String> testWorkerGroupPathZkChildren = zookeeperRegistryCenter.getChildrenKeys(workerPath + "/" + TEST_WORKER_GROUP);
List<String> defaultWorkerGroupPathZkChildren = zookeeperRegistryCenter.getChildrenKeys(workerPath + "/" + DEFAULT_WORKER_GROUP);
Assert.assertEquals(0, testWorkerGroupPathZkChildren.size());
Assert.assertEquals(0, defaultWorkerGroupPathZkChildren.size());
}
@Test
public void testUnRegistry() {
workerRegistry.init();
workerRegistry.registry();
workerRegistry.unRegistry();
String workerPath = zookeeperRegistryCenter.getWorkerPath();
String workerGroupPath = workerPath + "/" + workerConfig.getWorkerGroup().trim();
List<String> childrenKeys = zookeeperRegistryCenter.getZookeeperCachedOperator().getChildrenKeys(workerGroupPath);
Assert.assertTrue(childrenKeys.isEmpty());
for (String workerGroup : workerConfig.getWorkerGroups()) {
String workerGroupPath = workerPath + "/" + workerGroup.trim();
List<String> childrenKeys = zookeeperRegistryCenter.getZookeeperCachedOperator().getChildrenKeys(workerGroupPath);
Assert.assertTrue(childrenKeys.isEmpty());
}
// testEmptyWorkerGroupsUnRegistry
workerConfig.getWorkerGroups().remove(TEST_WORKER_GROUP);
workerConfig.getWorkerGroups().remove(DEFAULT_WORKER_GROUP);
workerRegistry.init();
workerRegistry.registry();
workerRegistry.unRegistry();
}
}

112
dolphinscheduler-ui/src/js/conf/home/pages/monitor/pages/servers/_source/zookeeperDirectories.vue Normal file
View File

@ -0,0 +1,112 @@
/*
* 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.
*/
<template>
<div class="container">
<div class="title-box">
<span class="name">{{$t('zkDirectory')}}</span>
</div>
<div class="table-box" v-if="zkDirectories.length > 0">
<table class="fixed">
<caption><!-- placeHolder --></caption>
<tr>
<th scope="col" style="min-width: 40px">
<span>#</span>
</th>
<th scope="col" style="min-width: 40px">
<span>{{$t('zkDirectory')}}</span>
</th>
</tr>
<tr v-for="(item, $index) in zkDirectories" :key="item.id">
<td>
<span>{{$index + 1}}</span>
</td>
<td>
<span>{{item.zkDirectory}}</span>
</td>
</tr>
</table>
</div>
<div v-if="zkDirectories.length === 0">
<m-no-data><!----></m-no-data>
</div>
<div v-if="zkDirectories.length > 0">
<div class="bottom-box">
</div>
</div>
</div>
</template>
<script>
import mNoData from '@/module/components/noData/noData'
export default {
name: 'zookeeperDirectoriesPopup',
data () {
return {
tableHeaders: [
{
label: $t('zkDirectory'),
prop: 'zkDirectory'
}
]
}
},
props: {
zkDirectories: Array
},
components: { mNoData }
}
</script>
<style lang="scss" rel="stylesheet/scss">
.container {
width: 500px;
.title-box {
height: 61px;
border-bottom: 1px solid #DCDEDC;
position: relative;
.name {
position: absolute;
left: 24px;
top: 18px;
font-size: 16px;
}
}
.bottom-box {
position: absolute;
bottom: 0;
left: 0;
width: 100%;
text-align: right;
height: 60px;
line-height: 60px;
border-top: 1px solid #DCDEDC;
background: #fff;
.ans-page {
display: inline-block;
}
}
.table-box {
overflow-y: scroll;
height: calc(100vh - 61px);
padding-bottom: 60px;
}
}
</style>

27
dolphinscheduler-ui/src/js/conf/home/pages/monitor/pages/servers/worker.vue
View File

@ -22,8 +22,8 @@
<div class="row-title">
<div class="left">
<span class="sp">IP: {{item.host}}</span>
<span class="sp">{{$t('Process Pid')}}: {{item.id}}</span>
<span class="sp">{{$t('Zk registration directory')}}: {{item.zkDirectory}}</span>
<span class="sp">{{$t('Process Pid')}}: {{item.port}}</span>
<span>{{$t('Zk registration directory')}}: <a href="javascript:" @click="_showZkDirectories(item)" class="links">{{$t('Directory detail')}}</a></span>
</div>
<div class="right">
<span class="sp">{{$t('Create Time')}}: {{item.createTime | formatDate}}</span>
@ -74,6 +74,7 @@
import mNoData from '@/module/components/noData/noData'
import themeData from '@/module/echarts/themeData.json'
import mListConstruction from '@/module/components/listConstruction/listConstruction'
import zookeeperDirectoriesPopup from './_source/zookeeperDirectories'
export default {
name: 'servers-worker',
@ -86,7 +87,25 @@
},
props: {},
methods: {
...mapActions('monitor', ['getWorkerData'])
...mapActions('monitor', ['getWorkerData']),
_showZkDirectories (item) {
let zkDirectories = []
item.zkDirectories.forEach(zkDirectory => {
zkDirectories.push({
zkDirectory: zkDirectory
})
})
this.$drawer({
direction: 'right',
render (h) {
return h(zookeeperDirectoriesPopup, {
props: {
zkDirectories: zkDirectories
}
})
}
})
}
},
watch: {},
created () {
@ -105,7 +124,7 @@
this.isLoading = true
})
},
components: { mList, mListConstruction, mSpin, mNoData, mGauge }
components: { mList, mListConstruction, mSpin, mNoData, mGauge, zookeeperDirectoriesPopup }
}
</script>
<style lang="scss" rel="stylesheet/scss">