[improve] add metrics: total_entry_log_space_bytes (apache#4507)

### Motivation
add metric `TOTAL_ENTRY_LOG_SPACE_BYTES` As a supplement to metric `ACTIVE_ENTRY_LOG_SPACE_BYTES`

This allows us to easily analyze the proportion of valid data in the cluster entry log.

### Changes
1. add metric `TOTAL_ENTRY_LOG_SPACE_BYTES` in GarbageCollectorStats.
2. Standardize totalEntryLogSize and activeEntryLogSize naming.

Signed-off-by: Zhangjian He <[email protected]>
Co-authored-by: Zhangjian He <[email protected]>

Author: ethqunzhong

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

@@ -146,6 +146,7 @@ public interface BookKeeperServerStats {
     // Compaction/Garbage Collection Related Counters
     String ACTIVE_ENTRY_LOG_COUNT = "ACTIVE_ENTRY_LOG_TOTAL";
     String ACTIVE_ENTRY_LOG_SPACE_BYTES = "ACTIVE_ENTRY_LOG_SPACE_BYTES";
+    String ENTRY_LOG_SPACE_BYTES = "ENTRY_LOG_SPACE_BYTES";
     String RECLAIMED_COMPACTION_SPACE_BYTES = "RECLAIMED_COMPACTION_SPACE_BYTES";
     String RECLAIMED_DELETION_SPACE_BYTES = "RECLAIMED_DELETION_SPACE_BYTES";
     String RECLAIM_FAILED_TO_DELETE = "RECLAIM_FAILED_TO_DELETE";

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

@@ -97,6 +97,7 @@ public class GarbageCollectorThread implements Runnable {
     // Stats loggers for garbage collection operations
     private final GarbageCollectorStats gcStats;
 
+    private volatile long activeEntryLogSize;
     private volatile long totalEntryLogSize;
     private volatile int numActiveEntryLogs;
     private volatile double entryLogCompactRatio;
@@ -175,13 +176,15 @@ public GarbageCollectorThread(ServerConfiguration conf,
         this.gcWaitTime = conf.getGcWaitTime();
 
         this.numActiveEntryLogs = 0;
+        this.activeEntryLogSize = 0L;
         this.totalEntryLogSize = 0L;
         this.entryLogCompactRatio = 0.0;
         this.currentEntryLogUsageBuckets = new int[ENTRY_LOG_USAGE_SEGMENT_COUNT];
         this.garbageCollector = new ScanAndCompareGarbageCollector(ledgerManager, ledgerStorage, conf, statsLogger);
         this.gcStats = new GarbageCollectorStats(
             statsLogger,
             () -> numActiveEntryLogs,
+            () -> activeEntryLogSize,
             () -> totalEntryLogSize,
             () -> garbageCollector.getNumActiveLedgers(),
             () -> entryLogCompactRatio,
@@ -524,6 +527,7 @@ private void doGcLedgers() {
      */
     private void doGcEntryLogs() throws EntryLogMetadataMapException {
         // Get a cumulative count, don't update until complete
+        AtomicLong activeEntryLogSizeAcc = new AtomicLong(0L);
         AtomicLong totalEntryLogSizeAcc = new AtomicLong(0L);
 
         // Loop through all of the entry logs and remove the non-active ledgers.
@@ -550,9 +554,10 @@ private void doGcEntryLogs() throws EntryLogMetadataMapException {
                 // schedule task
                 LOG.warn("Failed to remove ledger from entry-log metadata {}", entryLogId, e);
             }
-           totalEntryLogSizeAcc.getAndAdd(meta.getRemainingSize());
+            activeEntryLogSizeAcc.getAndAdd(meta.getRemainingSize());
+            totalEntryLogSizeAcc.getAndAdd(meta.getTotalSize());
         });
-
+        this.activeEntryLogSize = activeEntryLogSizeAcc.get();
         this.totalEntryLogSize = totalEntryLogSizeAcc.get();
         this.numActiveEntryLogs = entryLogMetaMap.size();
     }

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/stats/GarbageCollectorStats.java

@@ -27,6 +27,7 @@
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.COMPACT_RUNTIME;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.DELETED_LEDGER_COUNT;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.ENTRY_LOG_COMPACT_RATIO;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.ENTRY_LOG_SPACE_BYTES;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.EXTRACT_META_RUNTIME;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.GC_LEDGER_RUNTIME;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.MAJOR_COMPACTION_COUNT;
@@ -101,6 +102,11 @@ public class GarbageCollectorStats {
         help = "Current number of active entry log space bytes"
     )
     private final Gauge<Long> activeEntryLogSpaceBytesGauge;
+    @StatsDoc(
+            name = ENTRY_LOG_SPACE_BYTES,
+            help = "Current number of total entry log space bytes"
+    )
+    private final Gauge<Long> entryLogSpaceBytesGauge;
     @StatsDoc(
         name = ACTIVE_LEDGER_COUNT,
         help = "Current number of active ledgers"
@@ -133,6 +139,7 @@ public class GarbageCollectorStats {
     public GarbageCollectorStats(StatsLogger statsLogger,
                                  Supplier<Integer> activeEntryLogCountSupplier,
                                  Supplier<Long> activeEntryLogSpaceBytesSupplier,
+                                 Supplier<Long> entryLogSpaceBytesSupplier,
                                  Supplier<Integer> activeLedgerCountSupplier,
                                  Supplier<Double> entryLogCompactRatioSupplier,
                                  Supplier<int[]> usageBuckets) {
@@ -174,6 +181,18 @@ public Long getSample() {
             }
         };
         statsLogger.registerGauge(ACTIVE_ENTRY_LOG_SPACE_BYTES, activeEntryLogSpaceBytesGauge);
+        this.entryLogSpaceBytesGauge = new Gauge<Long>() {
+            @Override
+            public Long getDefaultValue() {
+                return 0L;
+            }
+
+            @Override
+            public Long getSample() {
+                return entryLogSpaceBytesSupplier.get();
+            }
+        };
+        statsLogger.registerGauge(ENTRY_LOG_SPACE_BYTES, entryLogSpaceBytesGauge);
         this.activeLedgerCountGauge = new Gauge<Integer>() {
             @Override
             public Integer getDefaultValue() {