Flush rocksdb before clean expired wal

src/kvstore/NebulaStore.cpp

@@ -18,6 +18,7 @@ DEFINE_string(engine_type, "rocksdb", "rocksdb, memory...");
 DEFINE_int32(custom_filter_interval_secs, 24 * 3600, "interval to trigger custom compaction");
 DEFINE_int32(num_workers, 4, "Number of worker threads");
 DEFINE_bool(check_leader, true, "Check leader or not");
+DEFINE_int32(clean_wal_interval_secs, 600, "inerval to trigger clean expired wal");
 
 namespace nebula {
 namespace kvstore {
@@ -185,6 +186,8 @@ bool NebulaStore::init() {
         }
     }
 
+    bgWorkers_->addDelayTask(FLAGS_clean_wal_interval_secs * 1000, &NebulaStore::cleanWAL, this);
+
     LOG(INFO) << "Register handler...";
     options_.partMan_->registerHandler(this);
     return true;
@@ -820,6 +823,19 @@ bool NebulaStore::checkLeader(std::shared_ptr<Part> part) const {
     return !FLAGS_check_leader || (part->isLeader() && part->leaseValid());
 }
 
+void NebulaStore::cleanWAL() {
+    for (const auto& spaceEntry : spaces_) {
+        for (const auto& engine : spaceEntry.second->engines_) {
+            engine->flush();
+        }
+        for (const auto& partEntry : spaceEntry.second->parts_) {
+            auto& part = partEntry.second;
+            if (part->needToCleanWal()) {
+                part->wal()->cleanWAL();
+            }
+        }
+    }
+}
 
 }  // namespace kvstore
 }  // namespace nebula

src/kvstore/NebulaStore.h

@@ -235,6 +235,8 @@ class NebulaStore : public KVStore, public Handler {
 
     bool checkLeader(std::shared_ptr<Part> part) const;
 
+    void cleanWAL();
+
 private:
     // The lock used to protect spaces_
     folly::RWSpinLock lock_;

src/kvstore/raftex/RaftPart.cpp

@@ -1223,9 +1223,6 @@ void RaftPart::statusPolling() {
         LOG(INFO) << idStr_ << "Clean up the snapshot";
         cleanupSnapshot();
     }
-    if (needToCleanWal()) {
-        wal_->cleanWAL(FLAGS_wal_ttl);
-    }
     {
         std::lock_guard<std::mutex> g(raftLock_);
         if (status_ == Status::RUNNING || status_ == Status::WAITING_SNAPSHOT) {

src/kvstore/raftex/RaftPart.h

@@ -211,6 +211,8 @@ class RaftPart : public std::enable_shared_from_this<RaftPart> {
 
     bool leaseValid();
 
+    bool needToCleanWal();
+
 protected:
     // Protected constructor to prevent from instantiating directly
     RaftPart(ClusterID clusterId,
@@ -332,8 +334,6 @@ class RaftPart : public std::enable_shared_from_this<RaftPart> {
 
     void cleanupSnapshot();
 
-    bool needToCleanWal();
-
     // The method sends out AskForVote request
     // It return true if a leader is elected, otherwise returns false
     bool leaderElection();