Skip to content

Commit ddd8faa

Browse files
CRZbulabulaCRZbulabula
authored
Improve ConfigNode leader warm-up before serving (#17821)
1 parent 07b9cb0 commit ddd8faa

18 files changed

Lines changed: 648 additions & 165 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/ainode/iotdb/ainode/core/constant.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@
8080
class TSStatusCode(Enum):
8181
SUCCESS_STATUS = 200
8282
REDIRECTION_RECOMMEND = 400
83+
CONFIG_NODE_LEADER_WARMING_UP = 1014
8384
MODEL_EXISTED_ERROR = 1503
8485
MODEL_NOT_EXIST_ERROR = 1504
8586
CREATE_MODEL_ERROR = 1505

‎iotdb-core/ainode/iotdb/ainode/core/rpc/client.py‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ def __init__(self, config_leader: TEndPoint):
6666
"Fail to connect to any config node. Please check status of ConfigNodes"
6767
)
6868
self._RETRY_NUM = 10
69+
self._STARTUP_RETRY_NUM = 60
6970
self._RETRY_INTERVAL_IN_S = 1
7071

7172
self._try_to_connect()
@@ -163,6 +164,12 @@ def _update_config_node_leader(self, status: TSStatus) -> bool:
163164
else:
164165
self._config_leader = None
165166
return True
167+
if status.code == TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.get_status_code():
168+
logger.info(
169+
"ConfigNode leader is warming up before serving AINode, will wait and retry. Reason: {}",
170+
status.message,
171+
)
172+
return True
166173
return False
167174

168175
def node_register(
@@ -177,7 +184,7 @@ def node_register(
177184
versionInfo=version_info,
178185
)
179186

180-
for _ in range(0, self._RETRY_NUM):
187+
for _ in range(0, self._STARTUP_RETRY_NUM):
181188
try:
182189
resp = self._client.registerAINode(req)
183190
if not self._update_config_node_leader(resp.status):
@@ -208,7 +215,7 @@ def node_restart(
208215
versionInfo=version_info,
209216
)
210217

211-
for _ in range(0, self._RETRY_NUM):
218+
for _ in range(0, self._STARTUP_RETRY_NUM):
212219
try:
213220
resp = self._client.restartAINode(req)
214221
if not self._update_config_node_leader(resp.status):

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ public void onComplete(TAIHeartbeatResp aiHeartbeatResp) {
5050
public void onError(Exception e) {
5151
if (ThriftClient.isConnectionBroken(e)) {
5252
loadManager.forceUpdateNodeCache(
53-
NodeType.DataNode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown));
53+
NodeType.AINode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown));
5454
}
5555
loadManager.getLoadCache().resetHeartbeatProcessing(nodeId);
5656
}

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

Lines changed: 84 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;
@@ -36,6 +37,7 @@
3637

3738
import org.apache.thrift.async.AsyncMethodCallback;
3839

40+
import java.util.Collections;
3941
import java.util.Map;
4042
import java.util.function.Consumer;
4143

@@ -82,54 +84,86 @@ public DataNodeHeartbeatHandler(
8284

8385
@Override
8486
public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
85-
// Update NodeCache
87+
cacheNodeHeartbeatSample(heartbeatResp);
88+
cacheRegionGroupHeartbeatSamples(heartbeatResp);
89+
cacheUsageSamples(heartbeatResp);
90+
cachePipeHeartbeat(heartbeatResp);
91+
cacheConfirmedConfigNodeEndPoints(heartbeatResp);
92+
cacheRegionSizeSamples(heartbeatResp);
93+
}
94+
95+
private void cacheNodeHeartbeatSample(TDataNodeHeartbeatResp heartbeatResp) {
8696
loadManager
8797
.getLoadCache()
8898
.cacheDataNodeHeartbeatSample(nodeId, new NodeHeartbeatSample(heartbeatResp));
99+
}
89100

101+
private void cacheRegionGroupHeartbeatSamples(TDataNodeHeartbeatResp heartbeatResp) {
90102
RegionStatus regionStatus = RegionStatus.valueOf(heartbeatResp.getStatus());
91103

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-
});
104+
Map<TConsensusGroupId, Boolean> judgedLeaders =
105+
heartbeatResp.isSetJudgedLeaders()
106+
? heartbeatResp.getJudgedLeaders()
107+
: Collections.emptyMap();
108+
judgedLeaders.forEach(
109+
(regionGroupId, isLeader) -> {
110+
cacheRegionHeartbeatSample(heartbeatResp, regionStatus, regionGroupId);
111+
cacheConsensusSampleIfNeeded(heartbeatResp, regionGroupId, isLeader);
112+
});
113+
}
114+
115+
private void cacheRegionHeartbeatSample(
116+
TDataNodeHeartbeatResp heartbeatResp,
117+
RegionStatus dataNodeRegionStatus,
118+
TConsensusGroupId regionGroupId) {
119+
loadManager
120+
.getLoadCache()
121+
.cacheRegionHeartbeatSample(
122+
regionGroupId,
123+
nodeId,
124+
new RegionHeartbeatSample(
125+
heartbeatResp.getHeartbeatTimestamp(),
126+
getRegionHeartbeatStatus(regionGroupId, dataNodeRegionStatus)),
127+
false);
128+
}
129+
130+
private RegionStatus getRegionHeartbeatStatus(
131+
TConsensusGroupId regionGroupId, RegionStatus dataNodeRegionStatus) {
132+
return dataNodeRegionStatus == RegionStatus.Removing
133+
? loadManager.getLoadCache().getRegionCacheLastSampleStatus(regionGroupId, nodeId)
134+
: dataNodeRegionStatus;
135+
}
136+
137+
private void cacheConsensusSampleIfNeeded(
138+
TDataNodeHeartbeatResp heartbeatResp, TConsensusGroupId regionGroupId, Boolean isLeader) {
139+
if (!Boolean.TRUE.equals(isLeader)
140+
|| !shouldCacheConsensusSample(regionGroupId)
141+
|| !hasConsensusLogicalTimestamp(heartbeatResp, regionGroupId)) {
142+
return;
143+
}
144+
145+
loadManager
146+
.getLoadCache()
147+
.cacheConsensusSample(
148+
regionGroupId,
149+
new ConsensusGroupHeartbeatSample(
150+
heartbeatResp.getConsensusLogicalTimeMap().get(regionGroupId), nodeId));
151+
}
152+
153+
private boolean shouldCacheConsensusSample(TConsensusGroupId regionGroupId) {
154+
return (TConsensusGroupType.SchemaRegion.equals(regionGroupId.getType())
155+
&& SCHEMA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE)
156+
|| (TConsensusGroupType.DataRegion.equals(regionGroupId.getType())
157+
&& DATA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE);
158+
}
159+
160+
private boolean hasConsensusLogicalTimestamp(
161+
TDataNodeHeartbeatResp heartbeatResp, TConsensusGroupId regionGroupId) {
162+
return heartbeatResp.isSetConsensusLogicalTimeMap()
163+
&& heartbeatResp.getConsensusLogicalTimeMap().containsKey(regionGroupId);
164+
}
132165

166+
private void cacheUsageSamples(TDataNodeHeartbeatResp heartbeatResp) {
133167
if (heartbeatResp.getRegionDeviceUsageMap() != null) {
134168
deviceNum.putAll(heartbeatResp.getRegionDeviceUsageMap());
135169
deviceUsageRespProcess.accept(heartbeatResp.getRegionDeviceUsageMap());
@@ -141,6 +175,9 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
141175
if (heartbeatResp.getRegionDisk() != null) {
142176
regionDisk.putAll(heartbeatResp.getRegionDisk());
143177
}
178+
}
179+
180+
private void cachePipeHeartbeat(TDataNodeHeartbeatResp heartbeatResp) {
144181
if (heartbeatResp.getPipeMetaList() != null) {
145182
pipeRuntimeCoordinator.parseHeartbeat(
146183
nodeId,
@@ -149,12 +186,18 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
149186
heartbeatResp.getPipeRemainingEventCountList(),
150187
heartbeatResp.getPipeRemainingTimeList());
151188
}
189+
}
190+
191+
private void cacheConfirmedConfigNodeEndPoints(TDataNodeHeartbeatResp heartbeatResp) {
152192
if (heartbeatResp.isSetConfirmedConfigNodeEndPoints()) {
153193
loadManager
154194
.getLoadCache()
155195
.updateConfirmedConfigNodeEndPoints(
156196
nodeId, heartbeatResp.getConfirmedConfigNodeEndPoints());
157197
}
198+
}
199+
200+
private void cacheRegionSizeSamples(TDataNodeHeartbeatResp heartbeatResp) {
158201
if (heartbeatResp.isSetRegionDisk()) {
159202
loadManager.getLoadCache().updateRegionSizeMap(nodeId, heartbeatResp.getRegionDisk());
160203
}

0 commit comments

Comments
 (0)