Skip to content

Commit 4e6d45f

Browse files
CRZbulabulaCRZbulabula
committed
Improve ConfigNode leader warm-up gating
1 parent 99f0af1 commit 4e6d45f

11 files changed

Lines changed: 435 additions & 49 deletions

File tree

‎iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,7 @@ public enum TSStatusCode {
209209
CAN_NOT_CONNECT_AINODE(1011),
210210
NO_AVAILABLE_REPLICA(1012),
211211
NO_AVAILABLE_AINODE(1013),
212+
CONFIG_NODE_LEADER_WARMING_UP(1014),
212213

213214
// Sync, Load TsFile
214215
LOAD_FILE_ERROR(1100),

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java‎

Lines changed: 52 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
package org.apache.iotdb.confignode.client.async.handlers.heartbeat;
2121

22+
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
2223
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
2324
import org.apache.iotdb.commons.client.ThriftClient;
2425
import org.apache.iotdb.commons.cluster.NodeStatus;
@@ -27,7 +28,6 @@
2728
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
2829
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
2930
import org.apache.iotdb.confignode.manager.load.LoadManager;
30-
import org.apache.iotdb.confignode.manager.load.cache.consensus.ConsensusGroupHeartbeatSample;
3131
import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample;
3232
import org.apache.iotdb.confignode.manager.load.cache.region.RegionHeartbeatSample;
3333
import org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.PipeRuntimeCoordinator;
@@ -36,6 +36,7 @@
3636

3737
import org.apache.thrift.async.AsyncMethodCallback;
3838

39+
import java.util.Collections;
3940
import java.util.Map;
4041
import java.util.function.Consumer;
4142

@@ -89,46 +90,55 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
8990

9091
RegionStatus regionStatus = RegionStatus.valueOf(heartbeatResp.getStatus());
9192

92-
heartbeatResp
93-
.getJudgedLeaders()
94-
.forEach(
95-
(regionGroupId, isLeader) -> {
96-
97-
// Do not allow regions to inherit the Removing state from datanode
98-
RegionStatus nextRegionStatus = regionStatus;
99-
if (nextRegionStatus == RegionStatus.Removing) {
100-
nextRegionStatus =
101-
loadManager
102-
.getLoadCache()
103-
.getRegionCacheLastSampleStatus(regionGroupId, nodeId);
104-
}
105-
106-
// Update RegionGroupCache
107-
loadManager
108-
.getLoadCache()
109-
.cacheRegionHeartbeatSample(
110-
regionGroupId,
111-
nodeId,
112-
new RegionHeartbeatSample(
113-
heartbeatResp.getHeartbeatTimestamp(),
114-
// Region will inherit DataNode's status
115-
nextRegionStatus),
116-
false);
117-
118-
if (((TConsensusGroupType.SchemaRegion.equals(regionGroupId.getType())
119-
&& SCHEMA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE)
120-
|| (TConsensusGroupType.DataRegion.equals(regionGroupId.getType())
121-
&& DATA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE))
122-
&& Boolean.TRUE.equals(isLeader)) {
123-
// Update ConsensusGroupCache when necessary
124-
loadManager
125-
.getLoadCache()
126-
.cacheConsensusSample(
127-
regionGroupId,
128-
new ConsensusGroupHeartbeatSample(
129-
heartbeatResp.getConsensusLogicalTimeMap().get(regionGroupId), nodeId));
130-
}
131-
});
93+
Map<TConsensusGroupId, Boolean> judgedLeaders =
94+
heartbeatResp.isSetJudgedLeaders()
95+
? heartbeatResp.getJudgedLeaders()
96+
: Collections.emptyMap();
97+
judgedLeaders.forEach(
98+
(regionGroupId, isLeader) -> {
99+
100+
// Do not allow regions to inherit the Removing state from datanode
101+
RegionStatus nextRegionStatus = regionStatus;
102+
if (nextRegionStatus == RegionStatus.Removing) {
103+
nextRegionStatus =
104+
loadManager.getLoadCache().getRegionCacheLastSampleStatus(regionGroupId, nodeId);
105+
}
106+
107+
// Update RegionGroupCache
108+
loadManager
109+
.getLoadCache()
110+
.cacheRegionHeartbeatSample(
111+
regionGroupId,
112+
nodeId,
113+
new RegionHeartbeatSample(
114+
heartbeatResp.getHeartbeatTimestamp(),
115+
// Region will inherit DataNode's status
116+
nextRegionStatus),
117+
false);
118+
119+
boolean shouldCacheConsensusSample =
120+
(TConsensusGroupType.SchemaRegion.equals(regionGroupId.getType())
121+
&& SCHEMA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE)
122+
|| (TConsensusGroupType.DataRegion.equals(regionGroupId.getType())
123+
&& DATA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE);
124+
long logicalTimestamp =
125+
heartbeatResp.isSetConsensusLogicalTimeMap()
126+
&& heartbeatResp.getConsensusLogicalTimeMap().containsKey(regionGroupId)
127+
? heartbeatResp.getConsensusLogicalTimeMap().get(regionGroupId)
128+
: heartbeatResp.getHeartbeatTimestamp();
129+
loadManager
130+
.getLoadCache()
131+
.cacheConsensusGroupHeartbeatSample(
132+
regionGroupId,
133+
nodeId,
134+
Boolean.TRUE.equals(isLeader),
135+
logicalTimestamp,
136+
shouldCacheConsensusSample);
137+
});
138+
loadManager
139+
.getLoadCache()
140+
.cacheUnreportedDataNodeRegionHeartbeatSamples(
141+
nodeId, judgedLeaders.keySet(), heartbeatResp.getHeartbeatTimestamp());
132142

133143
if (heartbeatResp.getRegionDeviceUsageMap() != null) {
134144
deviceNum.putAll(heartbeatResp.getRegionDeviceUsageMap());
@@ -170,6 +180,7 @@ public void onError(Exception e) {
170180
if (ThriftClient.isConnectionBroken(e)) {
171181
loadManager.forceUpdateNodeCache(
172182
NodeType.DataNode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown));
183+
loadManager.getLoadCache().cacheDataNodeHeartbeatFailureSample(nodeId, System.nanoTime());
173184
}
174185
loadManager.getLoadCache().resetHeartbeatProcessing(nodeId);
175186
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -288,13 +288,34 @@ public void notifyLeaderReady() {
288288
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
289289
currentNodeTEndPoint);
290290

291-
// Always start load services first
292-
configManager.getLoadManager().startLoadServices();
291+
// Always start load services first and wait for its first full warm-up before serving.
292+
long loadReadyEpoch = configManager.getLoadManager().startLoadServices();
293293

294294
if (CONF.isEnableTopologyProbing()) {
295295
configManager.getLoadManager().startTopologyService();
296296
}
297297

298+
threadPool.submit(() -> startLeaderServicesAfterLoadReady(loadReadyEpoch));
299+
}
300+
301+
private void startLeaderServicesAfterLoadReady(long loadReadyEpoch) {
302+
if (!configManager.getLoadManager().waitForLoadReady(loadReadyEpoch)) {
303+
LOGGER.info(
304+
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_IS_NO_LONGER_THE_LEADER
305+
+ "skip starting leader services because load warm-up is interrupted",
306+
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
307+
currentNodeTEndPoint);
308+
return;
309+
}
310+
if (!configManager.getConsensusManager().isLeaderReady()) {
311+
LOGGER.info(
312+
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_IS_NO_LONGER_THE_LEADER
313+
+ "skip starting leader services because consensus leader is no longer ready",
314+
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
315+
currentNodeTEndPoint);
316+
return;
317+
}
318+
298319
// Start leader scheduling services
299320
configManager.getProcedureManager().startExecutor();
300321
threadPool.submit(

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1247,7 +1247,13 @@ protected TSStatus confirmLeader() {
12471247
"ConsensusManager of target-ConfigNode is not initialized, "
12481248
+ "please make sure the target-ConfigNode has been started successfully.");
12491249
}
1250-
return getConsensusManager().confirmLeader();
1250+
TSStatus status = getConsensusManager().confirmLeader();
1251+
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
1252+
&& !getLoadManager().isLoadReady()) {
1253+
return new TSStatus(TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.getStatusCode())
1254+
.setMessage(getLoadManager().getLoadReadyReason());
1255+
}
1256+
return status;
12511257
}
12521258

12531259
@Override

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java‎

Lines changed: 109 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,17 +48,26 @@
4848
import org.apache.iotdb.confignode.manager.partition.RegionGroupStatus;
4949
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
5050

51+
import java.util.ArrayList;
52+
import java.util.Collections;
5153
import java.util.List;
5254
import java.util.Map;
55+
import java.util.concurrent.TimeUnit;
56+
import java.util.concurrent.atomic.AtomicBoolean;
5357
import java.util.concurrent.atomic.AtomicInteger;
58+
import java.util.concurrent.atomic.AtomicLong;
5459
import java.util.function.Function;
60+
import java.util.stream.Collectors;
5561

5662
/**
5763
* The {@link LoadManager} at ConfigNodeGroup-Leader is active. It proactively implements the
5864
* cluster dynamic load balancing policy and passively accepts the PartitionTable expansion request.
5965
*/
6066
public class LoadManager {
6167

68+
private static final long LOAD_READY_CHECK_INTERVAL_MS =
69+
Math.max(10, Math.min(100, StatisticsService.STATISTICS_UPDATE_INTERVAL / 10));
70+
6271
protected final IManager configManager;
6372

6473
/** Balancers. */
@@ -74,6 +83,10 @@ public class LoadManager {
7483
private final StatisticsService statisticsService;
7584
private final EventService eventService;
7685
private final TopologyService topologyService;
86+
private final AtomicBoolean loadServicesStarted;
87+
private final AtomicLong loadReadyEpoch;
88+
private final AtomicBoolean loadReady;
89+
private volatile String loadReadyReason;
7790

7891
public LoadManager(IManager configManager) {
7992
this.configManager = configManager;
@@ -90,6 +103,10 @@ public LoadManager(IManager configManager) {
90103
this.eventService.register(configManager.getPipeManager().getPipeRuntimeCoordinator());
91104
this.eventService.register(routeBalancer);
92105
this.eventService.register(topologyService);
106+
this.loadServicesStarted = new AtomicBoolean(false);
107+
this.loadReadyEpoch = new AtomicLong(0);
108+
this.loadReady = new AtomicBoolean(false);
109+
this.loadReadyReason = "ConfigNode leader load services are not started.";
93110
}
94111

95112
protected void setHeartbeatService(IManager configManager, LoadCache loadCache) {
@@ -146,15 +163,24 @@ public void reBalanceDataPartitionPolicy(String database) {
146163
partitionBalancer.reBalanceDataPartitionPolicy(database);
147164
}
148165

149-
public void startLoadServices() {
166+
public long startLoadServices() {
167+
long epoch = loadReadyEpoch.incrementAndGet();
168+
loadReady.set(false);
169+
loadReadyReason = "ConfigNode leader is waiting for cluster heartbeat sampling.";
150170
loadCache.initHeartbeatCache(configManager);
171+
loadServicesStarted.set(true);
151172
heartbeatService.startHeartbeatService();
152173
statisticsService.startLoadStatisticsService();
153174
eventService.startEventService();
154175
partitionBalancer.setupPartitionBalancer();
176+
return epoch;
155177
}
156178

157179
public void stopLoadServices() {
180+
loadReadyEpoch.incrementAndGet();
181+
loadServicesStarted.set(false);
182+
loadReady.set(false);
183+
loadReadyReason = "ConfigNode leader load services are stopped.";
158184
heartbeatService.stopHeartbeatService();
159185
statisticsService.stopLoadStatisticsService();
160186
eventService.stopEventService();
@@ -163,6 +189,88 @@ public void stopLoadServices() {
163189
routeBalancer.clearRegionPriority();
164190
}
165191

192+
public boolean waitForLoadReady(long epoch) {
193+
while (epoch == loadReadyEpoch.get() && !Thread.currentThread().isInterrupted()) {
194+
if (tryUpdateLoadReady()) {
195+
return true;
196+
}
197+
try {
198+
TimeUnit.MILLISECONDS.sleep(LOAD_READY_CHECK_INTERVAL_MS);
199+
} catch (InterruptedException e) {
200+
Thread.currentThread().interrupt();
201+
return false;
202+
}
203+
}
204+
return false;
205+
}
206+
207+
public boolean isLoadReady() {
208+
return loadReady.get() || tryUpdateLoadReady();
209+
}
210+
211+
public String getLoadReadyReason() {
212+
return loadReadyReason;
213+
}
214+
215+
private synchronized boolean tryUpdateLoadReady() {
216+
if (loadReady.get()) {
217+
return true;
218+
}
219+
if (!loadServicesStarted.get()) {
220+
loadReadyReason = "ConfigNode leader load services are not started.";
221+
return false;
222+
}
223+
224+
loadCache.updateNodeStatistics(false);
225+
loadCache.updateRegionGroupStatistics();
226+
loadCache.updateConsensusGroupStatistics();
227+
eventService.checkAndBroadcastNodeStatisticsChangeEventIfNecessary();
228+
eventService.checkAndBroadcastRegionGroupStatisticsChangeEventIfNecessary();
229+
eventService.checkAndBroadcastConsensusGroupStatisticsChangeEventIfNecessary();
230+
231+
List<String> unreadyReasons = loadCache.getLoadWarmUpUnreadyReasons();
232+
if (!unreadyReasons.isEmpty()
233+
&& unreadyReasons.stream().anyMatch(reason -> !reason.startsWith("consensusGroups="))) {
234+
loadReadyReason = "ConfigNode leader is warming up LoadCache: " + unreadyReasons;
235+
return false;
236+
}
237+
238+
routeBalancer.balanceRegionLeaderAndPriority();
239+
240+
unreadyReasons = loadCache.getLoadWarmUpUnreadyReasons();
241+
if (!unreadyReasons.isEmpty()) {
242+
loadReadyReason = "ConfigNode leader is warming up LoadCache: " + unreadyReasons;
243+
return false;
244+
}
245+
246+
List<TConsensusGroupId> unreadyRegionPriorities = getUnreadyRegionPriorities();
247+
if (!unreadyRegionPriorities.isEmpty()) {
248+
loadReadyReason =
249+
"ConfigNode leader is warming up region priority: "
250+
+ unreadyRegionPriorities.subList(0, Math.min(10, unreadyRegionPriorities.size()))
251+
+ (unreadyRegionPriorities.size() > 10
252+
? "...(" + (unreadyRegionPriorities.size() - 10) + " more)"
253+
: "");
254+
return false;
255+
}
256+
257+
loadReadyReason = "ConfigNode leader load services are ready.";
258+
loadReady.set(true);
259+
return true;
260+
}
261+
262+
private List<TConsensusGroupId> getUnreadyRegionPriorities() {
263+
List<TConsensusGroupId> regionGroupIds = loadCache.getAllRegionGroupIds();
264+
if (regionGroupIds.isEmpty()) {
265+
return Collections.emptyList();
266+
}
267+
Map<TConsensusGroupId, TRegionReplicaSet> regionPriorityMap =
268+
routeBalancer.getRegionPriorityMap();
269+
return regionGroupIds.stream()
270+
.filter(regionGroupId -> !regionPriorityMap.containsKey(regionGroupId))
271+
.collect(Collectors.toCollection(ArrayList::new));
272+
}
273+
166274
public void startTopologyService() {
167275
topologyService.startTopologyService();
168276
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,10 @@ public AbstractHeartbeatSample getLastSample() {
9797
return slidingWindow.isEmpty() ? null : slidingWindow.get(slidingWindow.size() - 1);
9898
}
9999

100+
public boolean hasHeartbeatSample() {
101+
return getLastSample() != null;
102+
}
103+
100104
/**
101105
* Update currentStatistics based on the latest heartbeat sample that cached in the slidingWindow.
102106
*/

0 commit comments

Comments
 (0)