From c85d0f62b2d2e039a1e5472b4d0ac793e8fc6f16 Mon Sep 17 00:00:00 2001
From: LinChen <1572139390@qq.com>
Date: Thu, 14 Jul 2022 23:01:35 +0800
Subject: [PATCH] add writeThreadQueuedLatency (#3363)

add writeThreadQueueStats :  calculate the time the request stays in the write threadpool queue

(cherry picked from commit 79808dccb084b057f3bb9dbc6ce300bf04443b13)
---
 .../bookkeeper/bookie/BookKeeperServerStats.java       |  1 +
 .../apache/bookkeeper/proto/PacketProcessorBase.java   |  2 ++
 .../java/org/apache/bookkeeper/proto/RequestStats.java | 10 ++++++++++
 .../apache/bookkeeper/proto/WriteEntryProcessorV3.java |  2 ++
 4 files changed, 15 insertions(+)

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
index a5ae48fd24b..ac2b5af1474 100644
--- 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
index a6ac0ad6ae8..59a053a74a3 100644
--- 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
index fd91957b233..c4879390da7 100644
--- 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
index d8ed70ce215..6669f294427 100644
--- 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.