lgcareer
4 years ago
committed by
GitHub
19 changed files with 939 additions and 503 deletions
@ -0,0 +1,61 @@ |
|||||||
|
/* |
||||||
|
* 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.server.registry; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.service.zk.RegisterOperator; |
||||||
|
import org.apache.dolphinscheduler.service.zk.ZookeeperConfig; |
||||||
|
|
||||||
|
import org.junit.Assert; |
||||||
|
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; |
||||||
|
|
||||||
|
/** |
||||||
|
* zookeeper registry center test |
||||||
|
*/ |
||||||
|
@RunWith(MockitoJUnitRunner.class) |
||||||
|
public class ZookeeperRegistryCenterTest { |
||||||
|
|
||||||
|
@InjectMocks |
||||||
|
private ZookeeperRegistryCenter zookeeperRegistryCenter; |
||||||
|
|
||||||
|
@Mock |
||||||
|
protected RegisterOperator registerOperator; |
||||||
|
|
||||||
|
@Mock |
||||||
|
private ZookeeperConfig zookeeperConfig; |
||||||
|
|
||||||
|
private static final String DS_ROOT = "/dolphinscheduler"; |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testGetDeadZNodeParentPath() { |
||||||
|
ZookeeperConfig zookeeperConfig = new ZookeeperConfig(); |
||||||
|
zookeeperConfig.setDsRoot(DS_ROOT); |
||||||
|
Mockito.when(registerOperator.getZookeeperConfig()).thenReturn(zookeeperConfig); |
||||||
|
|
||||||
|
String deadZNodeParentPath = zookeeperRegistryCenter.getDeadZNodeParentPath(); |
||||||
|
|
||||||
|
Assert.assertEquals(deadZNodeParentPath, DS_ROOT + Constants.ZOOKEEPER_DOLPHINSCHEDULER_DEAD_SERVERS); |
||||||
|
|
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,155 @@ |
|||||||
|
/* |
||||||
|
* 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.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.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.enums.ZKNodeType; |
||||||
|
import org.apache.dolphinscheduler.common.utils.StringUtils; |
||||||
|
|
||||||
|
import java.util.List; |
||||||
|
import java.util.Set; |
||||||
|
|
||||||
|
import org.slf4j.Logger; |
||||||
|
import org.slf4j.LoggerFactory; |
||||||
|
import org.springframework.stereotype.Component; |
||||||
|
|
||||||
|
/** |
||||||
|
* register operator |
||||||
|
*/ |
||||||
|
@Component |
||||||
|
public class RegisterOperator extends ZookeeperCachedOperator { |
||||||
|
|
||||||
|
private final Logger logger = LoggerFactory.getLogger(RegisterOperator.class); |
||||||
|
|
||||||
|
/** |
||||||
|
* @return get dead server node parent path |
||||||
|
*/ |
||||||
|
protected String getDeadZNodeParentPath() { |
||||||
|
return getZookeeperConfig().getDsRoot() + Constants.ZOOKEEPER_DOLPHINSCHEDULER_DEAD_SERVERS; |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* remove dead server by host |
||||||
|
* |
||||||
|
* @param host host |
||||||
|
* @param serverType serverType |
||||||
|
* @throws Exception |
||||||
|
*/ |
||||||
|
public void removeDeadServerByHost(String host, String serverType) throws Exception { |
||||||
|
List<String> deadServers = super.getChildrenKeys(getDeadZNodeParentPath()); |
||||||
|
for (String serverPath : deadServers) { |
||||||
|
if (serverPath.startsWith(serverType + UNDERLINE + host)) { |
||||||
|
String server = getDeadZNodeParentPath() + SINGLE_SLASH + serverPath; |
||||||
|
super.remove(server); |
||||||
|
logger.info("{} server {} deleted from zk dead server path success", serverType, host); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* get host ip, string format: masterParentPath/ip |
||||||
|
* |
||||||
|
* @param path path |
||||||
|
* @return host ip, string format: masterParentPath/ip |
||||||
|
*/ |
||||||
|
protected String getHostByEventDataPath(String path) { |
||||||
|
if (StringUtils.isEmpty(path)) { |
||||||
|
logger.error("empty path!"); |
||||||
|
return ""; |
||||||
|
} |
||||||
|
String[] pathArray = path.split(SINGLE_SLASH); |
||||||
|
if (pathArray.length < 1) { |
||||||
|
logger.error("parse ip error: {}", path); |
||||||
|
return ""; |
||||||
|
} |
||||||
|
return pathArray[pathArray.length - 1]; |
||||||
|
|
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* opType(add): if find dead server , then add to zk deadServerPath |
||||||
|
* opType(delete): delete path from zk |
||||||
|
* |
||||||
|
* @param zNode node path |
||||||
|
* @param zkNodeType master or worker |
||||||
|
* @param opType delete or add |
||||||
|
* @throws Exception errors |
||||||
|
*/ |
||||||
|
public void handleDeadServer(String zNode, ZKNodeType zkNodeType, String opType) throws Exception { |
||||||
|
String host = getHostByEventDataPath(zNode); |
||||||
|
String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_PREFIX : WORKER_PREFIX; |
||||||
|
|
||||||
|
//check server restart, if restart , dead server path in zk should be delete
|
||||||
|
if (opType.equals(DELETE_ZK_OP)) { |
||||||
|
removeDeadServerByHost(host, type); |
||||||
|
|
||||||
|
} else if (opType.equals(ADD_ZK_OP)) { |
||||||
|
String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH + type + UNDERLINE + host; |
||||||
|
if (!super.isExisted(deadServerPath)) { |
||||||
|
//add dead server info to zk dead server path : /dead-servers/
|
||||||
|
|
||||||
|
super.persist(deadServerPath, (type + UNDERLINE + host)); |
||||||
|
|
||||||
|
logger.info("{} server dead , and {} added to zk dead server path success", |
||||||
|
zkNodeType, zNode); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* opType(add): if find dead server , then add to zk deadServerPath |
||||||
|
* opType(delete): delete path from zk |
||||||
|
* |
||||||
|
* @param zNodeSet node path set |
||||||
|
* @param zkNodeType master or worker |
||||||
|
* @param opType delete or add |
||||||
|
* @throws Exception errors |
||||||
|
*/ |
||||||
|
public void handleDeadServer(Set<String> zNodeSet, ZKNodeType zkNodeType, String opType) throws Exception { |
||||||
|
|
||||||
|
String type = (zkNodeType == ZKNodeType.MASTER) ? MASTER_PREFIX : WORKER_PREFIX; |
||||||
|
for (String zNode : zNodeSet) { |
||||||
|
String host = getHostByEventDataPath(zNode); |
||||||
|
//check server restart, if restart , dead server path in zk should be delete
|
||||||
|
if (opType.equals(DELETE_ZK_OP)) { |
||||||
|
removeDeadServerByHost(host, type); |
||||||
|
|
||||||
|
} else if (opType.equals(ADD_ZK_OP)) { |
||||||
|
String deadServerPath = getDeadZNodeParentPath() + SINGLE_SLASH + type + UNDERLINE + host; |
||||||
|
if (!super.isExisted(deadServerPath)) { |
||||||
|
//add dead server info to zk dead server path : /dead-servers/
|
||||||
|
|
||||||
|
super.persist(deadServerPath, (type + UNDERLINE + host)); |
||||||
|
|
||||||
|
logger.info("{} server dead , and {} added to zk dead server path success", |
||||||
|
zkNodeType, zNode); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,116 @@ |
|||||||
|
/* |
||||||
|
* 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.service.zk; |
||||||
|
|
||||||
|
import org.apache.dolphinscheduler.common.Constants; |
||||||
|
import org.apache.dolphinscheduler.common.enums.ZKNodeType; |
||||||
|
|
||||||
|
import java.util.concurrent.TimeUnit; |
||||||
|
|
||||||
|
import org.junit.After; |
||||||
|
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; |
||||||
|
|
||||||
|
/** |
||||||
|
* register operator test |
||||||
|
*/ |
||||||
|
@RunWith(MockitoJUnitRunner.Silent.class) |
||||||
|
public class RegisterOperatorTest { |
||||||
|
|
||||||
|
private static ZKServer zkServer; |
||||||
|
|
||||||
|
@InjectMocks |
||||||
|
private RegisterOperator registerOperator; |
||||||
|
|
||||||
|
@Mock |
||||||
|
private ZookeeperConfig zookeeperConfig; |
||||||
|
|
||||||
|
private static final String DS_ROOT = "/dolphinscheduler"; |
||||||
|
private static final String MASTER_NODE = "127.0.0.1:5678"; |
||||||
|
|
||||||
|
@Before |
||||||
|
public void before() { |
||||||
|
new Thread(() -> { |
||||||
|
if (zkServer == null) { |
||||||
|
zkServer = new ZKServer(); |
||||||
|
} |
||||||
|
zkServer.startLocalZkServer(2185); |
||||||
|
}).start(); |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testAfterPropertiesSet() throws Exception { |
||||||
|
TimeUnit.SECONDS.sleep(10); |
||||||
|
Mockito.when(zookeeperConfig.getServerList()).thenReturn("127.0.0.1:2185"); |
||||||
|
Mockito.when(zookeeperConfig.getBaseSleepTimeMs()).thenReturn(100); |
||||||
|
Mockito.when(zookeeperConfig.getMaxRetries()).thenReturn(10); |
||||||
|
Mockito.when(zookeeperConfig.getMaxSleepMs()).thenReturn(30000); |
||||||
|
Mockito.when(zookeeperConfig.getSessionTimeoutMs()).thenReturn(60000); |
||||||
|
Mockito.when(zookeeperConfig.getConnectionTimeoutMs()).thenReturn(30000); |
||||||
|
Mockito.when(zookeeperConfig.getDigest()).thenReturn(""); |
||||||
|
Mockito.when(zookeeperConfig.getDsRoot()).thenReturn(DS_ROOT); |
||||||
|
Mockito.when(zookeeperConfig.getMaxWaitTime()).thenReturn(30000); |
||||||
|
|
||||||
|
registerOperator.afterPropertiesSet(); |
||||||
|
Assert.assertNotNull(registerOperator.getZkClient()); |
||||||
|
} |
||||||
|
|
||||||
|
@After |
||||||
|
public void after() { |
||||||
|
if (zkServer != null) { |
||||||
|
zkServer.stop(); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testGetDeadZNodeParentPath() throws Exception { |
||||||
|
|
||||||
|
testAfterPropertiesSet(); |
||||||
|
String path = registerOperator.getDeadZNodeParentPath(); |
||||||
|
|
||||||
|
Assert.assertEquals(DS_ROOT + Constants.ZOOKEEPER_DOLPHINSCHEDULER_DEAD_SERVERS, path); |
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testHandleDeadServer() throws Exception { |
||||||
|
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))); |
||||||
|
|
||||||
|
} |
||||||
|
|
||||||
|
@Test |
||||||
|
public void testRemoveDeadServerByHost() throws Exception { |
||||||
|
testAfterPropertiesSet(); |
||||||
|
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))); |
||||||
|
|
||||||
|
registerOperator.removeDeadServerByHost(MASTER_NODE,Constants.MASTER_PREFIX); |
||||||
|
Assert.assertFalse(registerOperator.getChildrenKeys(path).contains(String.format("%s_%s",Constants.MASTER_PREFIX,MASTER_NODE))); |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
Loading…
Reference in new issue