This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 473070e9c6 [IOTDB-3182] Judge DataNode status through heartbeat(Simple
version) (#6285)
473070e9c6 is described below
commit 473070e9c6bf6d3aff109d8d7f69cbcc0e657ba6
Author: YongzaoDan <[email protected]>
AuthorDate: Thu Jun 16 11:15:47 2022 +0800
[IOTDB-3182] Judge DataNode status through heartbeat(Simple version) (#6285)
---
.../client/handlers/HeartbeatHandler.java | 1 -
.../iotdb/confignode/manager/load/LoadManager.java | 83 ++++++++++++----------
.../manager/load/heartbeat/HeartbeatCache.java | 74 +++++++++++++++----
.../manager/load/heartbeat/HeartbeatPackage.java | 1 +
.../manager/load/heartbeat/HeartbeatWindow.java | 51 -------------
.../load/heartbeat/IHeartbeatStatistic.java | 19 +++--
.../iotdb/confignode/persistence/NodeInfo.java | 2 +
.../apache/iotdb/commons/cluster/NodeStatus.java | 25 ++++---
8 files changed, 133 insertions(+), 123 deletions(-)
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/HeartbeatHandler.java
b/confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/HeartbeatHandler.java
index a1b42599c4..ed4c4e5024 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/HeartbeatHandler.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/HeartbeatHandler.java
@@ -43,7 +43,6 @@ public class HeartbeatHandler implements
AsyncMethodCallback<THeartbeatResp> {
@Override
public void onComplete(THeartbeatResp tHeartbeatResp) {
heartbeatCache.cacheHeartBeat(
- dataNodeLocation.getDataNodeId(),
new HeartbeatPackage(tHeartbeatResp.getHeartbeatTimestamp(),
System.currentTimeMillis()));
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
index 857c188bcc..74453b7d23 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
@@ -18,11 +18,9 @@
*/
package org.apache.iotdb.confignode.manager.load;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.THeartbeatReq;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot;
import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot;
import org.apache.iotdb.commons.partition.DataPartitionTable;
@@ -40,6 +38,7 @@ import org.apache.iotdb.confignode.manager.NodeManager;
import org.apache.iotdb.confignode.manager.load.balancer.PartitionBalancer;
import org.apache.iotdb.confignode.manager.load.balancer.RegionBalancer;
import org.apache.iotdb.confignode.manager.load.heartbeat.HeartbeatCache;
+import org.apache.iotdb.confignode.manager.load.heartbeat.IHeartbeatStatistic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,7 +46,7 @@ import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
@@ -62,24 +61,19 @@ public class LoadManager implements Runnable {
private final long heartbeatInterval =
ConfigNodeDescriptor.getInstance().getConf().getHeartbeatInterval();
- private final HeartbeatCache heartbeatCache;
+ // Map<NodeId, HeartbeatCache>
+ private final Map<Integer, HeartbeatCache> heartbeatCacheMap;
// Balancers
private final RegionBalancer regionBalancer;
private final PartitionBalancer partitionBalancer;
- private final Map<TConsensusGroupId, TRegionReplicaSet> replicaScoreMap;
-
- // TODO: Interfaces for active, interrupt and reset LoadBalancer
-
public LoadManager(Manager configManager) {
this.configManager = configManager;
- this.heartbeatCache = new HeartbeatCache();
+ this.heartbeatCacheMap = new ConcurrentHashMap<>();
this.regionBalancer = new RegionBalancer(configManager);
this.partitionBalancer = new PartitionBalancer(configManager);
-
- this.replicaScoreMap = new TreeMap<>();
}
/**
@@ -134,15 +128,6 @@ public class LoadManager implements Runnable {
return
partitionBalancer.allocateDataPartition(unassignedDataPartitionSlotsMap);
}
- private THeartbeatReq genHeartbeatReq() {
- return new THeartbeatReq(System.currentTimeMillis());
- }
-
- private void doLoadBalancing() {
- // regionExpansion();
- // TODO: update replicaScoreMap
- }
-
@Override
public void run() {
int balanceCount = 0;
@@ -150,25 +135,17 @@ public class LoadManager implements Runnable {
try {
if (getConsensusManager().isLeader()) {
- // Ask DataNode for heartbeat in every heartbeat interval
- List<TDataNodeInfo> onlineDataNodes =
getNodeManager().getOnlineDataNodes(-1);
- for (TDataNodeInfo dataNodeInfo : onlineDataNodes) {
- HeartbeatHandler handler =
- new HeartbeatHandler(dataNodeInfo.getLocation(),
heartbeatCache);
- AsyncDataNodeClientPool.getInstance()
- .getHeartBeat(
- dataNodeInfo.getLocation().getInternalEndPoint(),
genHeartbeatReq(), handler);
- }
+ // Send heartbeat requests to all the online DataNodes
+ pingOnlineDataNodes(getNodeManager().getOnlineDataNodes(-1));
+ // TODO: Send heartbeat requests to all the online ConfigNodes
+ // Do load balancing
+ doLoadBalancing(balanceCount);
balanceCount += 1;
- // TODO: Adjust load balancing period
- if (balanceCount == 10) {
- // Pause load balancing temporary
- // doLoadBalancing();
- balanceCount = 0;
- }
} else {
- heartbeatCache.discardAllCache();
+ // Discard all cache when current ConfigNode is not longer the leader
+ heartbeatCacheMap.clear();
+ balanceCount = 0;
}
TimeUnit.MILLISECONDS.sleep(heartbeatInterval);
@@ -179,6 +156,40 @@ public class LoadManager implements Runnable {
}
}
+ private THeartbeatReq genHeartbeatReq() {
+ return new THeartbeatReq(System.currentTimeMillis());
+ }
+
+ private void doLoadBalancing(int balanceCount) {
+ if (balanceCount % 5 == 0) {
+ // We update nodes' load statistic in every 5s
+ updateNodeLoadStatistic();
+ }
+ }
+
+ private void updateNodeLoadStatistic() {
+
heartbeatCacheMap.values().forEach(IHeartbeatStatistic::updateLoadStatistic);
+ }
+
+ /**
+ * Send heartbeat requests to all the online DataNodes
+ *
+ * @param onlineDataNodes DataNodes that currently online
+ */
+ private void pingOnlineDataNodes(List<TDataNodeInfo> onlineDataNodes) {
+ // Send heartbeat requests
+ for (TDataNodeInfo dataNodeInfo : onlineDataNodes) {
+ HeartbeatHandler handler =
+ new HeartbeatHandler(
+ dataNodeInfo.getLocation(),
+ heartbeatCacheMap.computeIfAbsent(
+ dataNodeInfo.getLocation().getDataNodeId(), empty -> new
HeartbeatCache()));
+ AsyncDataNodeClientPool.getInstance()
+ .getHeartBeat(
+ dataNodeInfo.getLocation().getInternalEndPoint(),
genHeartbeatReq(), handler);
+ }
+ }
+
private ConsensusManager getConsensusManager() {
return configManager.getConsensusManager();
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatCache.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatCache.java
index 29a29ff4f8..92156c06dd 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatCache.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatCache.java
@@ -18,34 +18,78 @@
*/
package org.apache.iotdb.confignode.manager.load.heartbeat;
-import java.util.HashMap;
-import java.util.Map;
+import org.apache.iotdb.commons.cluster.NodeStatus;
+
+import java.util.LinkedList;
/** HeartbeatCache caches and maintains all the heartbeat data */
public class HeartbeatCache implements IHeartbeatStatistic {
- private boolean containsCache = false;
+ // Cache heartbeat samples
+ private static final int maximumWindowSize = 100;
+ private final LinkedList<HeartbeatPackage> slidingWindow;
- // Map<DataNodeId, HeartbeatWindow>
- private final Map<Integer, HeartbeatWindow> windowMap;
+ // For guiding queries, the higher the score the higher the load
+ private volatile float loadScore;
+ // For showing cluster
+ private volatile NodeStatus status;
public HeartbeatCache() {
- this.windowMap = new HashMap<>();
+ this.slidingWindow = new LinkedList<>();
+
+ this.loadScore = 0;
+ this.status = NodeStatus.Running;
+ }
+
+ @Override
+ public void cacheHeartBeat(HeartbeatPackage newHeartbeat) {
+ synchronized (slidingWindow) {
+ // Only sequential heartbeats are accepted.
+ // And un-sequential heartbeats will be discarded.
+ if (slidingWindow.size() == 0
+ || slidingWindow.getLast().getSendTimestamp() <
newHeartbeat.getSendTimestamp()) {
+ slidingWindow.add(newHeartbeat);
+ }
+
+ while (slidingWindow.size() > maximumWindowSize) {
+ slidingWindow.removeFirst();
+ }
+ }
+ }
+
+ @Override
+ public void updateLoadStatistic() {
+ long lastSendTime = 0;
+ synchronized (slidingWindow) {
+ if (slidingWindow.size() > 0) {
+ lastSendTime = slidingWindow.getLast().getSendTimestamp();
+ }
+ }
+
+ // TODO: Optimize
+ loadScore = -lastSendTime;
+ if (System.currentTimeMillis() - lastSendTime > 20_000) {
+ status = NodeStatus.Unknown;
+ } else {
+ status = NodeStatus.Running;
+ }
}
@Override
- public void cacheHeartBeat(int dataNodeId, HeartbeatPackage newHeartbeat) {
- containsCache = true;
- windowMap
- .computeIfAbsent(dataNodeId, window -> new HeartbeatWindow())
- .addHeartbeat(newHeartbeat);
+ public float getLoadScore() {
+ // Return a copy of loadScore
+ return loadScore;
}
@Override
- public void discardAllCache() {
- if (containsCache) {
- containsCache = false;
- windowMap.clear();
+ public NodeStatus getNodeStatus() {
+ // Return a copy of status
+ switch (status) {
+ case Running:
+ return NodeStatus.Running;
+ case Unknown:
+ default:
+ return NodeStatus.Unknown;
}
}
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatPackage.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatPackage.java
index b7bbcb9bbb..66327611aa 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatPackage.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatPackage.java
@@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.manager.load.heartbeat;
public class HeartbeatPackage {
+ // Unit: ms
private final long sendTimestamp;
private final long receiveTimestamp;
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatWindow.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatWindow.java
deleted file mode 100644
index 6478b10a1a..0000000000
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatWindow.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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.iotdb.confignode.manager.load.heartbeat;
-
-import java.util.LinkedList;
-
-/**
- * HeartbeatWindow contains Heartbeat's sending and receiving time, which is
used for estimating
- * when the next heartbeat will arrive.
- */
-public class HeartbeatWindow {
-
- private static final int maximumWindowSize = 1000;
-
- private final LinkedList<HeartbeatPackage> slidingWindow;
-
- public HeartbeatWindow() {
- this.slidingWindow = new LinkedList<>();
- }
-
- public void addHeartbeat(HeartbeatPackage newHeartbeat) {
- synchronized (slidingWindow) {
- // Only sequential heartbeats are accepted.
- // And un-sequential heartbeats will be discarded.
- if (slidingWindow.size() == 0
- || slidingWindow.getLast().getSendTimestamp() <
newHeartbeat.getSendTimestamp()) {
- slidingWindow.add(newHeartbeat);
- }
-
- while (slidingWindow.size() > maximumWindowSize) {
- slidingWindow.removeFirst();
- }
- }
- }
-}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/IHeartbeatStatistic.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/IHeartbeatStatistic.java
index 9cfdb0890d..413f54792a 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/IHeartbeatStatistic.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/IHeartbeatStatistic.java
@@ -18,19 +18,24 @@
*/
package org.apache.iotdb.confignode.manager.load.heartbeat;
-/** All the interfaces that provided by HeartbeatCache */
+import org.apache.iotdb.commons.cluster.NodeStatus;
+
+/** All the statistic interfaces that provided by HeartbeatCache */
public interface IHeartbeatStatistic {
/**
- * Cache the newest HeartbeatData of the specific DataNode
+ * Cache the newest HeartbeatPackage
*
- * @param dataNodeId The specific DataNodeId
* @param newHeartbeat The newest HeartbeatData
*/
- void cacheHeartBeat(int dataNodeId, HeartbeatPackage newHeartbeat);
+ void cacheHeartBeat(HeartbeatPackage newHeartbeat);
+
+ /** Invoking periodically to update node load statistics */
+ void updateLoadStatistic();
- // TODO: Interfaces for statistics
+ /** @return The latest load score of a node, the higher the score the higher
the load */
+ float getLoadScore();
- /** Only use this interface when current ConfigNode is not the leader */
- void discardAllCache();
+ /** @return The latest status of a node for showing cluster */
+ NodeStatus getNodeStatus();
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/NodeInfo.java
b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/NodeInfo.java
index bb8d20966b..fd335f978f 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/NodeInfo.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/NodeInfo.java
@@ -221,6 +221,7 @@ public class NodeInfo implements SnapshotProcessor {
List<TDataNodeInfo> result;
dataNodeInfoReadWriteLock.readLock().lock();
try {
+ // TODO: Check DataNode status, ensure the returned DataNode isn't
removed
if (dataNodeId == -1) {
result = new ArrayList<>(onlineDataNodes.values());
} else {
@@ -276,6 +277,7 @@ public class NodeInfo implements SnapshotProcessor {
List<TConfigNodeLocation> result;
configNodeInfoReadWriteLock.readLock().lock();
try {
+ // TODO: Check ConfigNode status, ensure the returned ConfigNode isn't
removed
result = new ArrayList<>(onlineConfigNodes);
} finally {
configNodeInfoReadWriteLock.readLock().unlock();
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatPackage.java
b/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java
similarity index 64%
copy from
confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatPackage.java
copy to
node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java
index b7bbcb9bbb..08c8d9963d 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/heartbeat/HeartbeatPackage.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java
@@ -16,23 +16,22 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.iotdb.confignode.manager.load.heartbeat;
+package org.apache.iotdb.commons.cluster;
-public class HeartbeatPackage {
+/** Node status for showing cluster */
+public enum NodeStatus {
+ // Node running properly
+ Running("Running"),
+ // Node connection failure
+ Unknown("Unknown");
- private final long sendTimestamp;
- private final long receiveTimestamp;
+ private final String status;
- public HeartbeatPackage(long sendTimestamp, long receiveTimestamp) {
- this.sendTimestamp = sendTimestamp;
- this.receiveTimestamp = receiveTimestamp;
+ NodeStatus(String status) {
+ this.status = status;
}
- public long getSendTimestamp() {
- return sendTimestamp;
- }
-
- public long getReceiveTimestamp() {
- return receiveTimestamp;
+ public String getStatus() {
+ return status;
}
}
