add writeThreadQueuedLatency (apache#3363)

add writeThreadQueueStats : calculate the time the request stays in the write threadpool queue (cherry picked from commit 79808dc)
datastax · Nov 7, 2022 · c85d0f6 · c85d0f6
1 parent cbaf9f8
commit c85d0f6
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 0 deletions.
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookKeeperServerStats.java
@@ -46,6 +46,7 @@ public interface BookKeeperServerStats {
     // Stats
     String ADD_ENTRY_REQUEST = "ADD_ENTRY_REQUEST";
     String ADD_ENTRY = "ADD_ENTRY";
+    String WRITE_THREAD_QUEUED_LATENCY = "WRITE_THREAD_QUEUED_LATENCY";
     String FORCE_LEDGER_REQUEST = "FORCE_LEDGER_REQUEST";
     String FORCE_LEDGER = "FORCE_LEDGER";
     String READ_ENTRY_REQUEST = "READ_ENTRY_REQUEST";

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PacketProcessorBase.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PacketProcessorBase.java
@@ -158,6 +158,8 @@ protected void sendResponseAndWait(int rc, Object response, OpStatsLogger statsL
 
     @Override
     public void safeRun() {
+        requestProcessor.getRequestStats().getWriteThreadQueuedLatency()
+                .registerSuccessfulEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
         if (!isVersionCompatible()) {
             sendResponse(BookieProtocol.EBADVERSION,
                          ResponseBuilder.buildErrorResponse(BookieProtocol.EBADVERSION, request),

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/RequestStats.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/RequestStats.java
@@ -51,6 +51,7 @@
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.SERVER_SCOPE;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.WRITE_LAC;
 import static org.apache.bookkeeper.bookie.BookKeeperServerStats.WRITE_LAC_REQUEST;
+import static org.apache.bookkeeper.bookie.BookKeeperServerStats.WRITE_THREAD_QUEUED_LATENCY;
 
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -90,6 +91,14 @@ public class RequestStats {
         parent = ADD_ENTRY_REQUEST
     )
     private final OpStatsLogger addEntryStats;
+
+    @StatsDoc(
+            name = WRITE_THREAD_QUEUED_LATENCY,
+            help = "operation stats of enqueuing requests to write threadpool queue",
+            parent = ADD_ENTRY_REQUEST
+    )
+    private final OpStatsLogger writeThreadQueuedLatency;
+
     @StatsDoc(
         name = READ_ENTRY_REQUEST,
         help = "request stats of ReadEntry on a bookie"
@@ -226,6 +235,7 @@ public class RequestStats {
 
     public RequestStats(StatsLogger statsLogger) {
         this.addEntryStats = statsLogger.getOpStatsLogger(ADD_ENTRY);
+        this.writeThreadQueuedLatency = statsLogger.getOpStatsLogger(WRITE_THREAD_QUEUED_LATENCY);
         this.addRequestStats = statsLogger.getOpStatsLogger(ADD_ENTRY_REQUEST);
         this.readEntryStats = statsLogger.getOpStatsLogger(READ_ENTRY);
         this.forceLedgerStats = statsLogger.getOpStatsLogger(FORCE_LEDGER);

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/WriteEntryProcessorV3.java
@@ -162,6 +162,8 @@ public void writeComplete(int rc, long ledgerId, long entryId,
 
     @Override
     public void safeRun() {
+        requestProcessor.getRequestStats().getWriteThreadQueuedLatency()
+                .registerSuccessfulEvent(MathUtils.elapsedNanos(enqueueNanos), TimeUnit.NANOSECONDS);
         AddResponse addResponse = getAddResponse();
         if (null != addResponse) {
             // This means there was an error and we should send this back.