ISSUE2620: RocksDB log path configurable

RocksDB configuration now allows a new value "dbStorage_rocksDB_logPath"
which allows the log files to be separated from the data storage. When
not set, Bookkeeper has no change in behavior.

Descriptions of the changes in this PR:



### Motivation

(Explain: why you're making that change, what is the problem you're trying to solve)

### Changes

(Describe: what changes you have made)

Master Issue: #<master-issue-number>



Reviewers: Enrico Olivelli <eolivelli@gmail.com>, lipenghui <penghui@apache.org>, Matteo Merli <mmerli@apache.org>

This closes apache#2698 from dinghram/RocksDBLogPath and squashes the following commits:

bd8e212 [Don Inghram] Issue2620-RocksDB log path configurable: conf
eeba052 [Don Inghram] Issue2620-RocksDB log path configurable

Author: dinghram

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java

@@ -314,7 +314,8 @@ public static void readLedgerIndexEntries(long ledgerId, ServerConfiguration ser
         String ledgerBasePath = ledgerDirs.get(dirIndex).toString();
 
         EntryLocationIndex entryLocationIndex = new EntryLocationIndex(serverConf,
-                (path, dbConfigType, conf1) -> new KeyValueStorageRocksDB(path, DbConfigType.Small, conf1, true),
+                (basePath, subPath, dbConfigType, conf1) ->
+                        new KeyValueStorageRocksDB(basePath, subPath, DbConfigType.Small, conf1, true),
                 ledgerBasePath, NullStatsLogger.INSTANCE);
         try {
             long lastEntryId = entryLocationIndex.getLastEntryInLedger(ledgerId);

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/EntryLocationIndex.java

@@ -24,7 +24,6 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.nio.file.FileSystems;
 import java.util.Map.Entry;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
@@ -54,8 +53,7 @@ public class EntryLocationIndex implements Closeable {
 
     public EntryLocationIndex(ServerConfiguration conf, KeyValueStorageFactory storageFactory, String basePath,
             StatsLogger stats) throws IOException {
-        String locationsDbPath = FileSystems.getDefault().getPath(basePath, "locations").toFile().toString();
-        locationsDb = storageFactory.newKeyValueStorage(locationsDbPath, DbConfigType.Huge, conf);
+        locationsDb = storageFactory.newKeyValueStorage(basePath, "locations", DbConfigType.Huge, conf);
 
         this.stats = new EntryLocationIndexStats(
             stats,

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageFactory.java

@@ -37,6 +37,7 @@ enum DbConfigType {
         Huge // Used for location index, lots of writes and much bigger dataset
     }
 
-    KeyValueStorage newKeyValueStorage(String path, DbConfigType dbConfigType, ServerConfiguration conf)
+    KeyValueStorage newKeyValueStorage(String defaultBasePath, String subPath, DbConfigType dbConfigType,
+                                       ServerConfiguration conf)
             throws IOException;
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageRocksDB.java

@@ -27,6 +27,9 @@
 //CHECKSTYLE.ON: IllegalImport
 
 import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Map.Entry;
 import java.util.concurrent.TimeUnit;
 
@@ -55,8 +58,8 @@
  */
 public class KeyValueStorageRocksDB implements KeyValueStorage {
 
-    static KeyValueStorageFactory factory = (path, dbConfigType, conf) -> new KeyValueStorageRocksDB(path, dbConfigType,
-            conf);
+    static KeyValueStorageFactory factory = (defaultBasePath, subPath, dbConfigType, conf) ->
+            new KeyValueStorageRocksDB(defaultBasePath, subPath, dbConfigType, conf);
 
     private final RocksDB db;
 
@@ -68,6 +71,7 @@ public class KeyValueStorageRocksDB implements KeyValueStorage {
 
     private final WriteBatch emptyBatch;
 
+    private static final String ROCKSDB_LOG_PATH = "dbStorage_rocksDB_logPath";
     private static final String ROCKSDB_LOG_LEVEL = "dbStorage_rocksDB_logLevel";
     private static final String ROCKSDB_LZ4_COMPRESSION_ENABLED = "dbStorage_rocksDB_lz4CompressionEnabled";
     private static final String ROCKSDB_WRITE_BUFFER_SIZE_MB = "dbStorage_rocksDB_writeBufferSizeMB";
@@ -79,11 +83,13 @@ public class KeyValueStorageRocksDB implements KeyValueStorage {
     private static final String ROCKSDB_NUM_FILES_IN_LEVEL0 = "dbStorage_rocksDB_numFilesInLevel0";
     private static final String ROCKSDB_MAX_SIZE_IN_LEVEL1_MB = "dbStorage_rocksDB_maxSizeInLevel1MB";
 
-    public KeyValueStorageRocksDB(String path, DbConfigType dbConfigType, ServerConfiguration conf) throws IOException {
-        this(path, dbConfigType, conf, false);
+    public KeyValueStorageRocksDB(String basePath, String subPath, DbConfigType dbConfigType, ServerConfiguration conf)
+            throws IOException {
+        this(basePath, subPath, dbConfigType, conf, false);
     }
 
-    public KeyValueStorageRocksDB(String path, DbConfigType dbConfigType, ServerConfiguration conf, boolean readOnly)
+    public KeyValueStorageRocksDB(String basePath, String subPath, DbConfigType dbConfigType, ServerConfiguration conf,
+                                  boolean readOnly)
             throws IOException {
         try {
             RocksDB.loadLibrary();
@@ -150,6 +156,16 @@ public KeyValueStorageRocksDB(String path, DbConfigType dbConfigType, ServerConf
                 options.setTableFormatConfig(tableOptions);
             }
 
+            // Configure file path
+            String logPath = conf.getString(ROCKSDB_LOG_PATH, "");
+            if (!logPath.isEmpty()) {
+                Path logPathSetting = FileSystems.getDefault().getPath(logPath, subPath);
+                Files.createDirectories(logPathSetting);
+                log.info("RocksDB<{}> log path: {}", subPath, logPathSetting);
+                options.setDbLogDir(logPathSetting.toString());
+            }
+            String path = FileSystems.getDefault().getPath(basePath, subPath).toFile().toString();
+
             // Configure log level
             String logLevel = conf.getString(ROCKSDB_LOG_LEVEL, "info");
             switch (logLevel) {

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java

@@ -27,7 +27,6 @@
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.nio.file.FileSystems;
 import java.util.AbstractMap.SimpleEntry;
 import java.util.Arrays;
 import java.util.Map.Entry;
@@ -66,8 +65,7 @@ public class LedgerMetadataIndex implements Closeable {
 
     public LedgerMetadataIndex(ServerConfiguration conf, KeyValueStorageFactory storageFactory, String basePath,
             StatsLogger stats) throws IOException {
-        String ledgersPath = FileSystems.getDefault().getPath(basePath, "ledgers").toFile().toString();
-        ledgersDb = storageFactory.newKeyValueStorage(ledgersPath, DbConfigType.Small, conf);
+        ledgersDb = storageFactory.newKeyValueStorage(basePath, "ledgers", DbConfigType.Small, conf);
 
         ledgers = new ConcurrentLongHashMap<>();
         ledgersCount = new AtomicInteger();

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LocationsIndexRebuildOp.java

@@ -73,13 +73,11 @@ public void initiate() throws IOException {
                 new DiskChecker(conf.getDiskUsageThreshold(), conf.getDiskUsageWarnThreshold())));
         Set<Long> entryLogs = entryLogger.getEntryLogsSet();
 
-        String locationsDbPath = FileSystems.getDefault().getPath(basePath, "locations").toFile().toString();
-
         Set<Long> activeLedgers = getActiveLedgers(conf, KeyValueStorageRocksDB.factory, basePath);
         LOG.info("Found {} active ledgers in ledger manager", activeLedgers.size());
 
-        KeyValueStorage newIndex = KeyValueStorageRocksDB.factory.newKeyValueStorage(locationsDbPath, DbConfigType.Huge,
-                conf);
+        KeyValueStorage newIndex = KeyValueStorageRocksDB.factory.newKeyValueStorage(basePath, "locations",
+                DbConfigType.Huge, conf);
 
         int totalEntryLogs = entryLogs.size();
         int completedEntryLogs = 0;

bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/KeyValueStorageTest.java

@@ -26,6 +26,8 @@
 import com.google.common.collect.Lists;
 
 import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
@@ -35,6 +37,7 @@
 import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorage.CloseableIterator;
 import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorageFactory.DbConfigType;
 import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.commons.io.FileUtils;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -71,10 +74,10 @@ private static byte[] toArray(long n) {
 
     @Test
     public void simple() throws Exception {
-        File tmpDir = File.createTempFile("bookie", "test");
-        tmpDir.delete();
+        File tmpDir = Files.createTempDirectory("junitTemporaryFolder").toFile();
+        Files.createDirectory(Paths.get(tmpDir.toString(), "subDir"));
 
-        KeyValueStorage db = storageFactory.newKeyValueStorage(tmpDir.getAbsolutePath(), DbConfigType.Small,
+        KeyValueStorage db = storageFactory.newKeyValueStorage(tmpDir.toString(), "subDir", DbConfigType.Small,
                 configuration);
 
         assertEquals(null, db.getFloor(toArray(3)));
@@ -167,6 +170,6 @@ public void simple() throws Exception {
         batch.close();
 
         db.close();
-        tmpDir.delete();
+        FileUtils.deleteDirectory(tmpDir);
     }
 }

conf/bk_server.conf

@@ -719,6 +719,7 @@ ledgerDirectories=/tmp/bk-data
 # dbStorage_rocksDB_numLevels=-1
 # dbStorage_rocksDB_numFilesInLevel0=4
 # dbStorage_rocksDB_maxSizeInLevel1MB=256
+# dbStorage_rocksDB_logPath=
 
 
 ############################################## Metadata Services ##############################################