use ts as suffix for LOG.old files

Summary: as subject and only maintain 10 log files.

Test Plan: new test in db_test

Reviewers: dhruba

Differential Revision: https://reviews.facebook.net/D4731

db/db_impl.cc

@@ -10,6 +10,7 @@
 #include <stdint.h>
 #include <stdio.h>
 #include <vector>
+#include <algorithm>
 #include "db/builder.h"
 #include "db/db_iter.h"
 #include "db/dbformat.h"
@@ -99,7 +100,8 @@ Options SanitizeOptions(const std::string& dbname,
   if (result.info_log == NULL) {
     // Open a log file in the same directory as the db
     src.env->CreateDir(dbname);  // In case it does not exist
-    src.env->RenameFile(InfoLogFileName(dbname), OldInfoLogFileName(dbname));
+    src.env->RenameFile(InfoLogFileName(dbname),
+        OldInfoLogFileName(dbname, src.env->NowMicros()));
     Status s = src.env->NewLogger(InfoLogFileName(dbname), &result.info_log);
     if (!s.ok()) {
       // No place suitable for logging
@@ -224,6 +226,7 @@ void DBImpl::DeleteObsoleteFiles() {
   env_->GetChildren(dbname_, &filenames); // Ignoring errors on purpose
   uint64_t number;
   FileType type;
+  std::vector<uint64_t> old_log_files_ts;
   for (size_t i = 0; i < filenames.size(); i++) {
     if (ParseFileName(filenames[i], &number, &type)) {
       bool keep = true;
@@ -245,9 +248,14 @@ void DBImpl::DeleteObsoleteFiles() {
           // be recorded in pending_outputs_, which is inserted into "live"
           keep = (live.find(number) != live.end());
           break;
+        case kInfoLogFile:
+          keep = true;
+          if (number != 0) {
+            old_log_files_ts.push_back(number);
+          }
+          break;
         case kCurrentFile:
         case kDBLockFile:
-        case kInfoLogFile:
           keep = true;
           break;
       }
@@ -263,6 +271,20 @@ void DBImpl::DeleteObsoleteFiles() {
       }
     }
   }
+
+  // Delete old log files.
+  int old_log_file_count = old_log_files_ts.size();
+  if (old_log_file_count >= KEEP_LOG_FILE_NUM) {
+    std::sort(old_log_files_ts.begin(), old_log_files_ts.end());
+    for (int i = 0; i >= (old_log_file_count - KEEP_LOG_FILE_NUM); i++) {
+      uint64_t ts = old_log_files_ts.at(i);
+      std::string to_delete = OldInfoLogFileName(dbname_, ts);
+      Log(options_.info_log, "Delete type=%d #%lld\n",
+          int(kInfoLogFile),
+          static_cast<unsigned long long>(ts));
+      env_->DeleteFile(dbname_ + "/" + to_delete);
+    }
+  }
 }
 
 Status DBImpl::Recover(VersionEdit* edit) {

db/db_impl.h

@@ -189,6 +189,8 @@ class DBImpl : public DB {
   };
   CompactionStats* stats_;
 
+  static const int KEEP_LOG_FILE_NUM = 1000;
+
   // No copying allowed
   DBImpl(const DBImpl&);
   void operator=(const DBImpl&);

db/db_test.cc

@@ -779,6 +779,22 @@ TEST(DBTest, Recover) {
   } while (ChangeOptions());
 }
 
+TEST(DBTest, RollLog) {
+  do {
+    ASSERT_OK(Put("foo", "v1"));
+    ASSERT_OK(Put("baz", "v5"));
+
+    Reopen();
+    for (int i = 0; i < 10; i++) {
+      Reopen();
+    }
+    ASSERT_OK(Put("foo", "v4"));
+    for (int i = 0; i < 10; i++) {
+      Reopen();
+    }
+  } while (ChangeOptions());
+}
+
 TEST(DBTest, WAL) {
   Options options = CurrentOptions();
   WriteOptions writeOpt = WriteOptions();

db/filename.cc

@@ -60,8 +60,10 @@ std::string InfoLogFileName(const std::string& dbname) {
 }
 
 // Return the name of the old info log file for "dbname".
-std::string OldInfoLogFileName(const std::string& dbname) {
+std::string OldInfoLogFileName(const std::string& dbname, uint64_t ts) {
-  return dbname + "/LOG.old";
+  char buf[50];
+  snprintf(buf, sizeof(buf), "%llu", static_cast<unsigned long long>(ts));
+  return dbname + "/LOG.old." + buf;
 }
 
 
@@ -69,7 +71,7 @@ std::string OldInfoLogFileName(const std::string& dbname) {
 //    dbname/CURRENT
 //    dbname/LOCK
 //    dbname/LOG
-//    dbname/LOG.old
+//    dbname/LOG.old.[0-9]+
 //    dbname/MANIFEST-[0-9]+
 //    dbname/[0-9]+.(log|sst)
 bool ParseFileName(const std::string& fname,
@@ -82,9 +84,17 @@ bool ParseFileName(const std::string& fname,
   } else if (rest == "LOCK") {
     *number = 0;
     *type = kDBLockFile;
-  } else if (rest == "LOG" || rest == "LOG.old") {
+  } else if (rest == "LOG") {
     *number = 0;
     *type = kInfoLogFile;
+  } else if (rest.starts_with("LOG.old.")) {
+    uint64_t ts_suffix;
+    rest.remove_prefix(sizeof("LOG.old."));
+    if (!ConsumeDecimalNumber(&rest, &ts_suffix)) {
+      return false;
+    }
+    *number = ts_suffix;
+    *type = kInfoLogFile;
   } else if (rest.starts_with("MANIFEST-")) {
     rest.remove_prefix(strlen("MANIFEST-"));
     uint64_t num;

db/filename.h

@@ -60,7 +60,7 @@ extern std::string TempFileName(const std::string& dbname, uint64_t number);
 extern std::string InfoLogFileName(const std::string& dbname);
 
 // Return the name of the old info log file for "dbname".
-extern std::string OldInfoLogFileName(const std::string& dbname);
+extern std::string OldInfoLogFileName(const std::string& dbname, uint64_t ts);
 
 // If filename is a leveldb file, store the type of the file in *type.
 // The number encoded in the filename is stored in *number.  If the