Metrics: add journalCbQueueLatency (#3364)

Author: lordcheng10

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

@@ -162,6 +162,7 @@ public interface BookKeeperServerStats {
     String INDEX_INMEM_ILLEGAL_STATE_DELETE = "INDEX_INMEM_ILLEGAL_STATE_DELETE";
     String JOURNAL_FORCE_WRITE_QUEUE_SIZE = "JOURNAL_FORCE_WRITE_QUEUE_SIZE";
     String JOURNAL_CB_QUEUE_SIZE = "JOURNAL_CB_QUEUE_SIZE";
+    String JOURNAL_CB_QUEUED_LATENCY = "JOURNAL_CB_QUEUED_LATENCY";
     String JOURNAL_NUM_FORCE_WRITES = "JOURNAL_NUM_FORCE_WRITES";
     String JOURNAL_NUM_FLUSH_EMPTY_QUEUE = "JOURNAL_NUM_FLUSH_EMPTY_QUEUE";
     String JOURNAL_NUM_FLUSH_MAX_OUTSTANDING_BYTES = "JOURNAL_NUM_FLUSH_MAX_OUTSTANDING_BYTES";

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

@@ -303,15 +303,18 @@ static class QueueEntry implements Runnable {
         WriteCallback cb;
         Object ctx;
         long enqueueTime;
+        long enqueueCbThreadPooleQueueTime;
         boolean ackBeforeSync;
 
         OpStatsLogger journalAddEntryStats;
+        OpStatsLogger journalCbQueuedLatency;
         Counter journalCbQueueSize;
         Counter callbackTime;
 
         static QueueEntry create(ByteBuf entry, boolean ackBeforeSync, long ledgerId, long entryId,
                 WriteCallback cb, Object ctx, long enqueueTime, OpStatsLogger journalAddEntryStats,
-                Counter journalCbQueueSize, Counter callbackTime) {
+                Counter journalCbQueueSize, OpStatsLogger journalCbQueuedLatency,
+                Counter callbackTime) {
             QueueEntry qe = RECYCLER.get();
             qe.entry = entry;
             qe.ackBeforeSync = ackBeforeSync;
@@ -321,13 +324,20 @@ static QueueEntry create(ByteBuf entry, boolean ackBeforeSync, long ledgerId, lo
             qe.entryId = entryId;
             qe.enqueueTime = enqueueTime;
             qe.journalAddEntryStats = journalAddEntryStats;
+            qe.journalCbQueuedLatency = journalCbQueuedLatency;
             qe.journalCbQueueSize = journalCbQueueSize;
             qe.callbackTime = callbackTime;
             return qe;
         }
 
+        public void setEnqueueCbThreadPooleQueueTime(long enqueueCbThreadPooleQueueTime) {
+            this.enqueueCbThreadPooleQueueTime = enqueueCbThreadPooleQueueTime;
+        }
+
         @Override
         public void run() {
+            journalCbQueuedLatency.registerSuccessfulEvent(
+                    MathUtils.elapsedNanos(enqueueCbThreadPooleQueueTime), TimeUnit.NANOSECONDS);
             long startTime = System.nanoTime();
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Acknowledge Ledger: {}, Entry: {}", ledgerId, entryId);
@@ -392,6 +402,7 @@ public int process(boolean shouldForceWrite) throws IOException {
                 for (int i = 0; i < forceWriteWaiters.size(); i++) {
                     QueueEntry qe = forceWriteWaiters.get(i);
                     if (qe != null) {
+                        qe.setEnqueueCbThreadPooleQueueTime(MathUtils.nowInNano());
                         cbThreadPool.execute(qe);
                     }
                 }
@@ -943,6 +954,7 @@ public void logAddEntry(long ledgerId, long entryId, ByteBuf entry,
                 entry, ackBeforeSync,  ledgerId, entryId, cb, ctx, MathUtils.nowInNano(),
                 journalStats.getJournalAddEntryStats(),
                 journalStats.getJournalCbQueueSize(),
+                journalStats.getJournalCbQueuedLatency(),
                 callbackTime));
     }
 
@@ -952,6 +964,7 @@ void forceLedger(long ledgerId, WriteCallback cb, Object ctx) {
                 BookieImpl.METAENTRY_ID_FORCE_LEDGER, cb, ctx, MathUtils.nowInNano(),
                 journalStats.getJournalForceLedgerStats(),
                 journalStats.getJournalCbQueueSize(),
+                journalStats.getJournalCbQueuedLatency(),
                 callbackTime));
         // Increment afterwards because the add operation could fail.
         journalStats.getJournalQueueSize().inc();
@@ -1118,6 +1131,7 @@ journalFormatVersionToWrite, getBufferedChannelBuilder(),
                                 if (entry != null && (!syncData || entry.ackBeforeSync)) {
                                     toFlush.set(i, null);
                                     numEntriesToFlush--;
+                                    entry.setEnqueueCbThreadPooleQueueTime(MathUtils.nowInNano());
                                     cbThreadPool.execute(entry);
                                 }
                             }

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

@@ -23,6 +23,7 @@
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.CATEGORY_SERVER;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.FORCE_LEDGER;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_ADD_ENTRY;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_CB_QUEUED_LATENCY;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_CB_QUEUE_SIZE;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_CREATION_LATENCY;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.JOURNAL_FLUSH_LATENCY;
@@ -153,6 +154,13 @@ public class JournalStats {
         help = "The journal callback queue size"
     )
     private final Counter journalCbQueueSize;
+
+    @StatsDoc(
+            name = JOURNAL_CB_QUEUED_LATENCY,
+            help = "The journal callback queued latency"
+    )
+    private final OpStatsLogger journalCbQueuedLatency;
+
     @StatsDoc(
         name = JOURNAL_NUM_FLUSH_MAX_WAIT,
         help = "The number of journal flushes triggered by MAX_WAIT time"
@@ -203,6 +211,7 @@ public JournalStats(StatsLogger statsLogger, final long maxJournalMemoryBytes,
         journalQueueSize = statsLogger.getCounter(BookKeeperServerStats.JOURNAL_QUEUE_SIZE);
         forceWriteQueueSize = statsLogger.getCounter(BookKeeperServerStats.JOURNAL_FORCE_WRITE_QUEUE_SIZE);
         journalCbQueueSize = statsLogger.getCounter(BookKeeperServerStats.JOURNAL_CB_QUEUE_SIZE);
+        journalCbQueuedLatency = statsLogger.getOpStatsLogger(BookKeeperServerStats.JOURNAL_CB_QUEUED_LATENCY);
         flushMaxWaitCounter = statsLogger.getCounter(BookKeeperServerStats.JOURNAL_NUM_FLUSH_MAX_WAIT);
         flushMaxOutstandingBytesCounter =
                 statsLogger.getCounter(BookKeeperServerStats.JOURNAL_NUM_FLUSH_MAX_OUTSTANDING_BYTES);