You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by we...@apache.org on 2022/11/25 09:37:37 UTC
[dolphinscheduler] branch dev updated: [Improvement-12907] Change heartbeat log level to debug (#12980)
This is an automated email from the ASF dual-hosted git repository.
wenjun pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 3106054ea7 [Improvement-12907] Change heartbeat log level to debug (#12980)
3106054ea7 is described below
commit 3106054ea78a0036069d1860a228723519efab38
Author: Yann Ann <xi...@gmail.com>
AuthorDate: Fri Nov 25 17:37:30 2022 +0800
[Improvement-12907] Change heartbeat log level to debug (#12980)
---
.../dispatch/host/LowerWeightHostManager.java | 70 +++++++++++-----------
.../server/master/registry/ServerNodeManager.java | 53 +++++++---------
.../server/master/task/MasterHeartBeatTask.java | 2 +-
.../server/worker/task/WorkerHeartBeatTask.java | 5 +-
4 files changed, 60 insertions(+), 70 deletions(-)
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java
index 75bd9c1a0a..5be20c23c1 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java
@@ -99,34 +99,44 @@ public class LowerWeightHostManager extends CommonHostManager {
public void notify(Map<String, Set<String>> workerGroups, Map<String, WorkerHeartBeat> workerNodeInfo) {
syncWorkerResources(workerGroups, workerNodeInfo);
}
- }
- /**
- * Sync worker resource.
- *
- * @param workerGroupNodes worker group nodes, key is worker group, value is worker group nodes.
- * @param workerNodeInfoMap worker node info map, key is worker node, value is worker info.
- */
- private void syncWorkerResources(final Map<String, Set<String>> workerGroupNodes,
- final Map<String, WorkerHeartBeat> workerNodeInfoMap) {
- try {
- Map<String, Set<HostWeight>> workerHostWeights = new HashMap<>();
- for (Map.Entry<String, Set<String>> entry : workerGroupNodes.entrySet()) {
- String workerGroup = entry.getKey();
- Set<String> nodes = entry.getValue();
- Set<HostWeight> hostWeights = new HashSet<>(nodes.size());
- for (String node : nodes) {
- WorkerHeartBeat heartbeat = workerNodeInfoMap.getOrDefault(node, null);
- Optional<HostWeight> hostWeightOpt = getHostWeight(node, workerGroup, heartbeat);
- hostWeightOpt.ifPresent(hostWeights::add);
- }
- if (!hostWeights.isEmpty()) {
- workerHostWeights.put(workerGroup, hostWeights);
+ /**
+ * Sync worker resource.
+ *
+ * @param workerGroupNodes worker group nodes, key is worker group, value is worker group nodes.
+ * @param workerNodeInfoMap worker node info map, key is worker node, value is worker info.
+ */
+ private void syncWorkerResources(final Map<String, Set<String>> workerGroupNodes,
+ final Map<String, WorkerHeartBeat> workerNodeInfoMap) {
+ try {
+ Map<String, Set<HostWeight>> workerHostWeights = new HashMap<>();
+ for (Map.Entry<String, Set<String>> entry : workerGroupNodes.entrySet()) {
+ String workerGroup = entry.getKey();
+ Set<String> nodes = entry.getValue();
+ Set<HostWeight> hostWeights = new HashSet<>(nodes.size());
+ for (String node : nodes) {
+ WorkerHeartBeat heartbeat = workerNodeInfoMap.getOrDefault(node, null);
+ Optional<HostWeight> hostWeightOpt = getHostWeight(node, workerGroup, heartbeat);
+ hostWeightOpt.ifPresent(hostWeights::add);
+ }
+ if (!hostWeights.isEmpty()) {
+ workerHostWeights.put(workerGroup, hostWeights);
+ }
}
+ syncWorkerHostWeight(workerHostWeights);
+ } catch (Throwable ex) {
+ logger.error("Sync worker resource error", ex);
+ }
+ }
+
+ private void syncWorkerHostWeight(Map<String, Set<HostWeight>> workerHostWeights) {
+ lock.lock();
+ try {
+ workerHostWeightsMap.clear();
+ workerHostWeightsMap.putAll(workerHostWeights);
+ } finally {
+ lock.unlock();
}
- syncWorkerHostWeight(workerHostWeights);
- } catch (Throwable ex) {
- logger.error("Sync worker resource error", ex);
}
}
@@ -155,16 +165,6 @@ public class LowerWeightHostManager extends CommonHostManager {
heartBeat.getStartupTime()));
}
- private void syncWorkerHostWeight(Map<String, Set<HostWeight>> workerHostWeights) {
- lock.lock();
- try {
- workerHostWeightsMap.clear();
- workerHostWeightsMap.putAll(workerHostWeights);
- } finally {
- lock.unlock();
- }
- }
-
private Set<HostWeight> getWorkerHostWeights(String workerGroup) {
lock.lock();
try {
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java
index e0ffc34373..def3829c8b 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/registry/ServerNodeManager.java
@@ -129,8 +129,7 @@ public class ServerNodeManager implements InitializingBean {
// load nodes from zookeeper
updateMasterNodes();
- updateWorkerNodes();
- updateWorkerGroupMappings();
+ refreshWorkerNodesAndGroupMappings();
// init executor service
executorService =
@@ -153,27 +152,21 @@ public class ServerNodeManager implements InitializingBean {
@Override
public void run() {
try {
-
// sync worker node info
- updateWorkerNodes();
- updateWorkerGroupMappings();
- notifyWorkerInfoChangeListeners();
+ refreshWorkerNodesAndGroupMappings();
} catch (Exception e) {
logger.error("WorkerNodeInfoAndGroupDbSyncTask error:", e);
}
}
}
- protected Set<String> getWorkerAddressByWorkerGroup(Map<String, String> newWorkerNodeInfo,
- WorkerGroup wg) {
- Set<String> nodes = new HashSet<>();
- String[] addrs = wg.getAddrList().split(Constants.COMMA);
- for (String addr : addrs) {
- if (newWorkerNodeInfo.containsKey(addr)) {
- nodes.add(addr);
- }
- }
- return nodes;
+ /**
+ * Refresh worker nodes and worker group mapping information
+ */
+ private void refreshWorkerNodesAndGroupMappings() {
+ updateWorkerNodes();
+ updateWorkerGroupMappings();
+ notifyWorkerInfoChangeListeners();
}
/**
@@ -204,7 +197,15 @@ public class ServerNodeManager implements InitializingBean {
} catch (Exception ex) {
logger.error("WorkerGroupListener capture data change and get data failed", ex);
}
+ }
+ }
+ private void syncSingleWorkerNodeInfo(String workerAddress, WorkerHeartBeat info) {
+ workerNodeInfoWriteLock.lock();
+ try {
+ workerNodeInfo.put(workerAddress, info);
+ } finally {
+ workerNodeInfoWriteLock.unlock();
}
}
}
@@ -241,8 +242,8 @@ public class ServerNodeManager implements InitializingBean {
try {
registryClient.getLock(nodeLock);
Collection<String> currentNodes = registryClient.getMasterNodesDirectly();
- List<Server> masterNodes = registryClient.getServerList(NodeType.MASTER);
- syncMasterNodes(currentNodes, masterNodes);
+ List<Server> masterNodeList = registryClient.getServerList(NodeType.MASTER);
+ syncMasterNodes(currentNodes, masterNodeList);
} catch (Exception e) {
logger.error("update master nodes error", e);
} finally {
@@ -289,7 +290,6 @@ public class ServerNodeManager implements InitializingBean {
try {
workerGroupNodes.clear();
workerGroupNodes.putAll(tmpWorkerGroupMappings);
- notifyWorkerInfoChangeListeners();
} finally {
workerGroupWriteLock.unlock();
}
@@ -363,15 +363,6 @@ public class ServerNodeManager implements InitializingBean {
}
}
- private void syncSingleWorkerNodeInfo(String workerAddress, WorkerHeartBeat info) {
- workerNodeInfoWriteLock.lock();
- try {
- workerNodeInfo.put(workerAddress, info);
- } finally {
- workerNodeInfoWriteLock.unlock();
- }
- }
-
/**
* Add the resource change listener, when the resource changed, the listener will be notified.
*
@@ -382,10 +373,10 @@ public class ServerNodeManager implements InitializingBean {
}
private void notifyWorkerInfoChangeListeners() {
- Map<String, Set<String>> workerGroupNodes = getWorkerGroupNodes();
- Map<String, WorkerHeartBeat> workerNodeInfo = getWorkerNodeInfo();
+ Map<String, Set<String>> workerGroupNodeMap = getWorkerGroupNodes();
+ Map<String, WorkerHeartBeat> workerNodeInfoMap = getWorkerNodeInfo();
for (WorkerInfoChangeListener listener : workerInfoChangeListeners) {
- listener.notify(workerGroupNodes, workerNodeInfo);
+ listener.notify(workerGroupNodeMap, workerNodeInfoMap);
}
}
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/task/MasterHeartBeatTask.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/task/MasterHeartBeatTask.java
index 45bf0d74cc..e7dad575ec 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/task/MasterHeartBeatTask.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/task/MasterHeartBeatTask.java
@@ -68,7 +68,7 @@ public class MasterHeartBeatTask extends BaseHeartBeatTask<MasterHeartBeat> {
public void writeHeartBeat(MasterHeartBeat masterHeartBeat) {
String masterHeartBeatJson = JSONUtils.toJsonString(masterHeartBeat);
registryClient.persistEphemeral(heartBeatPath, masterHeartBeatJson);
- log.info("Success write master heartBeatInfo into registry, masterRegistryPath: {}, heartBeatInfo: {}",
+ log.debug("Success write master heartBeatInfo into registry, masterRegistryPath: {}, heartBeatInfo: {}",
heartBeatPath, masterHeartBeatJson);
}
}
diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/task/WorkerHeartBeatTask.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/task/WorkerHeartBeatTask.java
index 18b19a9eb7..2909bd686b 100644
--- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/task/WorkerHeartBeatTask.java
+++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/task/WorkerHeartBeatTask.java
@@ -59,9 +59,8 @@ public class WorkerHeartBeatTask extends BaseHeartBeatTask<WorkerHeartBeat> {
double reservedMemory = workerConfig.getReservedMemory();
double availablePhysicalMemorySize = OSUtils.availablePhysicalMemorySize();
int execThreads = workerConfig.getExecThreads();
- int workerWaitingTaskCount = this.workerWaitingTaskCount.get();
int serverStatus = getServerStatus(loadAverage, maxCpuLoadAvg, availablePhysicalMemorySize, reservedMemory,
- execThreads, workerWaitingTaskCount);
+ execThreads, this.workerWaitingTaskCount.get());
return WorkerHeartBeat.builder()
.startupTime(ServerLifeCycleManager.getServerStartupTime())
@@ -86,7 +85,7 @@ public class WorkerHeartBeatTask extends BaseHeartBeatTask<WorkerHeartBeat> {
String workerHeartBeatJson = JSONUtils.toJsonString(workerHeartBeat);
String workerRegistryPath = workerConfig.getWorkerRegistryPath();
registryClient.persistEphemeral(workerRegistryPath, workerHeartBeatJson);
- log.info(
+ log.debug(
"Success write worker group heartBeatInfo into registry, workerRegistryPath: {} workerHeartBeatInfo: {}",
workerRegistryPath, workerHeartBeatJson);
}