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-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;