You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/07/11 08:33:31 UTC
[GitHub] [iotdb] SzyWilliam commented on a diff in pull request #6236: [IOTDB-3457] start new feature: data node shrink
SzyWilliam commented on code in PR #6236:
URL: https://github.com/apache/iotdb/pull/6236#discussion_r917645491
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/SyncDataNodeClientPool.java:
##########
@@ -121,6 +125,157 @@ private void deleteRegions(
}
}
+ /**
+ * broadcast the datanode is disabled. when the datanode removed form cluster, will call it.
+ *
+ * @param targetServer the online target datanode.
+ * @param disabledDataNode the disabled datanode, it will be removed
+ * @return TSStatus
+ */
+ public TSStatus disableDataNode(TEndPoint targetServer, TDataNodeLocation disabledDataNode) {
+ LOGGER.info(
+ "send RPC to data node: {} for disabling the data node: {}",
+ targetServer,
+ disabledDataNode);
+ TDisableDataNodeReq req = new TDisableDataNodeReq(disabledDataNode);
+ TSStatus status;
+ try (SyncDataNodeInternalServiceClient client = clientManager.borrowClient(targetServer)) {
+ status = client.disableDataNode(req);
+ } catch (IOException e) {
+ LOGGER.error("Can't connect to Data node: {}", targetServer, e);
+ status = new TSStatus(TSStatusCode.NO_CONNECTION.getStatusCode());
+ status.setMessage(e.getMessage());
+ } catch (TException e) {
+ LOGGER.error(
+ "Disable DataNode {} on target server {} error", disabledDataNode, targetServer, e);
+ status = new TSStatus(TSStatusCode.NODE_DELETE_FAILED_ERROR.getStatusCode());
+ status.setMessage(e.getMessage());
+ }
+ return status;
+ }
+
+ /**
+ * stop datanode
+ *
+ * @param dataNode TDataNodeLocation
+ * @return TSStatus
+ */
+ public TSStatus stopDataNode(TDataNodeLocation dataNode) {
+ LOGGER.info("send RPC to Data Node: {} to stop it", dataNode);
+ TSStatus status;
+ try (SyncDataNodeInternalServiceClient client =
+ clientManager.borrowClient(dataNode.getInternalEndPoint())) {
+ status = client.stopDataNode();
+ } catch (IOException e) {
+ LOGGER.error("Can't connect to Data Node: {}", dataNode, e);
+ status = new TSStatus(TSStatusCode.NO_CONNECTION.getStatusCode());
+ status.setMessage(e.getMessage());
+ } catch (TException e) {
+ LOGGER.error("Stop Data Node {} error", dataNode, e);
+ status = new TSStatus(TSStatusCode.DATANODE_STOP_ERROR.getStatusCode());
+ status.setMessage(e.getMessage());
+ }
+ return status;
+ }
+
+ /**
+ * change a region leader from the datanode to other datanode other datanode should be in same
+ * raft group
+ *
+ * @param regionId the region which will changer leader
+ * @param dataNode data node server, change regions leader from it
+ * @param newLeaderNode target data node, change regions leader to it
+ * @return TSStatus
+ */
+ public TSStatus changeRegionLeader(
+ TConsensusGroupId regionId, TEndPoint dataNode, TDataNodeLocation newLeaderNode) {
+ LOGGER.info("send RPC to data node: {} for changing regions leader on it", dataNode);
+ TSStatus status;
+ try (SyncDataNodeInternalServiceClient client = clientManager.borrowClient(dataNode)) {
+ TRegionLeaderChangeReq req = new TRegionLeaderChangeReq(regionId, newLeaderNode);
+ status = client.changeRegionLeader(req);
+ } catch (IOException e) {
+ LOGGER.error("Can't connect to Data node: {}", dataNode, e);
+ status = new TSStatus(TSStatusCode.NO_CONNECTION.getStatusCode());
+ status.setMessage(e.getMessage());
+ } catch (TException e) {
+ LOGGER.error("Change regions leader error on Date node: {}", dataNode, e);
+ status = new TSStatus(TSStatusCode.REGION_LEADER_CHANGE_FAILED.getStatusCode());
+ status.setMessage(e.getMessage());
+ }
+ return status;
+ }
+
+ public TSStatus addToRegionConsensusGroup(
+ List<TDataNodeLocation> regionOriginalPeerNodes,
+ TConsensusGroupId regionId,
+ TDataNodeLocation newPeerNode,
+ String storageGroup,
+ long ttl) {
+ TSStatus status;
+ // do addConsensusGroup in new node locally
+ try (SyncDataNodeInternalServiceClient client =
+ clientManager.borrowClient(newPeerNode.getInternalEndPoint())) {
+ List<TDataNodeLocation> currentPeerNodes = new ArrayList<>(regionOriginalPeerNodes);
+ currentPeerNodes.add(newPeerNode);
+ TAddConsensusGroup req = new TAddConsensusGroup(regionId, currentPeerNodes, storageGroup);
+ req.setTtl(ttl);
+ status = client.addToRegionConsensusGroup(req);
+ } catch (IOException e) {
+ LOGGER.error("Can't connect to Data Node {}.", newPeerNode.getInternalEndPoint(), e);
+ status = new TSStatus(TSStatusCode.NO_CONNECTION.getStatusCode());
+ status.setMessage(e.getMessage());
+ } catch (TException e) {
+ LOGGER.error(
+ "Add region consensus {} group failed to Date node: {}", regionId, newPeerNode, e);
+ status = new TSStatus(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage(e.getMessage());
+ }
+ return status;
+ }
+
+ private boolean isSucceed(TSStatus status) {
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+
+ private boolean isFailed(TSStatus status) {
+ return !isSucceed(status);
+ }
+
+ /**
+ * migrate a region from the datanode
+ *
+ * @param dataNode region location
+ * @param regionId region id
+ * @param newNode migrate region to it
+ * @param newLeaderNode new leader for the region
+ * @return TSStatus
+ */
+ public TSStatus migrateRegion(
+ TDataNodeLocation dataNode,
+ TConsensusGroupId regionId,
+ TDataNodeLocation newNode,
+ TDataNodeLocation newLeaderNode) {
+ TSStatus status;
+ try (SyncDataNodeInternalServiceClient client =
+ clientManager.borrowClient(dataNode.getInternalEndPoint())) {
+ TMigrateRegionReq req = new TMigrateRegionReq(regionId, dataNode, newNode);
+ if (newLeaderNode != null) {
+ req.setNewLoader(newLeaderNode);
Review Comment:
typo setNewLeader
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/DataNodeRemoveManager.java:
##########
@@ -0,0 +1,793 @@
+/*
+ * 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;
+
+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.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateFailedType;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.enums.DataNodeRemoveState;
+import org.apache.iotdb.commons.enums.RegionMigrateState;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
+import org.apache.iotdb.confignode.consensus.request.write.RemoveDataNodePlan;
+import org.apache.iotdb.confignode.consensus.request.write.UpdateRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.response.DataNodeToStatusResp;
+import org.apache.iotdb.confignode.persistence.NodeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
+import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
+
+public class DataNodeRemoveManager {
+ private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeRemoveManager.class);
+ private static final int QUEUE_SIZE_LIMIT = 200;
+
+ private boolean hasSetUp = false;
+
+ private ConfigManager configManager;
+
+ private final LinkedBlockingQueue<RemoveDataNodePlan> removeQueue;
+
+ // which request is running
+ private RemoveDataNodePlan headRequest = null;
+
+ // which data node is removing in `headRequest`, [0, req TDataNodeLocation list size)
+ private int headNodeIndex = -1;
+
+ private DataNodeRemoveState headNodeState = DataNodeRemoveState.NORMAL;
+
+ // the region ids belongs to head node
+ private List<TConsensusGroupId> headNodeRegionIds = new ArrayList<>();
+
+ // which region is migrating on head node
+ private int headRegionIndex = -1;
+
+ private RegionMigrateState headRegionState = RegionMigrateState.ONLINE;
+
+ private volatile boolean stopped = false;
+
+ private volatile boolean isLeader = false;
+ private final Object leaderLock = new Object();
+ private Thread workThread;
+ private Thread waitLeaderThread;
+
+ private final Object regionMigrateLock = new Object();
+ private TRegionMigrateResultReportReq lastRegionMigrateResult = null;
+
+ public DataNodeRemoveManager(ConfigManager configManager) {
+ this.configManager = configManager;
+ removeQueue = new LinkedBlockingQueue<>(QUEUE_SIZE_LIMIT);
+
+ createWaitLeaderThread();
+ createWorkThread();
+ }
+
+ /** start the manager when Config node startup */
+ public void start() {
+ if (!hasSetUp) {
+ // TODO 1. when restart，reload info from NoInfo and continue
+ setUp();
+ hasSetUp = true;
+ }
+ LOGGER.info("Data Node remove service start");
+ // 2. if it is not leader, loop check
+ // configManager.getConsensusManager().isLeader())
+ waitUntilBeLeader();
+
+ // 3. Take and exec request one by one
+ // 4. When Data Node's state or Region's state change, then modify the request and write it to
+ // Consensus.
+ // 5. TODO if leader change?
+ execRequestFormQueue();
+ }
+
+ private void createWaitLeaderThread() {
+ waitLeaderThread =
+ new Thread(
+ () -> {
+ while (!stopped && !configManager.getConsensusManager().isLeader()) {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ LOGGER.info("the config node is leader now!");
+ isLeader = true;
+ synchronized (leaderLock) {
+ leaderLock.notify();
+ }
+ });
+ waitLeaderThread.setName("Wait-Leader-Thread");
+ }
+
+ private void createWorkThread() {
+ workThread =
+ new Thread(
+ () -> {
+ while (!stopped) {
+ RemoveDataNodePlan req = null;
+ try {
+ while (!isLeader) {
+ LOGGER.warn("the ConfigNode is not leader, waiting...");
+ synchronized (leaderLock) {
+ leaderLock.wait();
+ }
+ }
+ LOGGER.info("the ConfigNode is leader now, will take request and run");
+ req = removeQueue.take();
+ LOGGER.info("exec the request : {}", req);
+ prepareHeadRequestInfo(req);
+ TSStatus status = execRemoveDataNodeRequest(headRequest);
+ LOGGER.info("exec the request: {}, result: {}", req, status);
+ // unRegisterRequest(req);
+ } catch (InterruptedException e) {
+ LOGGER.warn("work thread interrupted", e);
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ LOGGER.warn("the request run failed", e);
+ } finally {
+ if (req != null) {
+ unRegisterRequest(req);
+ }
+ }
+ }
+ });
+ workThread.setName("Exec-RemoveDataNode-Thread");
+ }
+
+ private void waitUntilBeLeader() {
+ if (waitLeaderThread != null) {
+ waitLeaderThread.start();
+ }
+ }
+
+ private void execRequestFormQueue() {
Review Comment:
typo: execRequestFromQueue
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/DataNodeRemoveManager.java:
##########
@@ -0,0 +1,793 @@
+/*
+ * 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;
+
+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.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateFailedType;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.enums.DataNodeRemoveState;
+import org.apache.iotdb.commons.enums.RegionMigrateState;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
+import org.apache.iotdb.confignode.consensus.request.write.RemoveDataNodePlan;
+import org.apache.iotdb.confignode.consensus.request.write.UpdateRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.response.DataNodeToStatusResp;
+import org.apache.iotdb.confignode.persistence.NodeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
+import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
+
+public class DataNodeRemoveManager {
+ private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeRemoveManager.class);
+ private static final int QUEUE_SIZE_LIMIT = 200;
+
+ private boolean hasSetUp = false;
+
+ private ConfigManager configManager;
+
+ private final LinkedBlockingQueue<RemoveDataNodePlan> removeQueue;
+
+ // which request is running
+ private RemoveDataNodePlan headRequest = null;
+
+ // which data node is removing in `headRequest`, [0, req TDataNodeLocation list size)
+ private int headNodeIndex = -1;
+
+ private DataNodeRemoveState headNodeState = DataNodeRemoveState.NORMAL;
+
+ // the region ids belongs to head node
+ private List<TConsensusGroupId> headNodeRegionIds = new ArrayList<>();
+
+ // which region is migrating on head node
+ private int headRegionIndex = -1;
+
+ private RegionMigrateState headRegionState = RegionMigrateState.ONLINE;
+
+ private volatile boolean stopped = false;
+
+ private volatile boolean isLeader = false;
+ private final Object leaderLock = new Object();
+ private Thread workThread;
+ private Thread waitLeaderThread;
+
+ private final Object regionMigrateLock = new Object();
+ private TRegionMigrateResultReportReq lastRegionMigrateResult = null;
+
+ public DataNodeRemoveManager(ConfigManager configManager) {
+ this.configManager = configManager;
+ removeQueue = new LinkedBlockingQueue<>(QUEUE_SIZE_LIMIT);
+
+ createWaitLeaderThread();
+ createWorkThread();
+ }
+
+ /** start the manager when Config node startup */
+ public void start() {
+ if (!hasSetUp) {
+ // TODO 1. when restart，reload info from NoInfo and continue
+ setUp();
+ hasSetUp = true;
+ }
+ LOGGER.info("Data Node remove service start");
+ // 2. if it is not leader, loop check
+ // configManager.getConsensusManager().isLeader())
+ waitUntilBeLeader();
+
+ // 3. Take and exec request one by one
+ // 4. When Data Node's state or Region's state change, then modify the request and write it to
+ // Consensus.
+ // 5. TODO if leader change?
+ execRequestFormQueue();
+ }
+
+ private void createWaitLeaderThread() {
+ waitLeaderThread =
+ new Thread(
+ () -> {
+ while (!stopped && !configManager.getConsensusManager().isLeader()) {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ LOGGER.info("the config node is leader now!");
+ isLeader = true;
+ synchronized (leaderLock) {
+ leaderLock.notify();
+ }
+ });
+ waitLeaderThread.setName("Wait-Leader-Thread");
+ }
+
+ private void createWorkThread() {
+ workThread =
+ new Thread(
+ () -> {
+ while (!stopped) {
+ RemoveDataNodePlan req = null;
+ try {
+ while (!isLeader) {
+ LOGGER.warn("the ConfigNode is not leader, waiting...");
+ synchronized (leaderLock) {
+ leaderLock.wait();
+ }
+ }
+ LOGGER.info("the ConfigNode is leader now, will take request and run");
+ req = removeQueue.take();
+ LOGGER.info("exec the request : {}", req);
+ prepareHeadRequestInfo(req);
+ TSStatus status = execRemoveDataNodeRequest(headRequest);
+ LOGGER.info("exec the request: {}, result: {}", req, status);
+ // unRegisterRequest(req);
+ } catch (InterruptedException e) {
+ LOGGER.warn("work thread interrupted", e);
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ LOGGER.warn("the request run failed", e);
+ } finally {
+ if (req != null) {
+ unRegisterRequest(req);
+ }
+ }
+ }
+ });
+ workThread.setName("Exec-RemoveDataNode-Thread");
+ }
+
+ private void waitUntilBeLeader() {
+ if (waitLeaderThread != null) {
+ waitLeaderThread.start();
+ }
+ }
+
+ private void execRequestFormQueue() {
+ if (workThread != null) {
+ workThread.start();
+ }
+ }
+
+ public void beLeader() {
+ // TODO
+ /** this.isLeader = true; synchronized (leaderLock) { leaderLock.notify(); } */
+ }
+
+ public void beFollower() {
+ // TODO
+ /** this.isLeader = false; */
+ }
+
+ /**
+ * prepare for the request. from first node, first region to exec
+ *
+ * @param req RemoveDataNodeReq
+ */
+ private void prepareHeadRequestInfo(RemoveDataNodePlan req) {
+ LOGGER.info("start to prepare for request: {}", req);
+ this.headRequest = req;
+ // to exec the request, will change it's state at different stage.
+ this.headNodeIndex = 0;
+ this.headNodeState = DataNodeRemoveState.NORMAL;
+
+ TDataNodeLocation headNode = req.getDataNodeLocations().get(headNodeIndex);
+ this.headNodeRegionIds =
+ configManager.getPartitionManager().getAllReplicaSets().stream()
+ .filter(rg -> rg.getDataNodeLocations().contains(headNode))
+ .filter(rg -> rg.regionId.getType() != TConsensusGroupType.PartitionRegion)
+ .map(TRegionReplicaSet::getRegionId)
+ .collect(Collectors.toList());
+ this.headRegionIndex = 0;
+ this.headRegionState = RegionMigrateState.ONLINE;
+
+ // modify head quest
+ this.headRequest.setUpdate(true);
+ this.headRequest.setExecDataNodeRegionIds(headNodeRegionIds);
+ this.headRequest.setExecDataNodeIndex(headNodeIndex);
+ this.headRequest.setExecRegionIndex(headRegionIndex);
+ this.headRequest.setExecDataNodeState(headNodeState);
+ this.headRequest.setExecRegionState(headRegionState);
+ configManager.getConsensusManager().write(headRequest);
+ LOGGER.info("finished to prepare for request: {}", req);
+ }
+
+ private void setUp() {
+ removeQueue.addAll(configManager.getNodeManager().getDataNodeRemoveRequestQueue());
+ headRequest = configManager.getNodeManager().getHeadRequestForDataNodeRemove();
+ if (headRequest == null) {
+ return;
+ }
+
+ // avoid duplication
+ if (removeQueue.contains(headRequest)) {
+ removeQueue.remove(headRequest);
+ }
+
+ headNodeIndex = headRequest.getExecDataNodeIndex();
+ headNodeState = headRequest.getExecDataNodeState();
+ headNodeRegionIds = headRequest.getExecDataNodeRegionIds();
+ headRegionIndex = headRequest.getExecRegionIndex();
+ headRegionState = headRequest.getExecRegionState();
+ }
+
+ /**
+ * exec the request loop all removed node 1: brocast it to cluster 2: loop region on it 2.1 change
Review Comment:
The code doesn't have 2.1 change region leader. we can delete comment 2.1 so it corresponds to code behavior.
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/DataNodeRemoveManager.java:
##########
@@ -0,0 +1,793 @@
+/*
+ * 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;
+
+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.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateFailedType;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.enums.DataNodeRemoveState;
+import org.apache.iotdb.commons.enums.RegionMigrateState;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
+import org.apache.iotdb.confignode.consensus.request.write.RemoveDataNodePlan;
+import org.apache.iotdb.confignode.consensus.request.write.UpdateRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.response.DataNodeToStatusResp;
+import org.apache.iotdb.confignode.persistence.NodeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
+import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
+
+public class DataNodeRemoveManager {
+ private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeRemoveManager.class);
+ private static final int QUEUE_SIZE_LIMIT = 200;
+
+ private boolean hasSetUp = false;
+
+ private ConfigManager configManager;
+
+ private final LinkedBlockingQueue<RemoveDataNodePlan> removeQueue;
+
+ // which request is running
+ private RemoveDataNodePlan headRequest = null;
+
+ // which data node is removing in `headRequest`, [0, req TDataNodeLocation list size)
+ private int headNodeIndex = -1;
+
+ private DataNodeRemoveState headNodeState = DataNodeRemoveState.NORMAL;
+
+ // the region ids belongs to head node
+ private List<TConsensusGroupId> headNodeRegionIds = new ArrayList<>();
+
+ // which region is migrating on head node
+ private int headRegionIndex = -1;
+
+ private RegionMigrateState headRegionState = RegionMigrateState.ONLINE;
+
+ private volatile boolean stopped = false;
+
+ private volatile boolean isLeader = false;
+ private final Object leaderLock = new Object();
+ private Thread workThread;
+ private Thread waitLeaderThread;
+
+ private final Object regionMigrateLock = new Object();
+ private TRegionMigrateResultReportReq lastRegionMigrateResult = null;
+
+ public DataNodeRemoveManager(ConfigManager configManager) {
+ this.configManager = configManager;
+ removeQueue = new LinkedBlockingQueue<>(QUEUE_SIZE_LIMIT);
+
+ createWaitLeaderThread();
+ createWorkThread();
+ }
+
+ /** start the manager when Config node startup */
+ public void start() {
+ if (!hasSetUp) {
+ // TODO 1. when restart，reload info from NoInfo and continue
+ setUp();
+ hasSetUp = true;
+ }
+ LOGGER.info("Data Node remove service start");
+ // 2. if it is not leader, loop check
+ // configManager.getConsensusManager().isLeader())
+ waitUntilBeLeader();
+
+ // 3. Take and exec request one by one
+ // 4. When Data Node's state or Region's state change, then modify the request and write it to
+ // Consensus.
+ // 5. TODO if leader change?
+ execRequestFormQueue();
+ }
+
+ private void createWaitLeaderThread() {
+ waitLeaderThread =
+ new Thread(
+ () -> {
+ while (!stopped && !configManager.getConsensusManager().isLeader()) {
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ LOGGER.info("the config node is leader now!");
+ isLeader = true;
+ synchronized (leaderLock) {
+ leaderLock.notify();
+ }
+ });
+ waitLeaderThread.setName("Wait-Leader-Thread");
+ }
+
+ private void createWorkThread() {
+ workThread =
+ new Thread(
+ () -> {
+ while (!stopped) {
+ RemoveDataNodePlan req = null;
+ try {
+ while (!isLeader) {
+ LOGGER.warn("the ConfigNode is not leader, waiting...");
+ synchronized (leaderLock) {
+ leaderLock.wait();
+ }
+ }
+ LOGGER.info("the ConfigNode is leader now, will take request and run");
+ req = removeQueue.take();
+ LOGGER.info("exec the request : {}", req);
+ prepareHeadRequestInfo(req);
+ TSStatus status = execRemoveDataNodeRequest(headRequest);
+ LOGGER.info("exec the request: {}, result: {}", req, status);
+ // unRegisterRequest(req);
+ } catch (InterruptedException e) {
+ LOGGER.warn("work thread interrupted", e);
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ LOGGER.warn("the request run failed", e);
+ } finally {
+ if (req != null) {
+ unRegisterRequest(req);
+ }
+ }
+ }
+ });
+ workThread.setName("Exec-RemoveDataNode-Thread");
+ }
+
+ private void waitUntilBeLeader() {
+ if (waitLeaderThread != null) {
+ waitLeaderThread.start();
+ }
+ }
+
+ private void execRequestFormQueue() {
+ if (workThread != null) {
+ workThread.start();
+ }
+ }
+
+ public void beLeader() {
+ // TODO
+ /** this.isLeader = true; synchronized (leaderLock) { leaderLock.notify(); } */
+ }
+
+ public void beFollower() {
+ // TODO
+ /** this.isLeader = false; */
+ }
+
+ /**
+ * prepare for the request. from first node, first region to exec
+ *
+ * @param req RemoveDataNodeReq
+ */
+ private void prepareHeadRequestInfo(RemoveDataNodePlan req) {
+ LOGGER.info("start to prepare for request: {}", req);
+ this.headRequest = req;
+ // to exec the request, will change it's state at different stage.
+ this.headNodeIndex = 0;
+ this.headNodeState = DataNodeRemoveState.NORMAL;
+
+ TDataNodeLocation headNode = req.getDataNodeLocations().get(headNodeIndex);
+ this.headNodeRegionIds =
+ configManager.getPartitionManager().getAllReplicaSets().stream()
+ .filter(rg -> rg.getDataNodeLocations().contains(headNode))
+ .filter(rg -> rg.regionId.getType() != TConsensusGroupType.PartitionRegion)
+ .map(TRegionReplicaSet::getRegionId)
+ .collect(Collectors.toList());
+ this.headRegionIndex = 0;
+ this.headRegionState = RegionMigrateState.ONLINE;
+
+ // modify head quest
+ this.headRequest.setUpdate(true);
+ this.headRequest.setExecDataNodeRegionIds(headNodeRegionIds);
+ this.headRequest.setExecDataNodeIndex(headNodeIndex);
+ this.headRequest.setExecRegionIndex(headRegionIndex);
+ this.headRequest.setExecDataNodeState(headNodeState);
+ this.headRequest.setExecRegionState(headRegionState);
+ configManager.getConsensusManager().write(headRequest);
+ LOGGER.info("finished to prepare for request: {}", req);
+ }
+
+ private void setUp() {
+ removeQueue.addAll(configManager.getNodeManager().getDataNodeRemoveRequestQueue());
+ headRequest = configManager.getNodeManager().getHeadRequestForDataNodeRemove();
+ if (headRequest == null) {
+ return;
+ }
+
+ // avoid duplication
+ if (removeQueue.contains(headRequest)) {
+ removeQueue.remove(headRequest);
+ }
+
+ headNodeIndex = headRequest.getExecDataNodeIndex();
+ headNodeState = headRequest.getExecDataNodeState();
+ headNodeRegionIds = headRequest.getExecDataNodeRegionIds();
+ headRegionIndex = headRequest.getExecRegionIndex();
+ headRegionState = headRequest.getExecRegionState();
+ }
+
+ /**
+ * exec the request loop all removed node 1: brocast it to cluster 2: loop region on it 2.1 change
+ * region leader 2.2 migrate region 3 stop the node or roll back
+ *
+ * @param req RemoveDataNodeReq
+ */
+ private TSStatus execRemoveDataNodeRequest(RemoveDataNodePlan req) {
+ TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+
+ for (TDataNodeLocation dataNodeLocation : req.getDataNodeLocations()) {
+ headNodeIndex = req.getDataNodeLocations().indexOf(dataNodeLocation);
+ status = broadcastDisableDataNode(req, dataNodeLocation);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.error("Disable Data Node Error {}", status);
+ return status;
+ }
+
+ // fetch data/schema region from one datanode
+ status = migrateSingleDataNodeRegions(req, dataNodeLocation);
+ if (isSucceed(status)) {
+ status = stopDataNode(req, dataNodeLocation);
+ if (isFailed(status)) {
+ LOGGER.error(
+ "send rpc to stop the Data Node {} error, please stop it use command",
+ dataNodeLocation);
+ }
+ } else {
+ LOGGER.error("the request run failed {}, result {}. will roll back", req, status);
+ rollBackSingleNode(req, dataNodeLocation);
+ }
+ }
+ return status;
+ }
+
+ private void updateRegionLocationCache(
+ TConsensusGroupId regionId, TDataNodeLocation oldNode, TDataNodeLocation newNode) {
+ LOGGER.debug(
+ "start to update region {} location from {} to {} when it migrate succeed",
+ regionId,
+ oldNode.getInternalEndPoint().getIp(),
+ newNode.getInternalEndPoint().getIp());
+ UpdateRegionLocationPlan req = new UpdateRegionLocationPlan(regionId, oldNode, newNode);
+ TSStatus status = configManager.getPartitionManager().updateRegionLocation(req);
+ LOGGER.debug(
+ "update region {} location finished, result:{}, old:{}, new:{}",
+ regionId,
+ status,
+ oldNode.getInternalEndPoint().getIp(),
+ newNode.getInternalEndPoint().getIp());
+ }
+
+ private void rollBackSingleNode(RemoveDataNodePlan req, TDataNodeLocation node) {
+ LOGGER.warn("roll back remove data node {} in the request {}", node, req);
+ if (headRegionState == RegionMigrateState.DATA_COPY_FAILED) {
+ // TODO delete target node the head region data
+ TConsensusGroupId tRegionId = lastRegionMigrateResult.getRegionId();
+ for (Map.Entry<TDataNodeLocation, TRegionMigrateFailedType> entry :
+ lastRegionMigrateResult.getFailedNodeAndReason().entrySet()) {
+ TDataNodeLocation failedNode = entry.getKey();
+ TRegionMigrateFailedType failedReason = entry.getValue();
+ switch (failedReason) {
+ // TODO how to impl roll back
+ case AddPeerFailed:
+ LOGGER.warn(
+ "add new peer node {} for region {} failed, will roll back",
+ failedNode.getInternalEndPoint().getIp(),
+ tRegionId);
+ break;
+ case RemovePeerFailed:
+ LOGGER.warn(
+ "remove old peer node {} for region {} failed, will roll back",
+ failedNode.getInternalEndPoint().getIp(),
+ tRegionId);
+ break;
+ case RemoveConsensusGroupFailed:
+ LOGGER.warn(
+ "remove consensus group on node {} for region {} failed, will roll back",
+ failedNode.getInternalEndPoint().getIp(),
+ tRegionId);
+ break;
+ case DeleteRegionFailed:
+ LOGGER.warn(
+ "create region {} instance on {} failed, will roll back",
+ failedNode.getInternalEndPoint().getIp(),
+ tRegionId);
+ break;
+ default:
+ LOGGER.warn(
+ "UnSupport reason {} for region {} migrate failed", failedReason, tRegionId);
+ }
+ }
+ }
+ // TODO if roll back failed, FAILED
+ storeDataNodeState(req, DataNodeRemoveState.REMOVE_FAILED);
+ }
+
+ /**
+ * broadcast these datanode in RemoveDataNodeReq are disabled, so they will not accept read/write
+ * request
+ *
+ * @param req RemoveDataNodeReq
+ * @param disabledDataNod TDataNodeLocation
+ */
+ private TSStatus broadcastDisableDataNode(
+ RemoveDataNodePlan req, TDataNodeLocation disabledDataNod) {
+ LOGGER.info(
+ "DataNodeRemoveService start send disable the Data Node to cluster, {}", disabledDataNod);
+ storeDataNodeState(req, DataNodeRemoveState.REMOVE_START);
+ TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ List<TEndPoint> otherOnlineDataNodes =
+ configManager.getLoadManager().getOnlineDataNodes(-1).stream()
+ .map(TDataNodeInfo::getLocation)
+ .filter(loc -> !loc.equals(disabledDataNod))
+ .map(TDataNodeLocation::getInternalEndPoint)
+ .collect(Collectors.toList());
+
+ for (TEndPoint server : otherOnlineDataNodes) {
+ status = SyncDataNodeClientPool.getInstance().disableDataNode(server, disabledDataNod);
+ // TODO add retry later
+ if (!isSucceed(status)) {
+ return status;
+ }
+ }
+ LOGGER.info(
+ "DataNodeRemoveService finished send disable the Data Node to cluster, {}",
+ disabledDataNod);
+ status.setMessage("Succeed disable the Data Node from cluster");
+ return status;
+ }
+
+ private void storeRegionState(RemoveDataNodePlan req, RegionMigrateState state) {
+ req.setExecRegionIndex(headRegionIndex);
+ headRegionState = state;
+ req.setExecRegionState(headRegionState);
+ configManager.getConsensusManager().write(req);
+ }
+
+ private void storeDataNodeState(RemoveDataNodePlan req, DataNodeRemoveState state) {
+ req.setExecDataNodeIndex(headNodeIndex);
+ headNodeState = state;
+ req.setExecDataNodeState(headNodeState);
+ configManager.getConsensusManager().write(req);
+ }
+
+ private TSStatus migrateSingleDataNodeRegions(
+ RemoveDataNodePlan req, TDataNodeLocation dataNodeLocation) {
+ LOGGER.info("start to migrate regions on the Data Node: {}", dataNodeLocation);
+ TSStatus status;
+ storeDataNodeState(req, DataNodeRemoveState.REGION_MIGRATING);
+ for (TConsensusGroupId regionId : headNodeRegionIds) {
+ headRegionIndex = headNodeRegionIds.indexOf(regionId);
+ // impl migrate region with twice rpc: CN-->DN(send), DN-->CN(report)
+ status = migrateSingleRegion(req, dataNodeLocation, regionId);
+ // if has one region migrate failed, the node remove failed
+ if (isFailed(status)) {
+ storeDataNodeState(req, DataNodeRemoveState.REGION_MIGRATE_FAILED);
+ return status;
+ }
+ }
+
+ // all regions on the node migrate succeed
+ storeDataNodeState(req, DataNodeRemoveState.REGION_MIGRATE_SUCCEED);
+ status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ status.setMessage("The Data Node migrate regions succeed");
+ LOGGER.info("finished to migrate regions on the Data Node: {}", dataNodeLocation);
+ return status;
+ }
+
+ private TSStatus migrateSingleRegion(
+ RemoveDataNodePlan req, TDataNodeLocation node, TConsensusGroupId regionId) {
+ // change region leader: change leader and then add peer, will cause ratis server exit
+ // so modify policy: add peer firs, then change region leader, see:
+ // RegionMigrateService.changeLeader()
+
+ // do migrate region
+ TSStatus status = doMigrateSingleRegion(req, node, regionId);
+ if (isFailed(status)) {
+ storeRegionState(req, RegionMigrateState.DATA_COPY_FAILED);
+ return status;
+ }
+ storeRegionState(req, RegionMigrateState.DATA_COPY_SUCCEED);
+ return status;
+ }
+
+ private TSStatus changeSingleRegionLeader(
+ RemoveDataNodePlan req, TDataNodeLocation node, TConsensusGroupId regionId) {
+ storeRegionState(req, RegionMigrateState.LEADER_CHANGING);
+ LOGGER.debug("start to send region leader change. {}", regionId);
+ TSStatus status;
+ // pick a node in same raft group to be new region leader
+ List<TRegionReplicaSet> regionReplicaSets =
+ configManager.getPartitionManager().getAllReplicaSets().stream()
+ .filter(rg -> rg.getRegionId().equals(regionId))
+ .collect(Collectors.toList());
+ if (regionReplicaSets.isEmpty()) {
+ LOGGER.warn("not find TRegionReplica for region: {}, ignore it", regionId);
+ status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
+ status.setMessage("not find TRegionReplica for region, ignore");
+ return status;
+ }
+ Optional<TDataNodeLocation> newLeaderNode =
+ regionReplicaSets.get(0).getDataNodeLocations().stream()
+ .filter(e -> !e.equals(node))
+ .findAny();
+ if (!newLeaderNode.isPresent()) {
+ LOGGER.warn("No enough Data node to change leader for region: {}", regionId);
+ status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
+ status.setMessage("No enough Data node to change leader for region " + regionId);
+ return status;
+ }
+ status =
+ SyncDataNodeClientPool.getInstance()
+ .changeRegionLeader(regionId, node.getInternalEndPoint(), newLeaderNode.get());
+ LOGGER.debug("finished to send region leader change. {}", regionId);
+ return status;
+ }
+
+ private TSStatus doMigrateSingleRegion(
+ RemoveDataNodePlan req, TDataNodeLocation node, TConsensusGroupId regionId) {
+ storeRegionState(req, RegionMigrateState.DATA_COPYING);
+ LOGGER.debug("start to migrate region {}", regionId);
+ TSStatus status;
+ List<TDataNodeLocation> regionReplicaNodes = findRegionReplicaNodes(regionId);
+ if (regionReplicaNodes.isEmpty()) {
+ LOGGER.warn("Not find region replica nodes, region: {}", regionId);
+ status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
+ status.setMessage("not find region replica nodes, region: " + regionId);
+ return status;
+ }
+
+ // will migrate the region to the new node, which should not be same raft
+ Optional<TDataNodeLocation> newNode = pickNewReplicaNodeForRegion(regionReplicaNodes);
+ if (!newNode.isPresent()) {
+ LOGGER.warn("No enough Data node to migrate region: {}", regionId);
+ status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
+ status.setMessage("No enough Data node to migrate region, region: " + regionId);
+ return status;
+ }
+
+ status = addNewNodeToRegionConsensusGroup(regionId, regionReplicaNodes, newNode.get());
+ if (isFailed(status)) {
+ return status;
+ }
+
+ Optional<TDataNodeLocation> newLeaderNode =
Review Comment:
Is it possible that the region contains 1 copy and user want to migrate it to another datanode? In this case newLeaderNode is null. we can add a todo here.
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java:
##########
@@ -164,12 +168,14 @@ public ConfigManager() throws IOException {
this.udfManager = new UDFManager(this, udfInfo);
this.loadManager = new LoadManager(this);
this.consensusManager = new ConsensusManager(this, stateMachine);
+ this.dataNodeRemoveManager = new DataNodeRemoveManager(this);
Review Comment:
Does dataNodeRemoveManager has any data persisted in disk? If it has, it should be initialized after ConsensusManager.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org