Support purging logs from separate log directory

Summary:
1. Support purging info logs from a separate paths from DB path. Refactor the codes of generating info log prefixes so that it can be called when generating new files and scanning log directory.
2. Fix the bug of not scanning multiple DB paths (should only impact multiple DB paths)

Test Plan:
Add unit test for generating and parsing info log files
Add end-to-end test in db_test

Reviewers: yhchiang, ljin

Reviewed By: ljin

Subscribers: leveldb, igor, dhruba

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

db/db_impl.cc

@@ -585,7 +585,8 @@ void DBImpl::FindObsoleteFiles(DeletionState& deletion_state,
       // set of all files in the directory. We'll exclude files that are still
       // alive in the subsequent processings.
       std::vector<std::string> files;
-      env_->GetChildren(dbname_, &files);  // Ignore errors
+      env_->GetChildren(options_.db_paths[path_id].path,
+                        &files);  // Ignore errors
       for (std::string file : files) {
         deletion_state.candidate_files.emplace_back(file, path_id);
       }
@@ -599,6 +600,14 @@ void DBImpl::FindObsoleteFiles(DeletionState& deletion_state,
         deletion_state.candidate_files.emplace_back(log_file, 0);
       }
     }
+    // Add info log files in db_log_dir
+    if (!options_.db_log_dir.empty() && options_.db_log_dir != dbname_) {
+      std::vector<std::string> info_log_files;
+      env_->GetChildren(options_.db_log_dir, &info_log_files);  // Ignore errors
+      for (std::string log_file : info_log_files) {
+        deletion_state.candidate_files.emplace_back(log_file, 0);
+      }
+    }
   }
 }
 
@@ -665,14 +674,14 @@ void DBImpl::PurgeObsoleteFiles(DeletionState& state) {
                         candidate_files.end());
 
   std::vector<std::string> old_info_log_files;
-
+  InfoLogPrefix info_log_prefix(!options_.db_log_dir.empty(), dbname_);
   for (const auto& candidate_file : candidate_files) {
     std::string to_delete = candidate_file.file_name;
     uint32_t path_id = candidate_file.path_id;
     uint64_t number;
     FileType type;
     // Ignore file if we cannot recognize it.
-    if (!ParseFileName(to_delete, &number, &type)) {
+    if (!ParseFileName(to_delete, &number, info_log_prefix.prefix, &type)) {
       continue;
     }
 
@@ -747,16 +756,17 @@ void DBImpl::PurgeObsoleteFiles(DeletionState& state) {
 
   // Delete old info log files.
   size_t old_info_log_file_count = old_info_log_files.size();
-  // NOTE: Currently we only support log purge when options_.db_log_dir is
-  // located in `dbname` directory.
-  if (old_info_log_file_count >= options_.keep_log_file_num &&
-      options_.db_log_dir.empty()) {
+  if (old_info_log_file_count >= options_.keep_log_file_num) {
     std::sort(old_info_log_files.begin(), old_info_log_files.end());
     size_t end = old_info_log_file_count - options_.keep_log_file_num;
     for (unsigned int i = 0; i <= end; i++) {
       std::string& to_delete = old_info_log_files.at(i);
-      Log(options_.info_log, "Delete info log file %s\n", to_delete.c_str());
-      Status s = env_->DeleteFile(dbname_ + "/" + to_delete);
+      std::string full_path_to_delete =
+          (options_.db_log_dir.empty() ? dbname_ : options_.db_log_dir) + "/" +
+          to_delete;
+      Log(options_.info_log, "Delete info log file %s\n",
+          full_path_to_delete.c_str());
+      Status s = env_->DeleteFile(full_path_to_delete);
       if (!s.ok()) {
         Log(options_.info_log, "Delete info log file %s FAILED -- %s\n",
             to_delete.c_str(), s.ToString().c_str());

db/db_test.cc

@@ -6054,6 +6054,37 @@ TEST(DBTest, WALArchivalSizeLimit) {
   } while (ChangeCompactOptions());
 }
 
+TEST(DBTest, PurgeInfoLogs) {
+  Options options = CurrentOptions();
+  options.keep_log_file_num = 5;
+  options.create_if_missing = true;
+  for (int mode = 0; mode <= 1; mode++) {
+    if (mode == 1) {
+      options.db_log_dir = dbname_ + "_logs";
+      env_->CreateDirIfMissing(options.db_log_dir);
+    } else {
+      options.db_log_dir = "";
+    }
+    for (int i = 0; i < 8; i++) {
+      Reopen(&options);
+    }
+
+    std::vector<std::string> files;
+    env_->GetChildren(options.db_log_dir.empty() ? dbname_ : options.db_log_dir,
+                      &files);
+    int info_log_count = 0;
+    for (std::string file : files) {
+      if (file.find("LOG") != std::string::npos) {
+        if (mode == 1) {
+          env_->DeleteFile(options.db_log_dir + "/" + file);
+        }
+        info_log_count++;
+      }
+    }
+    ASSERT_EQ(5, info_log_count);
+  }
+}
+
 namespace {
 SequenceNumber ReadRecords(
     std::unique_ptr<TransactionLogIterator>& iter,

db/filename.cc

@@ -20,14 +20,16 @@
 namespace rocksdb {
 
 // Given a path, flatten the path name by replacing all chars not in
-// {[0-9,a-z,A-Z,-,_,.]} with _. And append '\0' at the end.
+// {[0-9,a-z,A-Z,-,_,.]} with _. And append '_LOG\0' at the end.
 // Return the number of chars stored in dest not including the trailing '\0'.
-static int FlattenPath(const std::string& path, char* dest, int len) {
-  int write_idx = 0;
-  int i = 0;
-  int src_len = path.size();
+static size_t GetInfoLogPrefix(const std::string& path, char* dest, int len) {
+  const char suffix[] = "_LOG";
 
-  while (i < src_len && write_idx < len - 1) {
+  size_t write_idx = 0;
+  size_t i = 0;
+  size_t src_len = path.size();
+
+  while (i < src_len && write_idx < len - sizeof(suffix)) {
     if ((path[i] >= 'a' && path[i] <= 'z') ||
         (path[i] >= '0' && path[i] <= '9') ||
         (path[i] >= 'A' && path[i] <= 'Z') ||
@@ -41,8 +43,10 @@ static int FlattenPath(const std::string& path, char* dest, int len) {
     }
     i++;
   }
-
-  dest[write_idx] = '\0';
+  assert(sizeof(suffix) <= len - write_idx);
+  // "\0" is automatically added by snprintf
+  snprintf(dest + write_idx, len - write_idx, suffix);
+  write_idx += sizeof(suffix) - 1;
   return write_idx;
 }
 
@@ -118,14 +122,26 @@ std::string TempFileName(const std::string& dbname, uint64_t number) {
   return MakeFileName(dbname, number, "dbtmp");
 }
 
+InfoLogPrefix::InfoLogPrefix(bool has_log_dir,
+                             const std::string& db_absolute_path) {
+  if (!has_log_dir) {
+    const char kInfoLogPrefix[] = "LOG";
+    // "\0" is automatically added to the end
+    snprintf(buf, sizeof(buf), kInfoLogPrefix);
+    prefix = Slice(buf, sizeof(kInfoLogPrefix) - 1);
+  } else {
+    size_t len = GetInfoLogPrefix(db_absolute_path, buf, sizeof(buf));
+    prefix = Slice(buf, len);
+  }
+}
+
 std::string InfoLogFileName(const std::string& dbname,
     const std::string& db_path, const std::string& log_dir) {
   if (log_dir.empty())
     return dbname + "/LOG";
 
-  char flatten_db_path[256];
-  FlattenPath(db_path, flatten_db_path, 256);
-  return log_dir + "/" + flatten_db_path + "_LOG";
+  InfoLogPrefix info_log_prefix(true, db_path);
+  return log_dir + "/" + info_log_prefix.buf;
 }
 
 // Return the name of the old info log file for "dbname".
@@ -137,9 +153,8 @@ std::string OldInfoLogFileName(const std::string& dbname, uint64_t ts,
   if (log_dir.empty())
     return dbname + "/LOG.old." + buf;
 
-  char flatten_db_path[256];
-  FlattenPath(db_path, flatten_db_path, 256);
-  return log_dir + "/" + flatten_db_path + "_LOG.old." + buf;
+  InfoLogPrefix info_log_prefix(true, db_path);
+  return log_dir + "/" + info_log_prefix.buf + ".old." + buf;
 }
 
 std::string MetaDatabaseName(const std::string& dbname, uint64_t number) {
@@ -157,8 +172,8 @@ std::string IdentityFileName(const std::string& dbname) {
 //    dbname/IDENTITY
 //    dbname/CURRENT
 //    dbname/LOCK
-//    dbname/LOG
-//    dbname/LOG.old.[0-9]+
+//    dbname/<info_log_name_prefix>
+//    dbname/<info_log_name_prefix>.old.[0-9]+
 //    dbname/MANIFEST-[0-9]+
 //    dbname/[0-9]+.(log|sst)
 //    dbname/METADB-[0-9]+
@@ -167,6 +182,12 @@ bool ParseFileName(const std::string& fname,
                    uint64_t* number,
                    FileType* type,
                    WalFileType* log_type) {
+  return ParseFileName(fname, number, "", type, log_type);
+}
+
+bool ParseFileName(const std::string& fname, uint64_t* number,
+                   const Slice& info_log_name_prefix, FileType* type,
+                   WalFileType* log_type) {
   Slice rest(fname);
   if (fname.length() > 1 && fname[0] == '/') {
     rest.remove_prefix(1);
@@ -180,18 +201,22 @@ bool ParseFileName(const std::string& fname,
   } else if (rest == "LOCK") {
     *number = 0;
     *type = kDBLockFile;
-  } else if (rest == "LOG" || rest == "LOG.old") {
+  } else if (info_log_name_prefix.size() > 0 &&
+             rest.starts_with(info_log_name_prefix)) {
+    rest.remove_prefix(info_log_name_prefix.size());
+    if (rest == "" || rest == ".old") {
       *number = 0;
       *type = kInfoLogFile;
-  } else if (rest.starts_with("LOG.old.")) {
+    } else if (rest.starts_with(".old.")) {
       uint64_t ts_suffix;
       // sizeof also counts the trailing '\0'.
-    rest.remove_prefix(sizeof("LOG.old.") - 1);
+      rest.remove_prefix(sizeof(".old.") - 1);
       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

@@ -86,6 +86,16 @@ extern std::string LockFileName(const std::string& dbname);
 // The result will be prefixed with "dbname".
 extern std::string TempFileName(const std::string& dbname, uint64_t number);
 
+// A helper structure for prefix of info log names.
+struct InfoLogPrefix {
+  char buf[260];
+  Slice prefix;
+  // Prefix with DB absolute path encoded
+  explicit InfoLogPrefix(bool has_log_dir, const std::string& db_absolute_path);
+  // Default Prefix
+  explicit InfoLogPrefix();
+};
+
 // Return the name of the info log file for "dbname".
 extern std::string InfoLogFileName(const std::string& dbname,
     const std::string& db_path="", const std::string& log_dir="");
@@ -107,10 +117,13 @@ extern std::string IdentityFileName(const std::string& dbname);
 // If filename is a rocksdb file, store the type of the file in *type.
 // The number encoded in the filename is stored in *number.  If the
 // filename was successfully parsed, returns true.  Else return false.
-extern bool ParseFileName(const std::string& filename,
-                          uint64_t* number,
-                          FileType* type,
+// info_log_name_prefix is the path of info logs.
+extern bool ParseFileName(const std::string& filename, uint64_t* number,
+                          const Slice& info_log_name_prefix, FileType* type,
                           WalFileType* log_type = nullptr);
+// Same as previous function, but skip info log files.
+extern bool ParseFileName(const std::string& filename, uint64_t* number,
+                          FileType* type, WalFileType* log_type = nullptr);
 
 // Make the CURRENT file point to the descriptor file with the
 // specified number.

db/filename_test.cc

@@ -23,31 +23,51 @@ TEST(FileNameTest, Parse) {
   FileType type;
   uint64_t number;
 
+  char kDefautInfoLogDir = 1;
+  char kDifferentInfoLogDir = 2;
+  char kNoCheckLogDir = 4;
+  char kAllMode = kDefautInfoLogDir | kDifferentInfoLogDir | kNoCheckLogDir;
+
   // Successful parses
   static struct {
     const char* fname;
     uint64_t number;
     FileType type;
+    char mode;
   } cases[] = {
-    { "100.log",            100,   kLogFile },
-    { "0.log",              0,     kLogFile },
-    { "0.sst",              0,     kTableFile },
-    { "CURRENT",            0,     kCurrentFile },
-    { "LOCK",               0,     kDBLockFile },
-    { "MANIFEST-2",         2,     kDescriptorFile },
-    { "MANIFEST-7",         7,     kDescriptorFile },
-    { "METADB-2",           2,     kMetaDatabase },
-    { "METADB-7",           7,     kMetaDatabase },
-    { "LOG",                0,     kInfoLogFile },
-    { "LOG.old",            0,     kInfoLogFile },
-    { "18446744073709551615.log", 18446744073709551615ull, kLogFile },
-  };
+        {"100.log", 100, kLogFile, kAllMode},
+        {"0.log", 0, kLogFile, kAllMode},
+        {"0.sst", 0, kTableFile, kAllMode},
+        {"CURRENT", 0, kCurrentFile, kAllMode},
+        {"LOCK", 0, kDBLockFile, kAllMode},
+        {"MANIFEST-2", 2, kDescriptorFile, kAllMode},
+        {"MANIFEST-7", 7, kDescriptorFile, kAllMode},
+        {"METADB-2", 2, kMetaDatabase, kAllMode},
+        {"METADB-7", 7, kMetaDatabase, kAllMode},
+        {"LOG", 0, kInfoLogFile, kDefautInfoLogDir},
+        {"LOG.old", 0, kInfoLogFile, kDefautInfoLogDir},
+        {"LOG.old.6688", 6688, kInfoLogFile, kDefautInfoLogDir},
+        {"rocksdb_dir_LOG", 0, kInfoLogFile, kDifferentInfoLogDir},
+        {"rocksdb_dir_LOG.old", 0, kInfoLogFile, kDifferentInfoLogDir},
+        {"rocksdb_dir_LOG.old.6688", 6688, kInfoLogFile, kDifferentInfoLogDir},
+        {"18446744073709551615.log", 18446744073709551615ull, kLogFile,
+         kAllMode}, };
+  for (char mode : {kDifferentInfoLogDir, kDefautInfoLogDir, kNoCheckLogDir}) {
     for (unsigned int i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
+      InfoLogPrefix info_log_prefix(mode != kDefautInfoLogDir, "/rocksdb/dir");
+      if (cases[i].mode & mode) {
         std::string f = cases[i].fname;
+        if (mode == kNoCheckLogDir) {
           ASSERT_TRUE(ParseFileName(f, &number, &type)) << f;
+        } else {
+          ASSERT_TRUE(ParseFileName(f, &number, info_log_prefix.prefix, &type))
+              << f;
+        }
         ASSERT_EQ(cases[i].type, type) << f;
         ASSERT_EQ(cases[i].number, number) << f;
       }
+    }
+  }
 
   // Errors
   static const char* errors[] = {
@@ -85,6 +105,22 @@ TEST(FileNameTest, Parse) {
   };
 }
 
+TEST(FileNameTest, InfoLogFileName) {
+  std::string dbname = ("/data/rocksdb");
+  std::string db_absolute_path;
+  Env::Default()->GetAbsolutePath(dbname, &db_absolute_path);
+
+  ASSERT_EQ("/data/rocksdb/LOG", InfoLogFileName(dbname, db_absolute_path, ""));
+  ASSERT_EQ("/data/rocksdb/LOG.old.666",
+            OldInfoLogFileName(dbname, 666u, db_absolute_path, ""));
+
+  ASSERT_EQ("/data/rocksdb_log/data_rocksdb_LOG",
+            InfoLogFileName(dbname, db_absolute_path, "/data/rocksdb_log"));
+  ASSERT_EQ(
+      "/data/rocksdb_log/data_rocksdb_LOG.old.666",
+      OldInfoLogFileName(dbname, 666u, db_absolute_path, "/data/rocksdb_log"));
+}
+
 TEST(FileNameTest, Construction) {
   uint64_t number;
   FileType type;