|
|
@ -14,8 +14,20 @@ |
|
|
|
* See the License for the specific language governing permissions and |
|
|
|
* See the License for the specific language governing permissions and |
|
|
|
* limitations under the License. |
|
|
|
* limitations under the License. |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
|
|
|
|
|
|
|
|
package org.apache.dolphinscheduler.server.master.registry; |
|
|
|
package org.apache.dolphinscheduler.server.master.registry; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.common.utils.DateUtils; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.common.utils.NetUtils; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.server.master.config.MasterConfig; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.server.registry.HeartBeatTask; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import org.apache.curator.framework.CuratorFramework; |
|
|
|
|
|
|
|
import org.apache.curator.framework.state.ConnectionState; |
|
|
|
|
|
|
|
import org.apache.curator.framework.state.ConnectionStateListener; |
|
|
|
|
|
|
|
|
|
|
|
import java.util.Date; |
|
|
|
import java.util.Date; |
|
|
|
import java.util.concurrent.Executors; |
|
|
|
import java.util.concurrent.Executors; |
|
|
|
import java.util.concurrent.ScheduledExecutorService; |
|
|
|
import java.util.concurrent.ScheduledExecutorService; |
|
|
@ -23,15 +35,6 @@ import java.util.concurrent.TimeUnit; |
|
|
|
|
|
|
|
|
|
|
|
import javax.annotation.PostConstruct; |
|
|
|
import javax.annotation.PostConstruct; |
|
|
|
|
|
|
|
|
|
|
|
import org.apache.curator.framework.CuratorFramework; |
|
|
|
|
|
|
|
import org.apache.curator.framework.state.ConnectionState; |
|
|
|
|
|
|
|
import org.apache.curator.framework.state.ConnectionStateListener; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.common.utils.DateUtils; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.common.utils.NetUtils; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.server.master.config.MasterConfig; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.server.registry.HeartBeatTask; |
|
|
|
|
|
|
|
import org.apache.dolphinscheduler.server.registry.ZookeeperRegistryCenter; |
|
|
|
|
|
|
|
import org.slf4j.Logger; |
|
|
|
import org.slf4j.Logger; |
|
|
|
import org.slf4j.LoggerFactory; |
|
|
|
import org.slf4j.LoggerFactory; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
@ -40,7 +43,7 @@ import org.springframework.stereotype.Service; |
|
|
|
import com.google.common.collect.Sets; |
|
|
|
import com.google.common.collect.Sets; |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* master registry |
|
|
|
* master registry |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
@Service |
|
|
|
@Service |
|
|
|
public class MasterRegistry { |
|
|
|
public class MasterRegistry { |
|
|
@ -48,7 +51,7 @@ public class MasterRegistry { |
|
|
|
private final Logger logger = LoggerFactory.getLogger(MasterRegistry.class); |
|
|
|
private final Logger logger = LoggerFactory.getLogger(MasterRegistry.class); |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* zookeeper registry center |
|
|
|
* zookeeper registry center |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
@Autowired |
|
|
|
@Autowired |
|
|
|
private ZookeeperRegistryCenter zookeeperRegistryCenter; |
|
|
|
private ZookeeperRegistryCenter zookeeperRegistryCenter; |
|
|
@ -65,19 +68,18 @@ public class MasterRegistry { |
|
|
|
private ScheduledExecutorService heartBeatExecutor; |
|
|
|
private ScheduledExecutorService heartBeatExecutor; |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* worker start time |
|
|
|
* master start time |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
private String startTime; |
|
|
|
private String startTime; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@PostConstruct |
|
|
|
@PostConstruct |
|
|
|
public void init(){ |
|
|
|
public void init() { |
|
|
|
this.startTime = DateUtils.dateToString(new Date()); |
|
|
|
this.startTime = DateUtils.dateToString(new Date()); |
|
|
|
this.heartBeatExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("HeartBeatExecutor")); |
|
|
|
this.heartBeatExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("HeartBeatExecutor")); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* registry |
|
|
|
* registry |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
public void registry() { |
|
|
|
public void registry() { |
|
|
|
String address = NetUtils.getHost(); |
|
|
|
String address = NetUtils.getHost(); |
|
|
@ -86,12 +88,12 @@ public class MasterRegistry { |
|
|
|
zookeeperRegistryCenter.getZookeeperCachedOperator().getZkClient().getConnectionStateListenable().addListener(new ConnectionStateListener() { |
|
|
|
zookeeperRegistryCenter.getZookeeperCachedOperator().getZkClient().getConnectionStateListenable().addListener(new ConnectionStateListener() { |
|
|
|
@Override |
|
|
|
@Override |
|
|
|
public void stateChanged(CuratorFramework client, ConnectionState newState) { |
|
|
|
public void stateChanged(CuratorFramework client, ConnectionState newState) { |
|
|
|
if(newState == ConnectionState.LOST){ |
|
|
|
if (newState == ConnectionState.LOST) { |
|
|
|
logger.error("master : {} connection lost from zookeeper", address); |
|
|
|
logger.error("master : {} connection lost from zookeeper", address); |
|
|
|
} else if(newState == ConnectionState.RECONNECTED){ |
|
|
|
} else if (newState == ConnectionState.RECONNECTED) { |
|
|
|
logger.info("master : {} reconnected to zookeeper", address); |
|
|
|
logger.info("master : {} reconnected to zookeeper", address); |
|
|
|
zookeeperRegistryCenter.getZookeeperCachedOperator().persistEphemeral(localNodePath, ""); |
|
|
|
zookeeperRegistryCenter.getZookeeperCachedOperator().persistEphemeral(localNodePath, ""); |
|
|
|
} else if(newState == ConnectionState.SUSPENDED){ |
|
|
|
} else if (newState == ConnectionState.SUSPENDED) { |
|
|
|
logger.warn("master : {} connection SUSPENDED ", address); |
|
|
|
logger.warn("master : {} connection SUSPENDED ", address); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
@ -103,36 +105,35 @@ public class MasterRegistry { |
|
|
|
Sets.newHashSet(getMasterPath()), |
|
|
|
Sets.newHashSet(getMasterPath()), |
|
|
|
zookeeperRegistryCenter); |
|
|
|
zookeeperRegistryCenter); |
|
|
|
|
|
|
|
|
|
|
|
this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, masterHeartbeatInterval, masterHeartbeatInterval, TimeUnit.SECONDS); |
|
|
|
this.heartBeatExecutor.scheduleAtFixedRate(heartBeatTask, 0, masterHeartbeatInterval, TimeUnit.SECONDS); |
|
|
|
logger.info("master node : {} registry to ZK successfully with heartBeatInterval : {}s", address, masterHeartbeatInterval); |
|
|
|
logger.info("master node : {} registry to ZK path {} successfully with heartBeatInterval : {}s" |
|
|
|
|
|
|
|
, address, localNodePath, masterHeartbeatInterval); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* remove registry info |
|
|
|
* remove registry info |
|
|
|
*/ |
|
|
|
*/ |
|
|
|
public void unRegistry() { |
|
|
|
public void unRegistry() { |
|
|
|
String address = getLocalAddress(); |
|
|
|
String address = getLocalAddress(); |
|
|
|
String localNodePath = getMasterPath(); |
|
|
|
String localNodePath = getMasterPath(); |
|
|
|
heartBeatExecutor.shutdownNow(); |
|
|
|
heartBeatExecutor.shutdownNow(); |
|
|
|
zookeeperRegistryCenter.getZookeeperCachedOperator().remove(localNodePath); |
|
|
|
zookeeperRegistryCenter.getZookeeperCachedOperator().remove(localNodePath); |
|
|
|
logger.info("master node : {} unRegistry to ZK.", address); |
|
|
|
logger.info("master node : {} unRegistry from ZK path {}." |
|
|
|
|
|
|
|
, address, localNodePath); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* get master path |
|
|
|
* get master path |
|
|
|
* @return |
|
|
|
|
|
|
|
*/ |
|
|
|
*/ |
|
|
|
private String getMasterPath() { |
|
|
|
private String getMasterPath() { |
|
|
|
String address = getLocalAddress(); |
|
|
|
String address = getLocalAddress(); |
|
|
|
String localNodePath = this.zookeeperRegistryCenter.getMasterPath() + "/" + address; |
|
|
|
return this.zookeeperRegistryCenter.getMasterPath() + "/" + address; |
|
|
|
return localNodePath; |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
/** |
|
|
|
/** |
|
|
|
* get local address |
|
|
|
* get local address |
|
|
|
* @return |
|
|
|
|
|
|
|
*/ |
|
|
|
*/ |
|
|
|
private String getLocalAddress(){ |
|
|
|
private String getLocalAddress() { |
|
|
|
|
|
|
|
|
|
|
|
return NetUtils.getHost() + ":" + masterConfig.getListenPort(); |
|
|
|
return NetUtils.getHost() + ":" + masterConfig.getListenPort(); |
|
|
|
|
|
|
|
|
|
|
|