[IOTDB-2780] Config node ratis consensus protocol implementation (#5347)

Author: CRZbulabula

confignode/pom.xml

@@ -55,6 +55,11 @@
             <artifactId>iotdb-consensus</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.dropwizard.metrics</groupId>
+            <artifactId>metrics-core</artifactId>
+            <version>3.2.5</version>
+        </dependency>
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

confignode/src/assembly/resources/conf/iotdb-confignode.properties

@@ -33,11 +33,7 @@ config_node_rpc_port=22277
 # Datatype: int
 # config_node_internal_port=22278
 
-# used for building the ConfigNode consensus group
-# all config node address and internal port
-# every node should have the same config_node_address_lists
-# Datatype: String
-# config_node_address_lists=host0:22278,host1:22278,host2:22278
+# this feature is under development, set this as false before it is done.
 # Datatype: boolean
 # rpc_thrift_compression_enable=false
 
@@ -57,6 +53,23 @@ config_node_rpc_port=22277
 # Datatype: int
 # thrift_init_buffer_size=1024
 
+####################
+### consensus protocol configuration
+####################
+
+# ConfigNodeGroup consensus protocol type
+# These consensus protocol are currently supported:
+# 1. standalone(No protocol, only supports stand-alone machine)
+# 2. ratis(Raft protocol)
+# Datatype: String
+# consensus_type=standalone
+
+# Used for building the ConfigNode consensus group
+# all config node address and internal port, use comma to distinguish
+# every node should have the same config_node_address_lists
+# Datatype: String
+# config_node_group_address_list=0.0.0.0:22278,0.0.0.0:22280,0.0.0.0:22282
+
 ####################
 ### DeviceGroup Configuration
 ####################
@@ -81,9 +94,9 @@ config_node_rpc_port=22277
 ####################
 
 # system dir
-# If this property is unset, system will save the data in the default relative path directory under the IoTDB folder(i.e., %IOTDB_HOME%/data/system).
+# If this property is unset, system will save the data in the default relative path directory under the confignode folder(i.e., %CONFIGNODE_HOME%/data/system).
 # If it is absolute, system will save the data in exact location it points to.
-# If it is relative, system will save the data in the relative path directory it indicates under the IoTDB folder.
+# If it is relative, system will save the data in the relative path directory it indicates under the confignode folder.
 # For windows platform
 # If its prefix is a drive specifier followed by "\\", or if its prefix is "\\\\", then the path is absolute. Otherwise, it is relative.
 # system_dir=data\\system
@@ -93,9 +106,9 @@ config_node_rpc_port=22277
 
 
 # data dirs
-# If this property is unset, system will save the data in the default relative path directory under the IoTDB folder(i.e., %IOTDB_HOME%/data/data).
+# If this property is unset, system will save the data in the default relative path directory under the confignode folder(i.e., %CONFIGNODE_HOME%/data/data).
 # If it is absolute, system will save the data in exact location it points to.
-# If it is relative, system will save the data in the relative path directory it indicates under the IoTDB folder.
+# If it is relative, system will save the data in the relative path directory it indicates under the confignode folder.
 # Note: If data_dir is assigned an empty string(i.e.,zero-size), it will be handled as a relative path.
 # For windows platform
 # If its prefix is a drive specifier followed by "\\", or if its prefix is "\\\\", then the path is absolute. Otherwise, it is relative.
@@ -104,6 +117,19 @@ config_node_rpc_port=22277
 # If its prefix is "/", then the path is absolute. Otherwise, it is relative.
 # data_dirs=data/data
 
+
+# consensus dir
+# If this property is unset, system will save the data in the default relative path directory under the confignode folder(i.e., %CONFIGNODE_HOME%/data/consensus).
+# If it is absolute, system will save the data in exact location it points to.
+# If it is relative, system will save the data in the relative path directory it indicates under the confignode folder.
+# Note: If data_dir is assigned an empty string(i.e.,zero-size), it will be handled as a relative path.
+# For windows platform
+# If its prefix is a drive specifier followed by "\\", or if its prefix is "\\\\", then the path is absolute. Otherwise, it is relative.
+# consensus_dir=data\\consensus
+# For Linux platform
+# If its prefix is "/", then the path is absolute. Otherwise, it is relative.
+# consensus_dir=data/consensus
+
 ####################
 ### Region Configuration
 ####################

confignode/src/assembly/resources/conf/logback.xml

@@ -135,6 +135,6 @@
         <appender-ref ref="FILEALL"/>
         <appender-ref ref="stdout"/>
     </root>
-    <logger level="info" name="org.apache.iotdb.confignode.service"/>
-    <logger level="info" name="org.apache.iotdb.confignode.conf"/>
+    <logger level="info" name="org.apache.iotdb.confignode"/>
+    <!-- <logger level="info" name="org.apache.ratis"/> -->
 </configuration>

confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConf.java

@@ -19,6 +19,8 @@
 package org.apache.iotdb.confignode.conf;
 
 import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.confignode.consensus.ConsensusType;
+import org.apache.iotdb.consensus.common.Endpoint;
 import org.apache.iotdb.rpc.RpcUtils;
 
 import java.io.File;
@@ -34,8 +36,11 @@ public class ConfigNodeConf {
   /** used for communication between data node and data node */
   private int internalPort = 22278;
 
-  /** every node should have the same config_node_address_lists */
-  private String addressLists;
+  /** ConfigNodeGroup consensus protocol */
+  private ConsensusType consensusType = ConsensusType.STANDALONE;
+
+  /** Used for building the ConfigNode consensus group */
+  private Endpoint[] configNodeGroupAddressList = null;
 
   /** Number of DeviceGroups per StorageGroup */
   private int deviceGroupCount = 10000;
@@ -70,6 +75,10 @@ public class ConfigNodeConf {
     ConfigNodeConstant.DATA_DIR + File.separator + ConfigNodeConstant.DATA_DIR
   };
 
+  /** Consensus directory, storage consensus protocol logs */
+  private String consensusDir =
+      ConfigNodeConstant.DATA_DIR + File.separator + ConfigNodeConstant.CONSENSUS_FOLDER;
+
   private int regionReplicaCount = 3;
   private int schemaRegionCount = 1;
   private int dataRegionCount = 1;
@@ -87,6 +96,7 @@ private void formulateFolders() {
     for (int i = 0; i < dataDirs.length; i++) {
       dataDirs[i] = addHomeDir(dataDirs[i]);
     }
+    consensusDir = addHomeDir(consensusDir);
   }
 
   private String addHomeDir(String dir) {
@@ -181,12 +191,28 @@ public void setInternalPort(int internalPort) {
     this.internalPort = internalPort;
   }
 
-  public String getAddressLists() {
-    return addressLists;
+  public String getConsensusDir() {
+    return consensusDir;
+  }
+
+  public void setConsensusDir(String consensusDir) {
+    this.consensusDir = consensusDir;
+  }
+
+  public ConsensusType getConsensusType() {
+    return consensusType;
+  }
+
+  public void setConsensusType(ConsensusType consensusType) {
+    this.consensusType = consensusType;
+  }
+
+  public Endpoint[] getConfigNodeGroupAddressList() {
+    return configNodeGroupAddressList;
   }
 
-  public void setAddressLists(String addressLists) {
-    this.addressLists = addressLists;
+  public void setConfigNodeGroupAddressList(Endpoint[] configNodeGroupAddressList) {
+    this.configNodeGroupAddressList = configNodeGroupAddressList;
   }
 
   public int getThriftServerAwaitTimeForStopService() {

confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConstant.java

@@ -41,6 +41,7 @@ public class ConfigNodeConstant {
 
   public static final String DATA_DIR = "data";
   public static final String CONF_DIR = "conf";
+  public static final String CONSENSUS_FOLDER = "consensus";
 
   public static final int MIN_SUPPORTED_JDK_VERSION = 8;
 

confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java

@@ -18,6 +18,9 @@
  */
 package org.apache.iotdb.confignode.conf;
 
+import org.apache.iotdb.confignode.consensus.ConsensusType;
+import org.apache.iotdb.consensus.common.Endpoint;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,7 +52,7 @@ public ConfigNodeConf getConf() {
   public URL getPropsUrl() {
     // Check if a config-directory was specified first.
     String urlString = System.getProperty(ConfigNodeConstant.CONFIGNODE_CONF, null);
-    // If it wasn't, check if a home directory was provided (This usually contains a config)
+    // If it wasn't, check if a home directory was provided
     if (urlString == null) {
       urlString = System.getProperty(ConfigNodeConstant.CONFIGNODE_HOME, null);
       if (urlString != null) {
@@ -60,15 +63,9 @@ public URL getPropsUrl() {
                 + File.separatorChar
                 + ConfigNodeConstant.CONF_NAME;
       } else {
-        // If this too wasn't provided, try to find a default config in the root of the classpath.
-        URL uri = ConfigNodeConf.class.getResource("/" + ConfigNodeConstant.CONF_NAME);
-        if (uri != null) {
-          return uri;
-        }
-        LOGGER.warn(
-            "Cannot find IOTDB_HOME or IOTDB_CONF environment variable when loading "
-                + "config file {}, use default configuration",
-            ConfigNodeConstant.CONF_NAME);
+        // When start ConfigNode with the script, the environment variables CONFIGNODE_CONF
+        // and CONFIGNODE_HOME will be set. But we didn't set these two in developer mode.
+        // Thus, just return null and use default Configuration in developer mode.
         return null;
       }
     }
@@ -125,8 +122,9 @@ private void loadProps() {
               properties.getProperty(
                   "config_node_internal_port", String.valueOf(conf.getInternalPort()))));
 
-      conf.setAddressLists(
-          properties.getProperty("config_node_address_lists", conf.getAddressLists()));
+      conf.setConsensusType(
+          ConsensusType.getConsensusType(
+              properties.getProperty("consensus_type", String.valueOf(conf.getConsensusType()))));
 
       conf.setRpcAdvancedCompressionEnable(
           Boolean.parseBoolean(
@@ -160,6 +158,8 @@ private void loadProps() {
 
       conf.setDataDirs(properties.getProperty("data_dirs", conf.getDataDirs()[0]).split(","));
 
+      conf.setConsensusDir(properties.getProperty("consensus_dir", conf.getConsensusDir()));
+
       conf.setRegionReplicaCount(
           Integer.parseInt(
               properties.getProperty(
@@ -175,6 +175,24 @@ private void loadProps() {
               properties.getProperty(
                   "data_region_count", String.valueOf(conf.getDataRegionCount()))));
 
+      String addresses = properties.getProperty("config_node_group_address_list", null);
+      if (addresses != null) {
+        String[] addressList = addresses.split(",");
+        Endpoint[] endpointList = new Endpoint[addressList.length];
+        for (int i = 0; i < addressList.length; i++) {
+          String[] ipPort = addressList[i].split(":");
+          if (ipPort.length != 2) {
+            throw new IOException(
+                String.format(
+                    "Parsing parameter config_node_group_address_list error. "
+                        + "The %d-th address must format to ip:port, but currently is %s",
+                    i, addressList[i]));
+          }
+          endpointList[i] = new Endpoint(ipPort[0], Integer.parseInt(ipPort[1]));
+        }
+        conf.setConfigNodeGroupAddressList(endpointList);
+      }
+
     } catch (IOException e) {
       LOGGER.warn("Couldn't load ConfigNode conf file, use default config", e);
     } finally {

confignode/src/main/java/org/apache/iotdb/confignode/consensus/ConsensusType.java

@@ -0,0 +1,54 @@
+/*
+ * 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.consensus;
+
+import java.util.Arrays;
+
+public enum ConsensusType {
+  STANDALONE("standalone"),
+  RATIS("ratis");
+
+  private final String typeName;
+
+  ConsensusType(String typeName) {
+    this.typeName = typeName;
+  }
+
+  public String getTypeName() {
+    return typeName;
+  }
+
+  @Override
+  public String toString() {
+    return typeName;
+  }
+
+  public static ConsensusType getConsensusType(String typeName) {
+    for (ConsensusType type : ConsensusType.values()) {
+      if (type.getTypeName().equals(typeName)) {
+        return type;
+      }
+    }
+
+    throw new IllegalArgumentException(
+        String.format(
+            "Unknown consensus type, found: %s expected: %s",
+            typeName, Arrays.toString(ConsensusType.values())));
+  }
+}

confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/PartitionRegionStateMachine.java

@@ -38,7 +38,11 @@ public class PartitionRegionStateMachine implements IStateMachine {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PartitionRegionStateMachine.class);
 
-  private final PlanExecutor executor = new PlanExecutor();
+  private final PlanExecutor executor;
+
+  public PartitionRegionStateMachine() {
+    this.executor = new PlanExecutor();
+  }
 
   @Override
   public TSStatus write(IConsensusRequest request) {
@@ -74,7 +78,14 @@ protected TSStatus write(PhysicalPlan plan) {
   @Override
   public DataSet read(IConsensusRequest request) {
     PhysicalPlan plan;
-    if (request instanceof PhysicalPlan) {
+    if (request instanceof ByteBufferConsensusRequest) {
+      try {
+        plan = PhysicalPlan.Factory.create(((ByteBufferConsensusRequest) request).getContent());
+      } catch (IOException e) {
+        LOGGER.error("Deserialization error for write plan : {}", request);
+        return null;
+      }
+    } else if (request instanceof PhysicalPlan) {
       plan = (PhysicalPlan) request;
     } else {
       LOGGER.error("Unexpected read plan : {}", request);