diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterHeartBeatTask.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterHeartBeatTask.java new file mode 100644 index 0000000000..2bcbad2616 --- /dev/null +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterHeartBeatTask.java @@ -0,0 +1,73 @@ +/* + * 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.master.registry; + +import org.apache.dolphinscheduler.common.Constants; +import org.apache.dolphinscheduler.common.utils.HeartBeat; +import org.apache.dolphinscheduler.service.registry.RegistryClient; + +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Master heart beat task + */ +public class MasterHeartBeatTask implements Runnable { + + private final Logger logger = LoggerFactory.getLogger(MasterHeartBeatTask.class); + + private final Set heartBeatPaths; + private final RegistryClient registryClient; + private final HeartBeat heartBeat; + private final AtomicInteger heartBeatErrorTimes = new AtomicInteger(); + + public MasterHeartBeatTask(long startupTime, + double maxCpuloadAvg, + double reservedMemory, + Set heartBeatPaths, + RegistryClient registryClient) { + this.heartBeatPaths = heartBeatPaths; + this.registryClient = registryClient; + this.heartBeat = new HeartBeat(startupTime, maxCpuloadAvg, reservedMemory); + } + + public String getHeartBeatInfo() { + return this.heartBeat.encodeHeartBeat(); + } + + @Override + public void run() { + try { + for (String heartBeatPath : heartBeatPaths) { + if (registryClient.checkIsDeadServer(heartBeatPath, Constants.MASTER_TYPE)) { + registryClient.getStoppable().stop("i was judged to death, release resources and stop myself"); + return; + } + } + for (String heartBeatPath : heartBeatPaths) { + registryClient.persistEphemeral(heartBeatPath, heartBeat.encodeHeartBeat()); + } + heartBeatErrorTimes.set(0); + } catch (Throwable ex) { + logger.error("HeartBeat task execute failed, errorTimes: {}", heartBeatErrorTimes.get(), ex); + } + } +} diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistryClient.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistryClient.java index a59b712828..56167f65b4 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistryClient.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/MasterRegistryClient.java @@ -30,7 +30,6 @@ import org.apache.dolphinscheduler.registry.api.RegistryException; import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory; import org.apache.dolphinscheduler.server.master.config.MasterConfig; import org.apache.dolphinscheduler.server.master.service.FailoverService; -import org.apache.dolphinscheduler.server.registry.HeartBeatTask; import org.apache.dolphinscheduler.service.registry.RegistryClient; import org.apache.commons.lang3.StringUtils; @@ -50,7 +49,7 @@ import com.google.common.collect.Sets; /** *

DolphinScheduler master register client, used to connect to registry and hand the registry events. - *

When the Master node startup, it will register in registry center. And schedule a {@link HeartBeatTask} to update its metadata in registry. + *

When the Master node startup, it will register in registry center. And schedule a {@link MasterHeartBeatTask} to update its metadata in registry. */ @Component public class MasterRegistryClient implements AutoCloseable { @@ -97,7 +96,7 @@ public class MasterRegistryClient implements AutoCloseable { // master registry registry(); registryClient.addConnectionStateListener(new MasterConnectionStateListener(getCurrentNodePath(), - registryClient)); + registryClient)); registryClient.subscribe(REGISTRY_DOLPHINSCHEDULER_NODE, new MasterRegistryDataListener()); } catch (Exception e) { throw new RegistryException("Master registry client start up error", e); @@ -190,13 +189,11 @@ public class MasterRegistryClient implements AutoCloseable { logger.info("Master node : {} registering to registry center", masterAddress); String localNodePath = getCurrentNodePath(); Duration masterHeartbeatInterval = masterConfig.getHeartbeatInterval(); - HeartBeatTask heartBeatTask = new HeartBeatTask(startupTime, - masterConfig.getMaxCpuLoadAvg(), - masterConfig.getReservedMemory(), - Sets.newHashSet(localNodePath), - Constants.MASTER_TYPE, - registryClient, - masterConfig.getHeartbeatErrorThreshold()); + MasterHeartBeatTask heartBeatTask = new MasterHeartBeatTask(startupTime, + masterConfig.getMaxCpuLoadAvg(), + masterConfig.getReservedMemory(), + Sets.newHashSet(localNodePath), + registryClient); // remove before persist registryClient.remove(localNodePath); @@ -247,4 +244,4 @@ public class MasterRegistryClient implements AutoCloseable { return NetUtils.getAddr(masterConfig.getListenPort()); } -} +} \ No newline at end of file diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerHeartBeatTask.java similarity index 63% rename from dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java rename to dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerHeartBeatTask.java index c84abb4182..72315a67b6 100644 --- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/registry/HeartBeatTask.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerHeartBeatTask.java @@ -15,64 +15,43 @@ * limitations under the License. */ -package org.apache.dolphinscheduler.server.registry; +package org.apache.dolphinscheduler.server.worker.registry; +import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.utils.HeartBeat; import org.apache.dolphinscheduler.service.registry.RegistryClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - /** * Heart beat task */ -public class HeartBeatTask implements Runnable { +public class WorkerHeartBeatTask implements Runnable { - private final Logger logger = LoggerFactory.getLogger(HeartBeatTask.class); + private final Logger logger = LoggerFactory.getLogger(WorkerHeartBeatTask.class); private final Set heartBeatPaths; private final RegistryClient registryClient; private int workerWaitingTaskCount; - private final String serverType; private final HeartBeat heartBeat; - private final int heartBeatErrorThreshold; - private final AtomicInteger heartBeatErrorTimes = new AtomicInteger(); - public HeartBeatTask(long startupTime, - double maxCpuloadAvg, - double reservedMemory, - Set heartBeatPaths, - String serverType, - RegistryClient registryClient, - int heartBeatErrorThreshold) { - this.heartBeatPaths = heartBeatPaths; - this.registryClient = registryClient; - this.serverType = serverType; - this.heartBeat = new HeartBeat(startupTime, maxCpuloadAvg, reservedMemory); - this.heartBeatErrorThreshold = heartBeatErrorThreshold; - } - - public HeartBeatTask(long startupTime, + public WorkerHeartBeatTask(long startupTime, double maxCpuloadAvg, double reservedMemory, int hostWeight, Set heartBeatPaths, - String serverType, RegistryClient registryClient, int workerThreadCount, - int workerWaitingTaskCount, - int heartBeatErrorThreshold) { + int workerWaitingTaskCount) { this.heartBeatPaths = heartBeatPaths; this.registryClient = registryClient; this.workerWaitingTaskCount = workerWaitingTaskCount; - this.serverType = serverType; this.heartBeat = new HeartBeat(startupTime, maxCpuloadAvg, reservedMemory, hostWeight, workerThreadCount); - this.heartBeatErrorThreshold = heartBeatErrorThreshold; } public String getHeartBeatInfo() { @@ -82,14 +61,12 @@ public class HeartBeatTask implements Runnable { @Override public void run() { try { - // check dead or not in zookeeper for (String heartBeatPath : heartBeatPaths) { - if (registryClient.checkIsDeadServer(heartBeatPath, serverType)) { + if (registryClient.checkIsDeadServer(heartBeatPath, Constants.WORKER_TYPE)) { registryClient.getStoppable().stop("i was judged to death, release resources and stop myself"); return; } } - // update waiting task count heartBeat.setWorkerWaitingTaskCount(workerWaitingTaskCount); @@ -98,11 +75,7 @@ public class HeartBeatTask implements Runnable { } heartBeatErrorTimes.set(0); } catch (Throwable ex) { - logger.error("HeartBeat task execute failed", ex); - if (heartBeatErrorTimes.incrementAndGet() >= heartBeatErrorThreshold) { - registryClient.getStoppable() - .stop("HeartBeat task connect to zk failed too much times: " + heartBeatErrorTimes); - } + logger.error("HeartBeat task execute failed, errorTimes: {}", heartBeatErrorTimes.get(), ex); } } } diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistryClient.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistryClient.java index 99d18acc30..6aff4dfb19 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistryClient.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/registry/WorkerRegistryClient.java @@ -28,7 +28,6 @@ import org.apache.dolphinscheduler.common.enums.NodeType; import org.apache.dolphinscheduler.common.thread.ThreadUtils; import org.apache.dolphinscheduler.common.utils.NetUtils; import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory; -import org.apache.dolphinscheduler.server.registry.HeartBeatTask; import org.apache.dolphinscheduler.server.worker.config.WorkerConfig; import org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread; import org.apache.dolphinscheduler.service.registry.RegistryClient; @@ -101,16 +100,14 @@ public class WorkerRegistryClient implements AutoCloseable { Set workerZkPaths = getWorkerZkPaths(); long workerHeartbeatInterval = workerConfig.getHeartbeatInterval().getSeconds(); - HeartBeatTask heartBeatTask = new HeartBeatTask(startupTime, - workerConfig.getMaxCpuLoadAvg(), - workerConfig.getReservedMemory(), - workerConfig.getHostWeight(), - workerZkPaths, - Constants.WORKER_TYPE, - registryClient, - workerConfig.getExecThreads(), - workerManagerThread.getThreadPoolQueueSize(), - workerConfig.getHeartbeatErrorThreshold()); + WorkerHeartBeatTask heartBeatTask = new WorkerHeartBeatTask(startupTime, + workerConfig.getMaxCpuLoadAvg(), + workerConfig.getReservedMemory(), + workerConfig.getHostWeight(), + workerZkPaths, + registryClient, + workerConfig.getExecThreads(), + workerManagerThread.getThreadPoolQueueSize()); for (String workerZKPath : workerZkPaths) { // remove before persist @@ -199,4 +196,4 @@ public class WorkerRegistryClient implements AutoCloseable { unRegistry(); } -} +} \ No newline at end of file