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/05/11 04:29:49 UTC
[GitHub] [iotdb] CRZbulabula opened a new pull request, #5862: [IoTDB-3053] Copyset algorithm when create timeseries
CRZbulabula opened a new pull request, #5862:
URL: https://github.com/apache/iotdb/pull/5862
Changes:
- Use Copyset algorithm for **the first Region allocation** of each StorageGroup. You can see this doc: [RegionGroup allocation](https://apache-iotdb.feishu.cn/docs/doccnHuDkDD6RiwLjaej3hkA9Vb) for more details.
- Modify the behaviors of some RPC interfaces, including **setStorageGroup**, **getOrCreateSchemaPartition** and **getOrCreateDataPartition** to adapt to Copyset algorithm. You can see this doc: [ConfigNode for DataNode RPC interfaces](https://apache-iotdb.feishu.cn/docs/doccnCp13O7QDU7O6GMvagUwEDc#4ZxOG4)
- Build basic **LoadBalance framework** for future development.
Incidentally framework adjust:
- Delete the **data_dirs** parameter.
- More secure restart parameter checking process. See ConfigNodeStartupCheck.java
- Allows to change the folder locations on restart. See ConfigNodeStartupCheck.java
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #5862: [IoTDB-3053] Copyset algorithm when create timeseries
Posted by GitBox <gi...@apache.org>.
CRZbulabula commented on code in PR #5862:
URL: https://github.com/apache/iotdb/pull/5862#discussion_r870055422
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/InitRegionHandler.java:
##########
@@ -36,11 +39,20 @@ public class InitRegionHandler implements AsyncMethodCallback<TSStatus> {
private final int index;
private final BitSet bitSet;
private final CountDownLatch latch;
+ private final TConsensusGroupId consensusGroupId;
+ private final TDataNodeLocation dataNodeLocation;
- public InitRegionHandler(int index, BitSet bitSet, CountDownLatch latch) {
+ public InitRegionHandler(
Review Comment:
Fixed.
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/LoadManager.java:
##########
@@ -18,17 +18,215 @@
*/
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.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.handlers.InitRegionHandler;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
+import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.manager.allocator.CopySetRegionAllocator;
+import org.apache.iotdb.confignode.manager.allocator.IRegionAllocator;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.BitSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
/**
- * The LoadBalancer at ConfigNodeGroup-Leader is active for cluster dynamic load balancing
- * scheduling
+ * The LoadManager at ConfigNodeGroup-Leader is active. It proactively implements the cluster
+ * dynamic load balancing policy and passively accepts the PartitionTable expansion request.
*/
public class LoadManager {
- private void metadataLoadBalance() {}
+ private static final Logger LOGGER = LoggerFactory.getLogger(LoadManager.class);
+
+ private static final int schemaReplicationFactor =
+ ConfigNodeDescriptor.getInstance().getConf().getSchemaReplicationFactor();
Review Comment:
Fixed.
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/LoadManager.java:
##########
@@ -18,17 +18,215 @@
*/
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.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.handlers.InitRegionHandler;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
+import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.manager.allocator.CopySetRegionAllocator;
+import org.apache.iotdb.confignode.manager.allocator.IRegionAllocator;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.BitSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
/**
- * The LoadBalancer at ConfigNodeGroup-Leader is active for cluster dynamic load balancing
- * scheduling
+ * The LoadManager at ConfigNodeGroup-Leader is active. It proactively implements the cluster
+ * dynamic load balancing policy and passively accepts the PartitionTable expansion request.
*/
public class LoadManager {
- private void metadataLoadBalance() {}
+ private static final Logger LOGGER = LoggerFactory.getLogger(LoadManager.class);
+
+ private static final int schemaReplicationFactor =
+ ConfigNodeDescriptor.getInstance().getConf().getSchemaReplicationFactor();
+ private static final int dataReplicationFactor =
+ ConfigNodeDescriptor.getInstance().getConf().getDataReplicationFactor();
- private void dataLoadBalance() {}
+ private final Manager configManager;
- private void cleanDataSlice() {}
+ private final IRegionAllocator regionAllocator;
// TODO: Interfaces for active, interrupt and reset LoadBalancer
+
+ public LoadManager(Manager configManager) {
+ this.configManager = configManager;
+ this.regionAllocator = new CopySetRegionAllocator();
+ }
+
+ /**
+ * Allocate and create one Region on DataNode for each StorageGroup.
+ *
+ * @param storageGroups List<StorageGroupName>
+ * @param consensusGroupType TConsensusGroupType of Region to be allocated
+ */
+ public void allocateAndCreateRegions(
+ List<String> storageGroups, TConsensusGroupType consensusGroupType)
+ throws NotEnoughDataNodeException {
+ CreateRegionsReq createRegionsReq = allocateRegions(storageGroups, consensusGroupType);
+
+ // TODO: use procedure to protect create Regions process
+ try {
+ createRegionsOnDataNodes(createRegionsReq);
+ } catch (Exception e) {
+ LOGGER.error("Meet error when create Regions", e);
+ }
+
+ getConsensusManager().write(createRegionsReq);
+ }
+
+ private CreateRegionsReq allocateRegions(
+ List<String> storageGroups, TConsensusGroupType consensusGroupType)
+ throws NotEnoughDataNodeException {
+ CreateRegionsReq createRegionsReq = new CreateRegionsReq();
+
+ int replicationFactor =
+ consensusGroupType.equals(TConsensusGroupType.SchemaRegion)
+ ? schemaReplicationFactor
+ : dataReplicationFactor;
+ List<TDataNodeLocation> onlineDataNodes = getNodeManager().getOnlineDataNodes();
+ List<TRegionReplicaSet> allocatedRegions = getPartitionManager().getAllocatedRegions();
+
+ if (onlineDataNodes.size() < replicationFactor) {
+ throw new NotEnoughDataNodeException();
+ }
+
+ for (String storageGroup : storageGroups) {
+ TRegionReplicaSet newRegion =
+ regionAllocator.allocateRegion(
+ onlineDataNodes,
+ allocatedRegions,
+ replicationFactor,
+ new TConsensusGroupId(
+ consensusGroupType, getPartitionManager().generateNextRegionGroupId()));
+ createRegionsReq.addRegion(storageGroup, newRegion);
+ }
+
+ return createRegionsReq;
+ }
+
+ private void createRegionsOnDataNodes(CreateRegionsReq createRegionsReq)
+ throws MetadataException {
+ int index = 0;
+ int regionNum = 0;
Review Comment:
Fixed.
--
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
[GitHub] [iotdb] qiaojialin merged pull request #5862: [IoTDB-3053] Copyset algorithm when create timeseries
Posted by GitBox <gi...@apache.org>.
qiaojialin merged PR #5862:
URL: https://github.com/apache/iotdb/pull/5862
--
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
[GitHub] [iotdb] qiaojialin commented on a diff in pull request #5862: [IoTDB-3053] Copyset algorithm when create timeseries
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on code in PR #5862:
URL: https://github.com/apache/iotdb/pull/5862#discussion_r869944243
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/LoadManager.java:
##########
@@ -18,17 +18,215 @@
*/
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.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.handlers.InitRegionHandler;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
+import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.manager.allocator.CopySetRegionAllocator;
+import org.apache.iotdb.confignode.manager.allocator.IRegionAllocator;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.BitSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
/**
- * The LoadBalancer at ConfigNodeGroup-Leader is active for cluster dynamic load balancing
- * scheduling
+ * The LoadManager at ConfigNodeGroup-Leader is active. It proactively implements the cluster
+ * dynamic load balancing policy and passively accepts the PartitionTable expansion request.
*/
public class LoadManager {
- private void metadataLoadBalance() {}
+ private static final Logger LOGGER = LoggerFactory.getLogger(LoadManager.class);
+
+ private static final int schemaReplicationFactor =
+ ConfigNodeDescriptor.getInstance().getConf().getSchemaReplicationFactor();
Review Comment:
This may be different in each storage group.
##########
confignode/src/main/java/org/apache/iotdb/confignode/client/handlers/InitRegionHandler.java:
##########
@@ -36,11 +39,20 @@ public class InitRegionHandler implements AsyncMethodCallback<TSStatus> {
private final int index;
private final BitSet bitSet;
private final CountDownLatch latch;
+ private final TConsensusGroupId consensusGroupId;
+ private final TDataNodeLocation dataNodeLocation;
- public InitRegionHandler(int index, BitSet bitSet, CountDownLatch latch) {
+ public InitRegionHandler(
Review Comment:
add javadoc for each parameter
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/LoadManager.java:
##########
@@ -18,17 +18,215 @@
*/
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.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.confignode.client.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.handlers.InitRegionHandler;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.write.CreateRegionsReq;
+import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.manager.allocator.CopySetRegionAllocator;
+import org.apache.iotdb.confignode.manager.allocator.IRegionAllocator;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateSchemaRegionReq;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.BitSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
/**
- * The LoadBalancer at ConfigNodeGroup-Leader is active for cluster dynamic load balancing
- * scheduling
+ * The LoadManager at ConfigNodeGroup-Leader is active. It proactively implements the cluster
+ * dynamic load balancing policy and passively accepts the PartitionTable expansion request.
*/
public class LoadManager {
- private void metadataLoadBalance() {}
+ private static final Logger LOGGER = LoggerFactory.getLogger(LoadManager.class);
+
+ private static final int schemaReplicationFactor =
+ ConfigNodeDescriptor.getInstance().getConf().getSchemaReplicationFactor();
+ private static final int dataReplicationFactor =
+ ConfigNodeDescriptor.getInstance().getConf().getDataReplicationFactor();
- private void dataLoadBalance() {}
+ private final Manager configManager;
- private void cleanDataSlice() {}
+ private final IRegionAllocator regionAllocator;
// TODO: Interfaces for active, interrupt and reset LoadBalancer
+
+ public LoadManager(Manager configManager) {
+ this.configManager = configManager;
+ this.regionAllocator = new CopySetRegionAllocator();
+ }
+
+ /**
+ * Allocate and create one Region on DataNode for each StorageGroup.
+ *
+ * @param storageGroups List<StorageGroupName>
+ * @param consensusGroupType TConsensusGroupType of Region to be allocated
+ */
+ public void allocateAndCreateRegions(
+ List<String> storageGroups, TConsensusGroupType consensusGroupType)
+ throws NotEnoughDataNodeException {
+ CreateRegionsReq createRegionsReq = allocateRegions(storageGroups, consensusGroupType);
+
+ // TODO: use procedure to protect create Regions process
+ try {
+ createRegionsOnDataNodes(createRegionsReq);
+ } catch (Exception e) {
+ LOGGER.error("Meet error when create Regions", e);
+ }
+
+ getConsensusManager().write(createRegionsReq);
+ }
+
+ private CreateRegionsReq allocateRegions(
+ List<String> storageGroups, TConsensusGroupType consensusGroupType)
+ throws NotEnoughDataNodeException {
+ CreateRegionsReq createRegionsReq = new CreateRegionsReq();
+
+ int replicationFactor =
+ consensusGroupType.equals(TConsensusGroupType.SchemaRegion)
+ ? schemaReplicationFactor
+ : dataReplicationFactor;
+ List<TDataNodeLocation> onlineDataNodes = getNodeManager().getOnlineDataNodes();
+ List<TRegionReplicaSet> allocatedRegions = getPartitionManager().getAllocatedRegions();
+
+ if (onlineDataNodes.size() < replicationFactor) {
+ throw new NotEnoughDataNodeException();
+ }
+
+ for (String storageGroup : storageGroups) {
+ TRegionReplicaSet newRegion =
+ regionAllocator.allocateRegion(
+ onlineDataNodes,
+ allocatedRegions,
+ replicationFactor,
+ new TConsensusGroupId(
+ consensusGroupType, getPartitionManager().generateNextRegionGroupId()));
+ createRegionsReq.addRegion(storageGroup, newRegion);
+ }
+
+ return createRegionsReq;
+ }
+
+ private void createRegionsOnDataNodes(CreateRegionsReq createRegionsReq)
+ throws MetadataException {
+ int index = 0;
+ int regionNum = 0;
Review Comment:
add javadoc
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/allocator/IRegionAllocator.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.allocator;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+
+import java.util.List;
+
+public interface IRegionAllocator {
+
+ /**
+ * Calculating the next optimal TRegionReplicaSet based on the current online DataNodes and
+ * allocated Regions
+ *
+ * @param onlineDataNodes DataNodes that currently communicable
+ * @param allocatedRegions Allocated Regions
+ * @param replicationFactor Replication factor of TRegionReplicaSet
+ * @param consensusGroupId TConsensusGroupId of result TRegionReplicaSet
+ * @return The optimal TRegionReplicaSet derived by the specific algorithm
+ */
+ TRegionReplicaSet allocateRegion(
+ List<TDataNodeLocation> onlineDataNodes,
+ List<TRegionReplicaSet> allocatedRegions,
Review Comment:
why not pass the PartitionInfo here
--
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
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #5862: [IoTDB-3053] Copyset algorithm when create timeseries
Posted by GitBox <gi...@apache.org>.
CRZbulabula commented on code in PR #5862:
URL: https://github.com/apache/iotdb/pull/5862#discussion_r869999095
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/allocator/IRegionAllocator.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.allocator;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+
+import java.util.List;
+
+public interface IRegionAllocator {
+
+ /**
+ * Calculating the next optimal TRegionReplicaSet based on the current online DataNodes and
+ * allocated Regions
+ *
+ * @param onlineDataNodes DataNodes that currently communicable
+ * @param allocatedRegions Allocated Regions
+ * @param replicationFactor Replication factor of TRegionReplicaSet
+ * @param consensusGroupId TConsensusGroupId of result TRegionReplicaSet
+ * @return The optimal TRegionReplicaSet derived by the specific algorithm
+ */
+ TRegionReplicaSet allocateRegion(
+ List<TDataNodeLocation> onlineDataNodes,
+ List<TRegionReplicaSet> allocatedRegions,
Review Comment:
1. I want the IRegionAllocator just be a functional interface. And every class that implement it just be a functional class. Which means their duty is simply to take input parameters and return the best results.
2. I want to cancel the singleton mode of the PartitionInfo and other Info classes in the future.
--
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