Support to create hard link for current WAL (vesoft-inc#1227)

src/kvstore/raftex/RaftPart.cpp

@@ -1726,6 +1726,12 @@ AppendLogResult RaftPart::isCatchedUp(const HostAddr& peer) {
  return AppendLogResult::E_INVALID_PEER;
 }
 
+bool RaftPart::linkCurrentWAL(const char* newPath) {
+ CHECK_NOTNULL(newPath);
+ std::lock_guard<std::mutex> g(raftLock_);
+ return wal_->linkCurrentWAL(newPath);
+}
+
 } // namespace raftex
 } // namespace nebula
 

src/kvstore/raftex/RaftPart.h

@@ -181,6 +181,8 @@ class RaftPart : public std::enable_shared_from_this<RaftPart> {
  * */
  AppendLogResult isCatchedUp(const HostAddr& peer);
 
+ bool linkCurrentWAL(const char* newPath);
+
  /*****************************************************
  *
  * Methods to process incoming raft requests

src/kvstore/wal/FileBasedWal.cpp

@@ -591,6 +591,23 @@ std::unique_ptr<LogIterator> FileBasedWal::iterator(LogID firstLogId,
  return std::make_unique<FileBasedWalIterator>(shared_from_this(), firstLogId, lastLogId);
 }
 
+bool FileBasedWal::linkCurrentWAL(const char* newPath) {
+ closeCurrFile();
+ std::lock_guard<std::mutex> g(walFilesMutex_);
+ if (walFiles_.empty()) {
+ LOG(INFO) << idStr_ << "Create link failed, there is no wal files!";
+ return false;
+ }
+ auto it = walFiles_.rbegin();
+ if (link(it->second->path(), newPath) != 0) {
+ LOG(INFO) << idStr_ << "Create link failed for " << it->second->path()
+ << " on " << newPath << ", error:" << strerror(errno);
+ return false;
+ }
+ LOG(INFO) << idStr_ << "Create link success for " << it->second->path()
+ << " on " << newPath;
+ return true;
+}
 
 bool FileBasedWal::rollbackToLog(LogID id) {
  if (id < firstLogId_ - 1 || id > lastLogId_) {

src/kvstore/wal/FileBasedWal.h

@@ -43,6 +43,7 @@ class FileBasedWal final : public Wal
  , public std::enable_shared_from_this<FileBasedWal> {
  FRIEND_TEST(FileBasedWal, TTLTest);
  FRIEND_TEST(FileBasedWal, CheckLastWalTest);
+ FRIEND_TEST(FileBasedWal, LinkTest);
  friend class FileBasedWalIterator;
 public:
  // A factory method to create a new WAL
@@ -108,6 +109,9 @@ class FileBasedWal final : public Wal
  std::unique_ptr<LogIterator> iterator(LogID firstLogId,
  LogID lastLogId) override;
 
+ /** It is not thread-safe */
+ bool linkCurrentWAL(const char* newPath) override;
+
  // Iterates through all wal file info in reversed order
  // (from the latest to the earliest)
  // The iteration finishes when the functor returns false or reaches

src/kvstore/wal/Wal.h

@@ -41,6 +41,11 @@ class Wal {
  // Rollback to the given id, all logs after the id will be discarded
  virtual bool rollbackToLog(LogID id) = 0;
 
+ /**
+ * Create hard link for current wal on the new path.
+ * */
+ virtual bool linkCurrentWAL(const char* newPath) = 0;
+
  // Clean all wal files
  // This method is *NOT* thread safe
  virtual bool reset() = 0;

src/kvstore/wal/test/FileBasedWalTest.cpp

@@ -528,6 +528,32 @@ TEST(FileBasedWal, CheckLastWalTest) {
  }
 }
 
+TEST(FileBasedWal, LinkTest) {
+ TempDir walDir("/tmp/testWal.XXXXXX");
+ FileBasedWalPolicy policy;
+ policy.fileSize = 1024 * 512;
+ auto wal = FileBasedWal::getWal(walDir.path(),
+ "",
+ policy,
+ [](LogID, TermID, ClusterID, const std::string&) {
+ return true;
+ });
+ EXPECT_EQ(0, wal->lastLogId());
+ for (int i = 1; i <= 1000; i++) {
+ EXPECT_TRUE(
+ wal->appendLog(i /*id*/, 1 /*term*/, 0 /*cluster*/,
+ folly::stringPrintf(kLongMsg, i)));
+ }
+ auto snapshotFile = folly::stringPrintf("%s/snapshot", walDir.path());
+ CHECK(wal->linkCurrentWAL(snapshotFile.c_str()));
+ auto it = wal->walFiles_.rbegin();
+ EXPECT_EQ(FileUtils::fileSize(it->second->path()), FileUtils::fileSize(snapshotFile.c_str()));
+ auto num = wal->walFiles_.size();
+ EXPECT_TRUE(
+ wal->appendLog(1001 /*id*/, 1 /*term*/, 0 /*cluster*/,
+ folly::stringPrintf(kLongMsg, 1001)));
+ EXPECT_EQ(num + 1, wal->walFiles_.size());
+}
 
 } // namespace wal
 } // namespace nebula