Add monitor to find length of queue at Routers

Author: ravisharm

docs/operations/metrics.md

@@ -46,6 +46,7 @@ Most metric values reset each emission period, as specified in `druid.monitoring
 |Metric|Description|Dimensions|Normal value|
 |------|-----------|----------|------------|
 |`query/time`|Milliseconds taken to complete a query.|Native Query: `dataSource`, `type`, `interval`, `hasFilters`, `duration`, `context`, `remoteAddress`, `id`.|< 1s|
+|`router/http/numRequestsQueued`|Total number of outbound HTTP requests currently queued across all destinations on the Router's HTTP client.|None.|0; sustained high values indicate backpressure toward downstream Brokers.|
 
 ### Broker
 

server/src/main/java/org/apache/druid/server/metrics/RouterHttpClientMonitor.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.metrics;
+
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.java.util.metrics.AbstractMonitor;
+import org.apache.druid.server.router.Router;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.HttpDestination;
+import org.eclipse.jetty.client.api.Destination;
+
+/**
+ * Monitor that emits the total number of outbound HTTP requests currently queued
+ * across all destinations on the Router's Jetty HttpClient.
+ */
+public class RouterHttpClientMonitor extends AbstractMonitor
+{
+  private final Provider<HttpClient> httpClientProvider;
+
+  @Inject
+  public RouterHttpClientMonitor(@Router Provider<HttpClient> httpClientProvider)
+  {
+    this.httpClientProvider = httpClientProvider;
+  }
+
+  @Override
+  public boolean doMonitor(ServiceEmitter emitter)
+  {
+    final HttpClient httpClient = httpClientProvider.get();
+
+    int totalQueuedRequests = 0;
+    for (Destination destination : httpClient.getDestinations()) {
+      if (destination instanceof HttpDestination) {
+        totalQueuedRequests += ((HttpDestination) destination).getQueuedRequestCount();
+      }
+    }
+
+    emitter.emit(new ServiceMetricEvent.Builder().setMetric("router/http/numRequestsQueued", totalQueuedRequests));
+
+    return true;
+  }
+}

server/src/test/java/org/apache/druid/server/metrics/RouterHttpClientMonitorTest.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.metrics;
+
+import com.google.inject.Provider;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.HttpDestination;
+import org.eclipse.jetty.client.api.Destination;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class RouterHttpClientMonitorTest
+{
+  private HttpClient httpClient;
+  private RouterHttpClientMonitor monitor;
+
+  @Before
+  public void setUp()
+  {
+    httpClient = Mockito.mock(HttpClient.class);
+    Provider<HttpClient> httpClientProvider = () -> httpClient;
+    monitor = new RouterHttpClientMonitor(httpClientProvider);
+  }
+
+  @Test
+  public void testDoMonitorReturnsTrue()
+  {
+    Mockito.when(httpClient.getDestinations()).thenReturn(Collections.emptyList());
+    Assert.assertTrue(monitor.doMonitor(new StubServiceEmitter("router", "localhost")));
+  }
+
+  @Test
+  public void testNoDestinationsEmitsZero()
+  {
+    Mockito.when(httpClient.getDestinations()).thenReturn(Collections.emptyList());
+    final StubServiceEmitter emitter = new StubServiceEmitter("router", "localhost");
+    monitor.doMonitor(emitter);
+
+    Assert.assertEquals(1, emitter.getEvents().size());
+    Map<String, Object> event = emitter.getEvents().get(0).toMap();
+    Assert.assertEquals("router/http/numRequestsQueued", event.get("metric"));
+    Assert.assertEquals(0, event.get("value"));
+  }
+
+  @Test
+  public void testMultipleDestinationsQueueCountSummed()
+  {
+    Mockito.when(httpClient.getDestinations()).thenReturn(List.of(mockDest(3), mockDest(5)));
+    final StubServiceEmitter emitter = new StubServiceEmitter("router", "localhost");
+    monitor.doMonitor(emitter);
+
+    Assert.assertEquals(1, emitter.getEvents().size());
+    Assert.assertEquals(8, emitter.getEvents().get(0).toMap().get("value"));
+  }
+
+  @Test
+  public void testNonHttpDestinationIsSkipped()
+  {
+    Destination plainDest = Mockito.mock(Destination.class);
+    Mockito.when(httpClient.getDestinations()).thenReturn(List.of(plainDest, mockDest(4)));
+    final StubServiceEmitter emitter = new StubServiceEmitter("router", "localhost");
+    monitor.doMonitor(emitter);
+
+    Assert.assertEquals(1, emitter.getEvents().size());
+    Assert.assertEquals(4, emitter.getEvents().get(0).toMap().get("value"));
+  }
+
+  private static HttpDestination mockDest(int queuedCount)
+  {
+    HttpDestination dest = Mockito.mock(HttpDestination.class);
+    Mockito.when(dest.getQueuedRequestCount()).thenReturn(queuedCount);
+    return dest;
+  }
+}

services/src/main/java/org/apache/druid/cli/CliRouter.java

@@ -46,7 +46,9 @@
 import org.apache.druid.server.http.RouterResource;
 import org.apache.druid.server.http.SelfDiscoveryResource;
 import org.apache.druid.server.initialization.jetty.JettyServerInitializer;
+import org.apache.druid.server.metrics.MetricsModule;
 import org.apache.druid.server.metrics.QueryCountStatsProvider;
+import org.apache.druid.server.metrics.RouterHttpClientMonitor;
 import org.apache.druid.server.router.AvaticaConnectionBalancer;
 import org.apache.druid.server.router.CoordinatorRuleManager;
 import org.apache.druid.server.router.ManagementProxyConfig;
@@ -114,6 +116,7 @@ protected List<? extends Module> getModules()
                 .in(LazySingleton.class);
 
           binder.bind(QueryCountStatsProvider.class).to(AsyncQueryForwardingServlet.class).in(LazySingleton.class);
+          MetricsModule.register(binder, RouterHttpClientMonitor.class);
           binder.bind(JettyServerInitializer.class).to(RouterJettyServerInitializer.class).in(LazySingleton.class);
 
           Jerseys.addResource(binder, RouterResource.class);