From 91c85ab8350dfc00c2bc07f0bed338ce4d87b2f6 Mon Sep 17 00:00:00 2001
From: Yiming Zang <yzang@twitter.com>
Date: Thu, 12 Jan 2017 10:49:03 -0800
Subject: [PATCH] Add aggregated read/write queue size metric

Summary: Reduce metric cardinality

Reviewers: psu, jbull, fcuny, drusek, sijieg, mpatidar, lstewart

Reviewed By: sijieg

Subscribers: #rb_ds-messaging

JIRA Issues: PUBSUB-18945

Differential Revision: https://phabricator.twitter.biz/D18693
---
 .../bookkeeper/util/OrderedSafeExecutor.java      | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/OrderedSafeExecutor.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/OrderedSafeExecutor.java
index b5a6ae5f..43f55912 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/OrderedSafeExecutor.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/util/OrderedSafeExecutor.java
@@ -224,6 +224,21 @@ public Number getSample() {
             });
         }
         // stats
+        statsLogger.registerGauge(String.format("%s-total-queue", name), new Gauge<Number>() {
+            @Override
+            public Number getDefaultValue() {
+                return 0;
+            }
+
+            @Override
+            public Number getSample() {
+                int totalSize = 0;
+                for (BlockingQueue<Runnable> queue : queues) {
+                    totalSize += queue.size();
+                }
+                return totalSize;
+            }
+        });
         this.taskExecutionStats = statsLogger.scope(name).getOpStatsLogger("task_execution");
         this.taskPendingStats = statsLogger.scope(name).getOpStatsLogger("task_queued");
         this.traceTaskExecution = traceTaskExecution;