Add trigger entry location index rocksDB compact REST API

bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpRouter.java

@@ -55,6 +55,7 @@ public abstract class HttpRouter<Handler> {
     public static final String BOOKIE_IS_READY              = "/api/v1/bookie/is_ready";
     public static final String BOOKIE_INFO                  = "/api/v1/bookie/info";
     public static final String CLUSTER_INFO                  = "/api/v1/bookie/cluster_info";
+    public static final String ENTRY_LOCATION_COMPACT       = "/api/v1/bookie/entry_location_compact";
     // autorecovery
     public static final String AUTORECOVERY_STATUS          = "/api/v1/autorecovery/status";
     public static final String RECOVERY_BOOKIE              = "/api/v1/autorecovery/bookie";
@@ -97,6 +98,8 @@ public HttpRouter(AbstractHttpHandlerFactory<Handler> handlerFactory) {
             handlerFactory.newHandler(HttpServer.ApiType.SUSPEND_GC_COMPACTION));
         this.endpointHandlers.put(RESUME_GC_COMPACTION,
             handlerFactory.newHandler(HttpServer.ApiType.RESUME_GC_COMPACTION));
+        this.endpointHandlers.put(ENTRY_LOCATION_COMPACT,
+                handlerFactory.newHandler(HttpServer.ApiType.TRIGGER_ENTRY_LOCATION_COMPACT));
 
         // autorecovery
         this.endpointHandlers.put(AUTORECOVERY_STATUS, handlerFactory

bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServer.java

@@ -36,6 +36,7 @@ enum StatusCode {
         BAD_REQUEST(400),
         FORBIDDEN(403),
         NOT_FOUND(404),
+        METHOD_NOT_ALLOWED(405),
         INTERNAL_ERROR(500),
         SERVICE_UNAVAILABLE(503);
 
@@ -89,6 +90,7 @@ enum ApiType {
         CLUSTER_INFO,
         RESUME_GC_COMPACTION,
         SUSPEND_GC_COMPACTION,
+        TRIGGER_ENTRY_LOCATION_COMPACT,
         // autorecovery
         AUTORECOVERY_STATUS,
         RECOVERY_BOOKIE,

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerStorage.java

@@ -29,6 +29,7 @@
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.PrimitiveIterator;
 import org.apache.bookkeeper.bookie.CheckpointSource.Checkpoint;
@@ -259,6 +260,26 @@ default boolean isMinorGcSuspended() {
         return false;
     }
 
+    default void entryLocationCompact() {
+        return;
+    }
+
+    default void entryLocationCompact(List<String> locations) {
+        return;
+    }
+
+    default boolean isEntryLocationCompacting() {
+        return false;
+    }
+
+    default Map<String, Boolean> isEntryLocationCompacting(List<String> locations) {
+        return Collections.emptyMap();
+    }
+
+    default List<String> getEntryLocationDBPath() {
+        return Collections.emptyList();
+    }
+
     /**
      * Class for describing location of a generic inconsistency.  Implementations should
      * ensure that detail is populated with an exception which adequately describes the

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

@@ -26,6 +26,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.util.concurrent.DefaultThreadFactory;
@@ -34,7 +35,9 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.PrimitiveIterator.OfLong;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -542,6 +545,47 @@ public boolean isMinorGcSuspended() {
         return ledgerStorageList.stream().allMatch(SingleDirectoryDbLedgerStorage::isMinorGcSuspended);
     }
 
+    @Override
+    public void entryLocationCompact() {
+        ledgerStorageList.forEach(SingleDirectoryDbLedgerStorage::entryLocationCompact);
+    }
+
+    @Override
+    public void entryLocationCompact(List<String> locations) {
+        for (SingleDirectoryDbLedgerStorage ledgerStorage : ledgerStorageList) {
+            String entryLocation = ledgerStorage.getEntryLocationDBPath().get(0);
+            if (locations.contains(entryLocation)) {
+                ledgerStorage.entryLocationCompact();
+            }
+        }
+    }
+
+    @Override
+    public boolean isEntryLocationCompacting() {
+        return ledgerStorageList.stream().anyMatch(SingleDirectoryDbLedgerStorage::isEntryLocationCompacting);
+    }
+
+    @Override
+    public Map<String, Boolean> isEntryLocationCompacting(List<String> locations) {
+        HashMap<String, Boolean> isCompacting = Maps.newHashMap();
+        for (SingleDirectoryDbLedgerStorage ledgerStorage : ledgerStorageList) {
+            String entryLocation = ledgerStorage.getEntryLocationDBPath().get(0);
+            if (locations.contains(entryLocation)) {
+                isCompacting.put(entryLocation, ledgerStorage.isEntryLocationCompacting());
+            }
+        }
+        return isCompacting;
+    }
+
+    @Override
+    public List<String> getEntryLocationDBPath() {
+        List<String> allEntryLocationDBPath = Lists.newArrayList();
+        for (SingleDirectoryDbLedgerStorage ledgerStorage : ledgerStorageList) {
+            allEntryLocationDBPath.addAll(ledgerStorage.getEntryLocationDBPath());
+        }
+        return allEntryLocationDBPath;
+    }
+
     @Override
     public List<GarbageCollectionStatus> getGarbageCollectionStatus() {
         return ledgerStorageList.stream()

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

@@ -48,6 +48,7 @@ public class EntryLocationIndex implements Closeable {
     private final KeyValueStorage locationsDb;
     private final ConcurrentLongHashSet deletedLedgers = ConcurrentLongHashSet.newBuilder().build();
     private final EntryLocationIndexStats stats;
+    private boolean isCompacting;
 
     public EntryLocationIndex(ServerConfiguration conf, KeyValueStorageFactory storageFactory, String basePath,
             StatsLogger stats) throws IOException {
@@ -189,6 +190,23 @@ public void delete(long ledgerId) throws IOException {
         deletedLedgers.add(ledgerId);
     }
 
+    public String getEntryLocationDBPath() {
+        return locationsDb.getDBPath();
+    }
+
+    public void compact() throws IOException {
+        try {
+            isCompacting = true;
+            locationsDb.compact();
+        } finally {
+            isCompacting = false;
+        }
+    }
+
+    public boolean isCompacting() {
+        return isCompacting;
+    }
+
     public void removeOffsetFromDeletedLedgers() throws IOException {
         LongPairWrapper firstKeyWrapper = LongPairWrapper.get(-1, -1);
         LongPairWrapper lastKeyWrapper = LongPairWrapper.get(-1, -1);

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

@@ -106,6 +106,16 @@ public interface KeyValueStorage extends Closeable {
      */
     default void compact(byte[] firstKey, byte[] lastKey) throws IOException {}
 
+    /**
+     * Compact storage full range.
+     */
+    default void compact() throws IOException {}
+
+    /**
+     * Get storage path.
+     */
+    String getDBPath();
+
     /**
      * Get an iterator over to scan sequentially through all the keys in the
      * database.

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

@@ -51,6 +51,7 @@
 import org.rocksdb.Env;
 import org.rocksdb.InfoLogLevel;
 import org.rocksdb.LRUCache;
+import org.rocksdb.LiveFileMetaData;
 import org.rocksdb.Options;
 import org.rocksdb.OptionsUtil;
 import org.rocksdb.ReadOptions;
@@ -84,6 +85,8 @@ public class KeyValueStorageRocksDB implements KeyValueStorage {
     private final ReadOptions optionDontCache;
     private final WriteBatch emptyBatch;
 
+    private String dbPath;
+
     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";
@@ -158,13 +161,13 @@ private RocksDB initializeRocksDBWithConfFile(String basePath, String subPath, D
                 log.info("RocksDB<{}> log path: {}", subPath, logPathSetting);
                 dbOptions.setDbLogDir(logPathSetting.toString());
             }
-            String path = FileSystems.getDefault().getPath(basePath, subPath).toFile().toString();
+            this.dbPath = FileSystems.getDefault().getPath(basePath, subPath).toFile().toString();
             this.options = dbOptions;
             this.columnFamilyDescriptors = cfDescs;
             if (readOnly) {
-                return RocksDB.openReadOnly(dbOptions, path, cfDescs, cfHandles);
+                return RocksDB.openReadOnly(dbOptions, dbPath, cfDescs, cfHandles);
             } else {
-                return RocksDB.open(dbOptions, path, cfDescs, cfHandles);
+                return RocksDB.open(dbOptions, dbPath, cfDescs, cfHandles);
             }
         } catch (RocksDBException e) {
             throw new IOException("Error open RocksDB database", e);
@@ -241,7 +244,7 @@ private RocksDB initializeRocksDBWithBookieConf(String basePath, String subPath,
             log.info("RocksDB<{}> log path: {}", subPath, logPathSetting);
             options.setDbLogDir(logPathSetting.toString());
         }
-        String path = FileSystems.getDefault().getPath(basePath, subPath).toFile().toString();
+        this.dbPath = FileSystems.getDefault().getPath(basePath, subPath).toFile().toString();
 
         // Configure log level
         String logLevel = conf.getString(ROCKSDB_LOG_LEVEL, "info");
@@ -268,9 +271,9 @@ private RocksDB initializeRocksDBWithBookieConf(String basePath, String subPath,
         this.options = options;
         try {
             if (readOnly) {
-                return RocksDB.openReadOnly(options, path);
+                return RocksDB.openReadOnly(options, dbPath);
             } else {
-                return RocksDB.open(options, path);
+                return RocksDB.open(options, dbPath);
             }
         } catch (RocksDBException e) {
             throw new IOException("Error open RocksDB database", e);
@@ -365,6 +368,11 @@ public void delete(byte[] key) throws IOException {
         }
     }
 
+    @Override
+    public String getDBPath() {
+        return dbPath;
+    }
+
     @Override
     public void compact(byte[] firstKey, byte[] lastKey) throws IOException {
         try {
@@ -374,6 +382,36 @@ public void compact(byte[] firstKey, byte[] lastKey) throws IOException {
         }
     }
 
+    @Override
+    public void compact() throws IOException {
+        try {
+            final long start = System.currentTimeMillis();
+            final int oriRocksDBFileCount = db.getLiveFilesMetaData().size();
+            final long oriRocksDBSize = getRocksDBSize();
+            log.info("Starting RocksDB {} compact, current RocksDB hold {} files and {} Bytes.",
+                    db.getName(), oriRocksDBFileCount, oriRocksDBSize);
+
+            db.compactRange();
+
+            final long end = System.currentTimeMillis();
+            final int rocksDBFileCount = db.getLiveFilesMetaData().size();
+            final long rocksDBSize = getRocksDBSize();
+            log.info("RocksDB {} compact finished {} ms, space reduced {} Bytes, current hold {} files and {} Bytes.",
+                    db.getName(), end - start, oriRocksDBSize - rocksDBSize, rocksDBFileCount, rocksDBSize);
+        } catch (RocksDBException e) {
+            throw new IOException("Error in RocksDB compact", e);
+        }
+    }
+
+    private long getRocksDBSize() {
+        List<LiveFileMetaData> liveFilesMetaData = db.getLiveFilesMetaData();
+        long rocksDBFileSize = 0L;
+        for (LiveFileMetaData fileMetaData : liveFilesMetaData) {
+            rocksDBFileSize += fileMetaData.size();
+        }
+        return rocksDBFileSize;
+    }
+
     @Override
     public void sync() throws IOException {
         try {

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

@@ -298,6 +298,34 @@ public boolean isMinorGcSuspended() {
         return gcThread.isMinorGcSuspend();
     }
 
+    @Override
+    public void entryLocationCompact() {
+        if (entryLocationIndex.isCompacting()) {
+            // RocksDB already running compact.
+            return;
+        }
+        cleanupExecutor.execute(() -> {
+            // There can only be one single cleanup task running because the cleanupExecutor
+            // is single-threaded
+            try {
+                log.info("Trigger entry location index RocksDB compact.");
+                entryLocationIndex.compact();
+            } catch (Throwable t) {
+                log.warn("Failed to trigger entry location index RocksDB compact", t);
+            }
+        });
+    }
+
+    @Override
+    public boolean isEntryLocationCompacting() {
+        return entryLocationIndex.isCompacting();
+    }
+
+    @Override
+    public List<String> getEntryLocationDBPath() {
+        return Lists.newArrayList(entryLocationIndex.getEntryLocationDBPath());
+    }
+
     @Override
     public void shutdown() throws InterruptedException {
         try {

bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/BKHttpServiceProvider.java

@@ -64,6 +64,7 @@
 import org.apache.bookkeeper.server.http.service.SuspendCompactionService;
 import org.apache.bookkeeper.server.http.service.TriggerAuditService;
 import org.apache.bookkeeper.server.http.service.TriggerGCService;
+import org.apache.bookkeeper.server.http.service.TriggerLocationCompactService;
 import org.apache.bookkeeper.server.http.service.WhoIsAuditorService;
 import org.apache.bookkeeper.stats.StatsProvider;
 import org.apache.zookeeper.KeeperException;
@@ -235,6 +236,8 @@ public HttpEndpointService provideHttpEndpointService(ApiType type) {
                 return new SuspendCompactionService(bookieServer);
             case RESUME_GC_COMPACTION:
                 return new ResumeCompactionService(bookieServer);
+            case TRIGGER_ENTRY_LOCATION_COMPACT:
+                return new TriggerLocationCompactService(bookieServer);
 
             // autorecovery
             case AUTORECOVERY_STATUS: