Adding `"segment/scan/active" metric for processing thread pool. (#15060

)

docs/design/extensions-contrib/dropwizard.md

@@ -137,6 +137,10 @@ Latest default metrics mapping can be found [here] (https://github.com/apache/dr
     "dimensions": [],
     "type": "gauge"
   },
+  "segment/scan/active": {
+    "dimensions": [],
+    "type": "gauge"
+  },
   "query/segmentAndCache/time": {
     "dimensions": [],
     "type": "timer",

docs/operations/metrics.md

@@ -89,6 +89,7 @@ Most metric values reset each emission period, as specified in `druid.monitoring
 |`query/segment/time`|Milliseconds taken to query individual segment. Includes time to page in the segment from disk.|`id`, `status`, `segment`, `vectorized`.|several hundred milliseconds|
 |`query/wait/time`|Milliseconds spent waiting for a segment to be scanned.|`id`, `segment`|< several hundred milliseconds|
 |`segment/scan/pending`|Number of segments in queue waiting to be scanned.||Close to 0|
+|`segment/scan/active`|Number of segments currently scanned. This metric also indicates how many threads from `druid.processing.numThreads` are currently being used.||Close to `druid.processing.numThreads`|
 |`query/segmentAndCache/time`|Milliseconds taken to query individual segment or hit the cache (if it is enabled on the Historical process).|`id`, `segment`|several hundred milliseconds|
 |`query/cpu/time`|Microseconds of CPU time taken to complete a query.|<p>Common: `dataSource`, `type`, `interval`, `hasFilters`, `duration`, `context`, `remoteAddress`, `id`.</p><p> Aggregation Queries: `numMetrics`, `numComplexMetrics`.</p><p> GroupBy: `numDimensions`.</p><p> TopN: `threshold`, `dimension`.</p>|Varies|
 |`query/count`|Total number of queries.|This metric is only available if the `QueryCountStatsMonitor` module is included.||
@@ -104,6 +105,7 @@ Most metric values reset each emission period, as specified in `druid.monitoring
 |`query/time`|Milliseconds taken to complete a query.|<p>Common: `dataSource`, `type`, `interval`, `hasFilters`, `duration`, `context`, `remoteAddress`, `id`.</p><p> Aggregation Queries: `numMetrics`, `numComplexMetrics`.</p><p> GroupBy: `numDimensions`.</p><p> TopN: `threshold`, `dimension`.</p>|< 1s|
 |`query/wait/time`|Milliseconds spent waiting for a segment to be scanned.|`id`, `segment`|several hundred milliseconds|
 |`segment/scan/pending`|Number of segments in queue waiting to be scanned.||Close to 0|
+|`segment/scan/active`|Number of segments currently scanned. This metric also indicates how many threads from `druid.processing.numThreads` are currently being used.||Close to `druid.processing.numThreads`|
 |`query/cpu/time`|Microseconds of CPU time taken to complete a query.|<p>Common: `dataSource`, `type`, `interval`, `hasFilters`, `duration`, `context`, `remoteAddress`, `id`.</p><p> Aggregation Queries: `numMetrics`, `numComplexMetrics`.</p><p> GroupBy: `numDimensions`. </p><p>TopN: `threshold`, `dimension`.</p>|Varies|
 |`query/count`|Number of total queries.|This metric is only available if the `QueryCountStatsMonitor` module is included.||
 |`query/success/count`|Number of queries successfully processed.|This metric is only available if the `QueryCountStatsMonitor` module is included.||

extensions-contrib/ambari-metrics-emitter/src/main/resources/defaultWhiteListMap.json

@@ -58,6 +58,7 @@
   "segment/loadQueue/failed": [],
   "segment/loadQueue/size": [],
   "segment/scan/pending": [],
+  "segment/scan/active": [],
   "segment/size": [
     "dataSource"
   ],

extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json

@@ -42,6 +42,10 @@
     "dimensions": [],
     "type": "gauge"
   },
+  "segment/scan/active": {
+    "dimensions": [],
+    "type": "gauge"
+  },
   "query/segmentAndCache/time": {
     "dimensions": [],
     "type": "timer",

extensions-contrib/graphite-emitter/src/main/resources/defaultWhiteListMap.json

@@ -43,6 +43,7 @@
   "segment/loadQueue/failed": [],
   "segment/loadQueue/size": [],
   "segment/scan/pending": [],
+  "segment/scan/active": [],
   "segment/size": [],
   "segment/usedPercent": []
 }

extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json

@@ -10,6 +10,7 @@
   "query/segment/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to query individual segment. Includes time to page in the segment from disk."},
   "query/wait/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds spent waiting for a segment to be scanned."},
   "segment/scan/pending" : { "dimensions" : [], "type" : "gauge", "help": "Number of segments in queue waiting to be scanned."},
+  "segment/scan/active" : { "dimensions" : [], "type" : "gauge", "help": "Number of segments currently scanned."},
   "query/segmentAndCache/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to query individual segment or hit the cache (if it is enabled on the Historical process)."},
   "query/cpu/time" : { "dimensions" : ["dataSource", "type"], "type" : "timer", "conversionFactor": "1000000", "help": "Seconds of CPU time taken to complete a query"},
 

extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json

@@ -9,6 +9,7 @@
   "query/segment/time" : { "dimensions" : [], "type" : "timer"},
   "query/wait/time" : { "dimensions" : [], "type" : "timer"},
   "segment/scan/pending" : { "dimensions" : [], "type" : "gauge"},
+  "segment/scan/active" : { "dimensions" : [], "type" : "gauge"},
   "query/segmentAndCache/time" : { "dimensions" : [], "type" : "timer" },
   "query/cpu/time" : { "dimensions" : ["dataSource", "type"], "type" : "timer" },
 

processing/src/main/java/org/apache/druid/query/MetricsEmittingQueryProcessingPool.java

@@ -41,6 +41,7 @@ public void emitMetrics(ServiceEmitter emitter, ServiceMetricEvent.Builder metri
   {
     if (delegate() instanceof PrioritizedExecutorService) {
       emitter.emit(metricBuilder.setMetric("segment/scan/pending", ((PrioritizedExecutorService) delegate()).getQueueSize()));
+      emitter.emit(metricBuilder.setMetric("segment/scan/active", ((PrioritizedExecutorService) delegate()).getActiveTasks()));
     }
   }
 

processing/src/main/java/org/apache/druid/query/PrioritizedExecutorService.java

@@ -203,6 +203,14 @@ public int getQueueSize()
   {
     return delegateQueue.size();
   }
+
+  /**
+   * Returns the approximate number of tasks being run by the thread pool currently.
+   */
+  public int getActiveTasks()
+  {
+    return threadPoolExecutor.getActiveCount();
+  }
 }
 
 class PrioritizedListenableFutureTask<V> implements RunnableFuture<V>,

processing/src/test/java/org/apache/druid/query/MetricsEmittingQueryProcessingPoolTest.java

@@ -39,6 +39,7 @@ public void testPrioritizedExecutorDelegate()
   {
     PrioritizedExecutorService service = Mockito.mock(PrioritizedExecutorService.class);
     Mockito.when(service.getQueueSize()).thenReturn(10);
+    Mockito.when(service.getActiveTasks()).thenReturn(2);
     ExecutorServiceMonitor monitor = new ExecutorServiceMonitor();
     List<Event> events = new ArrayList<>();
     MetricsEmittingQueryProcessingPool processingPool = new MetricsEmittingQueryProcessingPool(service, monitor);
@@ -53,9 +54,11 @@ public void emit(Event event)
       }
     };
     monitor.doMonitor(serviceEmitter);
-    Assert.assertEquals(1, events.size());
+    Assert.assertEquals(2, events.size());
     Assert.assertEquals(((ServiceMetricEvent) (events.get(0))).getMetric(), "segment/scan/pending");
     Assert.assertEquals(((ServiceMetricEvent) (events.get(0))).getValue(), 10);
+    Assert.assertEquals(((ServiceMetricEvent) (events.get(1))).getMetric(), "segment/scan/active");
+    Assert.assertEquals(((ServiceMetricEvent) (events.get(1))).getValue(), 2);
   }
 
   @Test