Support multi-raft (#150)

* feat(support Multi-DLedger): support Multi-DLedger

1.add DLedgerProxy to be the container of DLedgerServer.
2.add
ConfigManager to manage the config about DLedgerProxy and
DLedgerServer.
3.add DLedgerManager to manage the
DLedgerServers.
4.support configuration file startup.
5.start parameters
are compatible with the old version.
6.refactor the tests.

* test(fix some tests for new model): fix some tests for new model

1.fix some tests for new model

* style(multi-raft): code format

1.code format

* refactor(multi-raft): refactor the code and test

1.refactor the code and test

* test(multi-raft): fix some test and add a new test

1.fix some test and add a new test

* style(multi-raft): delete the print

1.delete the print

* fix(multi-raft): fix some bugs and add some tests about leaderElector

1.fix some bugs and add some tests about leaderElector

* test(multi-raft): add more tests about leaderElector

1.add more tests about leaderElector

* feat(multi-raft): update conflict code

1.update conflict code

* refactor(multi-raft): Refactor the startup args

1.add startup args "server -c" to start with config file.
2.remove DLedgerRpcNettyServer#memstate.
3.check config before initialize DLedgerRpcNettyService.

* refactor(multi-raft): refactor some code

1.remove DLedgerRpcNettyService#checkOnePort
2.remove printStack

* fix(multi-raft): fix some code

1.use "groupId#selfId" to identify different DLedgerServer.

* style(multi-raft): remove unused imports

1.remove unused imports

* feat(support Multi-DLedger): make ConfigManager more flexible

1. make ConfigManager more flexible
2. refactor structure about
multi-raft
3. more friendly for updating version

* fix(support Multi-DLedger): fix incompatible code after merge master branch

1. fix incompatible code after merge master branch

* refactor(main): refactor the main method in DLedger

1. refactor the main method in DLedger

Co-authored-by: TheR1sing3un <[email protected]>

Author: TheR1sing3un

pom.xml

@@ -98,6 +98,11 @@
             <version>${mockito.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.yaml</groupId>
+            <artifactId>snakeyaml</artifactId>
+            <version>1.30</version>
+        </dependency>
 
     </dependencies>
 

src/main/java/io/openmessaging/storage/dledger/DLedger.java

@@ -18,30 +18,47 @@
 
 import com.alibaba.fastjson.JSON;
 import com.beust.jcommander.JCommander;
+import io.openmessaging.storage.dledger.cmdline.ConfigCommand;
+import io.openmessaging.storage.dledger.dledger.DLedgerProxy;
+import io.openmessaging.storage.dledger.dledger.DLedgerProxyConfig;
+import io.openmessaging.storage.dledger.utils.ConfigUtils;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class DLedger {
 
     private static Logger logger = LoggerFactory.getLogger(DLedger.class);
-    
-    @Deprecated
-    public static void main(String args[]) {
-        DLedgerConfig dLedgerConfig = new DLedgerConfig();
-        JCommander.newBuilder().addObject(dLedgerConfig).build().parse(args);
-        bootstrapDLedger(dLedgerConfig);
-    }
-
-    public static void bootstrapDLedger(DLedgerConfig dLedgerConfig) {
 
-        if (null == dLedgerConfig) {
-            logger.error("Bootstrap DLedger server error", new IllegalArgumentException("DLedgerConfig is null"));
-            System.exit(-1);
+    public static void main(String[] args) {
+        List<DLedgerConfig> dLedgerConfigs = new LinkedList<>();
+        if ("--config".equals(args[0]) || "-c".equals(args[0])) {
+            ConfigCommand configCommand = new ConfigCommand();
+            JCommander.newBuilder().addObject(configCommand).build().parse(args);
+            try {
+                DLedgerProxyConfig dLedgerProxyConfig = ConfigUtils.parseDLedgerProxyConfig(configCommand.getConfigPath());
+                dLedgerConfigs.addAll(dLedgerProxyConfig.getConfigs());
+            } catch (Exception e) {
+                logger.error("Create DLedgerProxyConfig error", e);
+                System.exit(-1);
+            }
+        } else {
+            DLedgerConfig dLedgerConfig = new DLedgerConfig();
+            JCommander.newBuilder().addObject(dLedgerConfig).build().parse(args);
+            dLedgerConfigs.add(dLedgerConfig);
         }
+        bootstrapDLedger(dLedgerConfigs);
+    }
 
-        DLedgerServer dLedgerServer = new DLedgerServer(dLedgerConfig);
-        dLedgerServer.startup();
-        logger.info("[{}] group {} start ok with config {}", dLedgerConfig.getSelfId(), dLedgerConfig.getGroup(), JSON.toJSONString(dLedgerConfig));
+    public static void bootstrapDLedger(List<DLedgerConfig> dLedgerConfigs) {
+        if (dLedgerConfigs == null || dLedgerConfigs.isEmpty()) {
+            logger.error("Bootstrap DLedger server error", new IllegalArgumentException("DLedgerConfigs is null or empty"));
+        }
+        DLedgerProxy dLedgerProxy = new DLedgerProxy(dLedgerConfigs);
+        dLedgerProxy.startup();
+        logger.info("DLedgers start ok with config {}", JSON.toJSONString(dLedgerConfigs));
         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
             private volatile boolean hasShutdown = false;
 
@@ -52,12 +69,16 @@ public void run() {
                     if (!this.hasShutdown) {
                         this.hasShutdown = true;
                         long beginTime = System.currentTimeMillis();
-                        dLedgerServer.shutdown();
+                        dLedgerProxy.shutdown();
                         long consumingTimeTotal = System.currentTimeMillis() - beginTime;
                         logger.info("Shutdown hook over, consuming total time(ms): {}", consumingTimeTotal);
                     }
                 }
             }
         }, "ShutdownHook"));
     }
+
+    public static void bootstrapDLedger(DLedgerConfig dLedgerConfig) {
+        bootstrapDLedger(Collections.singletonList(dLedgerConfig));
+    }
 }

src/main/java/io/openmessaging/storage/dledger/DLedgerConfig.java

@@ -18,14 +18,22 @@
 
 import com.beust.jcommander.Parameter;
 import io.openmessaging.storage.dledger.store.file.DLedgerMmapFileStore;
+
+import io.openmessaging.storage.dledger.utils.DLedgerUtils;
 import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class DLedgerConfig {
 
     public static final String MEMORY = "MEMORY";
     public static final String FILE = "FILE";
     public static final String MULTI_PATH_SPLITTER = System.getProperty("dLedger.multiPath.Splitter", ",");
 
+    @Parameter(names = {"--config", "-c"}, description = "Config path of DLedger")
+    private String configFilePath;
+
     @Parameter(names = {"--group", "-g"}, description = "Group of this server")
     private String group = "default";
 
@@ -85,8 +93,8 @@ public class DLedgerConfig {
     @Parameter(names = {"--preferred-leader-id"}, description = "Preferred LeaderId")
     private String preferredLeaderIds;
     private long maxLeadershipTransferWaitIndex = 1000;
-    private int minTakeLeadershipVoteIntervalMs =  30;
-    private int maxTakeLeadershipVoteIntervalMs =  100;
+    private int minTakeLeadershipVoteIntervalMs = 30;
+    private int maxTakeLeadershipVoteIntervalMs = 100;
 
     private boolean isEnableBatchPush = false;
     private int maxBatchPushSize = 4 * 1024;
@@ -418,4 +426,50 @@ public String getReadOnlyDataStoreDirs() {
     public void setReadOnlyDataStoreDirs(String readOnlyDataStoreDirs) {
         this.readOnlyDataStoreDirs = readOnlyDataStoreDirs;
     }
+
+    private String selfAddress;
+
+    // groupId#selfIf -> address
+    private Map<String, String> peerAddressMap;
+
+    private final AtomicBoolean inited = new AtomicBoolean(false);
+
+    public void init() {
+        if (inited.compareAndSet(false, true)) {
+            initSelfAddress();
+            initPeerAddressMap();
+        }
+    }
+
+    private void initSelfAddress() {
+        for (String peerInfo : this.peers.split(";")) {
+            String peerSelfId = peerInfo.split("-")[0];
+            String peerAddress = peerInfo.substring(peerSelfId.length() + 1);
+            if (this.selfId.equals(peerSelfId)) {
+                this.selfAddress = peerAddress;
+                return;
+            }
+        }
+        // can't find itself
+        throw new IllegalArgumentException("[DLedgerConfig] fail to init self address, config: " + this);
+    }
+
+    private void initPeerAddressMap() {
+        Map<String, String> peerMap = new HashMap<>();
+        for (String peerInfo : this.peers.split(";")) {
+            String peerSelfId = peerInfo.split("-")[0];
+            String peerAddress = peerInfo.substring(peerSelfId.length() + 1);
+            peerMap.put(DLedgerUtils.generateDLedgerId(this.group, peerSelfId), peerAddress);
+        }
+        this.peerAddressMap = peerMap;
+    }
+
+    public String getSelfAddress() {
+        return this.selfAddress;
+    }
+
+
+    public Map<String, String> getPeerAddressMap() {
+        return this.peerAddressMap;
+    }
 }