[IOTDB-351] Serialize raft log

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

@@ -47,3 +47,6 @@ REPLICA_NUM=3
 # connection time out (ms) among raft nodes
 CONNECTION_TIME_OUT_MS=20000
 
+# when the logs size larger than this, we actually delete snapshoted logs, the unit is bytes
+MAX_REMOVED_LOG_SIZE=134217728
+

cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java

@@ -43,7 +43,19 @@ public class ClusterConfig {
   private int replicationNum = 3;
 
   private int connectionTimeoutInMS = 20 * 1000;
+  /**
+   * This parameter controls when to actually delete snapshoted logs because we can't remove
+   * snapshoted logs directly from disk now
+   */
+  private long maxRemovedLogSize = 1024 * 1024 * 128;
 
+  public long getMaxRemovedLogSize() {
+    return maxRemovedLogSize;
+  }
+
+  public void setMaxRemovedLogSize(long maxRemovedLogSize) {
+    this.maxRemovedLogSize = maxRemovedLogSize;
+  }
 
   public String getLocalIP() {
     return localIP;

cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java

@@ -26,7 +26,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
-import org.apache.iotdb.cluster.server.RaftServer;
 import org.apache.iotdb.db.conf.IoTDBConstant;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -109,6 +108,10 @@ private void loadProps() {
         .setConnectionTimeoutInMS(Integer.parseInt(properties.getProperty("CONNECTION_TIME_OUT_MS",
             String.valueOf(config.getConnectionTimeoutInMS()))));
 
+    config
+        .setMaxRemovedLogSize(Long.parseLong(properties.getProperty("MAX_REMOVED_LOG_SIZE",
+            String.valueOf(config.getMaxRemovedLogSize()))));
+
     String seedUrls = properties.getProperty("SEED_NODES");
     if (seedUrls != null) {
       List<String> urlList = new ArrayList<>();

cluster/src/main/java/org/apache/iotdb/cluster/exception/MemberReadOnlyException.java

@@ -1,3 +1,22 @@
+/*
+ * 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.cluster.exception;
 
 import org.apache.iotdb.cluster.rpc.thrift.Node;

cluster/src/main/java/org/apache/iotdb/cluster/log/Log.java

@@ -96,4 +96,5 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hash(currLogIndex, currLogTerm, previousLogIndex, previousLogTerm);
   }
+
 }

cluster/src/main/java/org/apache/iotdb/cluster/log/LogManager.java

@@ -21,7 +21,6 @@
 
 import java.io.IOException;
 import java.util.List;
-import org.apache.iotdb.db.exception.query.QueryProcessException;
 
 /**
  * LogManager manages the logs that are still in memory and the last snapshot which can be used

cluster/src/main/java/org/apache/iotdb/cluster/log/logtypes/AddNodeLog.java

@@ -40,6 +40,7 @@ public void setNewNode(Node newNode) {
     this.newNode = newNode;
   }
 
+
   @Override
   public ByteBuffer serialize() {
     byte[] ipBytes = newNode.getIp().getBytes();

cluster/src/main/java/org/apache/iotdb/cluster/log/logtypes/PhysicalPlanLog.java

@@ -45,6 +45,7 @@ public class PhysicalPlanLog extends Log {
   private PhysicalPlan plan;
 
   public PhysicalPlanLog() {
+
   }
 
   public PhysicalPlanLog(PhysicalPlan plan) {

cluster/src/main/java/org/apache/iotdb/cluster/log/logtypes/RemoveNodeLog.java

@@ -1,3 +1,22 @@
+/*
+ * 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.cluster.log.logtypes;
 
 import java.io.IOException;

cluster/src/main/java/org/apache/iotdb/cluster/log/manage/DiskLogManager.java

@@ -0,0 +1,143 @@
+/*
+ * 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.cluster.log.manage;
+
+import org.apache.iotdb.cluster.log.Log;
+import org.apache.iotdb.cluster.log.LogApplier;
+import org.apache.iotdb.cluster.log.manage.serializable.LogDequeSerializer;
+import org.apache.iotdb.cluster.log.manage.serializable.LogManagerMeta;
+import org.apache.iotdb.cluster.log.manage.serializable.SyncLogDequeSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class DiskLogManager extends MemoryLogManager {
+  private static final Logger logger = LoggerFactory.getLogger(DiskLogManager.class);
+
+  // manage logs in disk
+  private LogDequeSerializer logDequeSerializer;
+
+  private LogManagerMeta managerMeta = new LogManagerMeta();
+
+
+  protected DiskLogManager(LogApplier logApplier) {
+    super(logApplier);
+    logDequeSerializer = new SyncLogDequeSerializer();
+    recovery();
+  }
+
+  private void recovery(){
+    // recover meta
+    LogManagerMeta logManagerMeta = logDequeSerializer.recoverMeta();
+    if(logManagerMeta != null){
+      setCommitLogIndex(logManagerMeta.getCommitLogIndex());
+      setLastLogId(logManagerMeta.getLastLogId());
+      setLastLogTerm(logManagerMeta.getLastLogTerm());
+    }
+    // recover logs
+    setLogBuffer(logDequeSerializer.recoverLog());
+  }
+
+
+  @Override
+  public long getLastLogIndex() {
+    return lastLogId;
+  }
+
+  @Override
+  public long getLastLogTerm() {
+    return lastLogTerm;
+  }
+
+  @Override
+  public void setLastLogTerm(long lastLogTerm) {
+    this.lastLogTerm = lastLogTerm;
+  }
+
+  @Override
+  public long getCommitLogIndex() {
+    return commitLogIndex;
+  }
+
+  @Override
+  public boolean appendLog(Log log) {
+    boolean result = super.appendLog(log);
+    if(result) {
+      logDequeSerializer.addLast(log, getMeta());
+    }
+
+    return result;
+  }
+
+
+  public void truncateLog(int count) {
+    if (logBuffer.size() > count) {
+      // do super truncate log
+      // super.truncateLog();
+      logDequeSerializer.truncateLog(count, getMeta());
+    }
+  }
+
+  @Override
+  public synchronized void commitLog(long maxLogIndex) {
+    super.commitLog(maxLogIndex);
+    // save commit log index
+    serializeMeta();
+  }
+
+
+  @Override
+  public void setLastLogId(long lastLogId) {
+    super.setLastLogId(lastLogId);
+    // save meta
+    serializeMeta();
+  }
+
+  /**
+   * refresh meta info
+   * @return meta info
+   */
+  public LogManagerMeta getMeta(){
+    managerMeta.setCommitLogIndex(commitLogIndex);
+    managerMeta.setLastLogId(lastLogId);
+    managerMeta.setLastLogTerm(lastLogTerm);
+
+    return managerMeta;
+  }
+
+  /**
+   * serialize meta data of this log manager
+   */
+  private void serializeMeta(){
+    logDequeSerializer.serializeMeta(getMeta());
+  }
+
+
+  @Override
+  public void removeFromHead(int length){
+    super.removeFromHead(length);
+    logDequeSerializer.removeFirst(length);
+  }
+
+  /**
+   * close file and release resource
+   */
+  public void close(){
+    logDequeSerializer.close();
+  }
+}

cluster/src/main/java/org/apache/iotdb/cluster/log/manage/FilePartitionedSnapshotLogManager.java

@@ -73,6 +73,7 @@ public void takeSnapshot() throws IOException {
     synchronized (slotSnapshots) {
       collectTimeseriesSchemas();
 
+
       int i = 0;
       for (; i < logBuffer.size(); i++) {
         if (logBuffer.get(i).getCurrLogIndex() > commitLogIndex) {
@@ -81,7 +82,7 @@ public void takeSnapshot() throws IOException {
         snapshotLastLogId = logBuffer.get(i).getCurrLogIndex();
         snapshotLastLogTerm = logBuffer.get(i).getCurrLogTerm();
       }
-      logBuffer.subList(0, i).clear();
+      removeFromHead(i);
 
       collectTsFiles();