WIP: dump query processing performance metrics from various stages

processing/src/main/java/io/druid/query/BaseQuery.java

@@ -67,6 +67,11 @@ public static <T> int getContextUncoveredIntervalsLimit(Query<T> query, int defa
     return parseInt(query, "uncoveredIntervalsLimit", defaultValue);
   }
 
+  public static <T> int getContextDumpPerf(Query<T> query, int defaultValue)
+  {
+    return parseInt(query, QueryContextKeys.DUMP_QUERY_PERF, defaultValue);
+  }
+
   private static <T> int parseInt(Query<T> query, String key, int defaultValue)
   {
     Object val = query.getContextValue(key);

processing/src/main/java/io/druid/query/MetricsEmittingQueryRunner.java

@@ -20,8 +20,6 @@
 package io.druid.query;
 
 import com.google.common.base.Function;
-import com.google.common.base.Strings;
-import com.google.common.collect.Maps;
 import com.metamx.common.guava.Accumulator;
 import com.metamx.common.guava.Sequence;
 import com.metamx.common.guava.Yielder;
@@ -36,6 +34,8 @@
  */
 public class MetricsEmittingQueryRunner<T> implements QueryRunner<T>
 {
+  private static final String METRIC_QUERY_WAIT_TIME = "query/wait/time";
+
   private final ServiceEmitter emitter;
   private final Function<Query<T>, ServiceMetricEvent.Builder> builderFn;
   private final QueryRunner<T> queryRunner;
@@ -112,12 +112,27 @@ public <OutType> OutType accumulate(OutType outType, Accumulator<OutType, T> acc
           throw e;
         }
         finally {
+          MetricsEmittingQueryRunnerStats stats = null;
+          QueryPerfStats perfStats = (QueryPerfStats) responseContext.get(QueryPerfStats.KEY_CTX);
+          if (perfStats != null) {
+            stats = new MetricsEmittingQueryRunnerStats(userDimensions);
+            perfStats.addMetricsEmittingQueryRunnerStats(stats);
+          }
+
+
           long timeTaken = System.currentTimeMillis() - startTime;
 
           emitter.emit(builder.build(metricName, timeTaken));
+          if (stats != null) {
+            stats.addMetric(metricName, timeTaken);
+          }
 
           if (creationTime > 0) {
-            emitter.emit(builder.build("query/wait/time", startTime - creationTime));
+            long waitTime = startTime - creationTime;
+            emitter.emit(builder.build(METRIC_QUERY_WAIT_TIME, waitTime));
+            if (stats != null) {
+              stats.addMetric(METRIC_QUERY_WAIT_TIME, waitTime);
+            }
           }
         }
 

processing/src/main/java/io/druid/query/QueryContextKeys.java

@@ -24,4 +24,5 @@ public class QueryContextKeys
   public static final String PRIORITY = "priority";
   public static final String TIMEOUT = "timeout";
   public static final String CHUNK_PERIOD = "chunkPeriod";
+  public static final String DUMP_QUERY_PERF = "dumpQueryPerf";
 }

processing/src/main/java/io/druid/query/QueryPerfStats.java

@@ -0,0 +1,196 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.query;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ */
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class QueryPerfStats
+{
+  public static final String KEY_CTX = "queryPerfStats";
+
+  private long queryResultTime;
+  private long queryResultBytes;
+
+  //holders for historical/realtime stats as reported from DirectDruidClient
+  private Map<String, ServerPerfStats> serverPerf;
+
+  //holders for per segment stats on historical/realtime node
+  private List<MetricsEmittingQueryRunnerStats> metricsEmittingQueryRunnerStats;
+
+  //limit maximum number of per segment stats, to keep context size bounded
+  private final int maxNumMetricsEmittingQueryRunnerStats;
+
+  public QueryPerfStats(int maxNumMetricsEmittingQueryRunnerStats)
+  {
+    this.maxNumMetricsEmittingQueryRunnerStats = maxNumMetricsEmittingQueryRunnerStats;
+  }
+
+  @JsonProperty
+  public long getQueryResultTime()
+  {
+    return queryResultTime;
+  }
+
+  @JsonProperty
+  public long getQueryResultBytes()
+  {
+    return queryResultBytes;
+  }
+
+  @JsonProperty
+  public Map<String, ServerPerfStats> getServerPerf()
+  {
+    return serverPerf;
+  }
+
+  @JsonProperty
+  public List<MetricsEmittingQueryRunnerStats> getMetricsEmittingQueryRunnerStats()
+  {
+    return metricsEmittingQueryRunnerStats;
+  }
+
+  public void updateQueryResultTime(long queryTime) {
+    this.queryResultTime = queryTime;
+  }
+
+  public void updateQueryResultBytes(long queryBytes) {
+    this.queryResultBytes = queryBytes;
+  }
+
+  public void updateServerTime(String host, long timeTaken)
+  {
+    getServerPerfStats(host).setQueryNodeTime(timeTaken);
+  }
+
+  public void updateServerBytes(String host, long l)
+  {
+    getServerPerfStats(host).setQueryNodeBytes(l);
+  }
+
+  public void updateServerTTFB(String host, long timeTaken)
+  {
+    getServerPerfStats(host).setQueryNodeTTFB(timeTaken);
+  }
+
+  public void addMetricsEmittingQueryRunnerStats(MetricsEmittingQueryRunnerStats stats)
+  {
+    if (metricsEmittingQueryRunnerStats == null) {
+      metricsEmittingQueryRunnerStats = new ArrayList<>();
+    }
+
+    //its possible for multiple threads to end up putting more entries than maxNumMetricsEmittingQueryRunnerStats
+    //but that is OK, limit does not have to be so strict.
+    if (metricsEmittingQueryRunnerStats.size() < maxNumMetricsEmittingQueryRunnerStats) {
+      metricsEmittingQueryRunnerStats.add(stats);
+    }
+  }
+
+  private ServerPerfStats getServerPerfStats(String host)
+  {
+    //This is only called in single thread, so concurrent Map is not necessary.
+    if (serverPerf == null) {
+      serverPerf = new HashMap<>();
+    }
+
+    ServerPerfStats sps = serverPerf.get(host);
+    if (sps == null) {
+      sps = new ServerPerfStats();
+      serverPerf.put(host, sps);
+    }
+    return sps;
+  }
+}
+
+class ServerPerfStats
+{
+  private long queryNodeTTFB;
+  private long queryNodeTime;
+  private long queryNodeBytes;
+
+  @JsonProperty
+  public long getQueryNodeTTFB()
+  {
+    return queryNodeTTFB;
+  }
+
+  public void setQueryNodeTTFB(long queryNodeTTFB)
+  {
+    this.queryNodeTTFB = queryNodeTTFB;
+  }
+
+  @JsonProperty
+  public long getQueryNodeTime()
+  {
+    return queryNodeTime;
+  }
+
+  public void setQueryNodeTime(long queryNodeTime)
+  {
+    this.queryNodeTime = queryNodeTime;
+  }
+
+  @JsonProperty
+  public long getQueryNodeBytes()
+  {
+    return queryNodeBytes;
+  }
+
+  public void setQueryNodeBytes(long queryNodeBytes)
+  {
+    this.queryNodeBytes = queryNodeBytes;
+  }
+}
+
+class MetricsEmittingQueryRunnerStats
+{
+  private final Map<String, String> dimensions;
+  private final Map<String, Object> metrics = new HashMap<>();
+
+  @JsonProperty
+  public Map<String, String> getDimensions()
+  {
+    return dimensions;
+  }
+
+  @JsonProperty
+  public Map<String, Object> getMetrics()
+  {
+    return metrics;
+  }
+
+  public MetricsEmittingQueryRunnerStats(Map<String, String> dimensions)
+  {
+    this.dimensions = dimensions;
+  }
+
+  public void addMetric(String metricName, Object value)
+  {
+    metrics.put(metricName, value);
+  }
+}

server/src/main/java/io/druid/client/BrokerServerView.java

@@ -44,7 +44,6 @@
 import io.druid.timeline.VersionedIntervalTimeline;
 import io.druid.timeline.partition.PartitionChunk;
 
-import javax.annotation.Nullable;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.concurrent.ConcurrentMap;
@@ -72,6 +71,8 @@ public class BrokerServerView implements TimelineServerView
   private final ServiceEmitter emitter;
   private final Predicate<Pair<DruidServerMetadata, DataSegment>> segmentFilter;
 
+  private final DirectDruidClientConfig directDruidClientConfig;
+
   private volatile boolean initialized = false;
 
   @Inject
@@ -83,7 +84,8 @@ public BrokerServerView(
       FilteredServerInventoryView baseView,
       TierSelectorStrategy tierSelectorStrategy,
       ServiceEmitter emitter,
-      final BrokerSegmentWatcherConfig segmentWatcherConfig
+      final BrokerSegmentWatcherConfig segmentWatcherConfig,
+      final DirectDruidClientConfig directDruidClientConfig
   )
   {
     this.warehouse = warehouse;
@@ -93,6 +95,7 @@ public BrokerServerView(
     this.baseView = baseView;
     this.tierSelectorStrategy = tierSelectorStrategy;
     this.emitter = emitter;
+    this.directDruidClientConfig = directDruidClientConfig;
     this.clients = Maps.newConcurrentMap();
     this.selectors = Maps.newHashMap();
     this.timelines = Maps.newHashMap();
@@ -200,7 +203,15 @@ private QueryableDruidServer addServer(DruidServer server)
 
   private DirectDruidClient makeDirectClient(DruidServer server)
   {
-    return new DirectDruidClient(warehouse, queryWatcher, smileMapper, httpClient, server.getHost(), emitter);
+    return new DirectDruidClient(
+        warehouse,
+        queryWatcher,
+        smileMapper,
+        httpClient,
+        server.getHost(),
+        emitter,
+        directDruidClientConfig.isUseV3QueryUrl()
+    );
   }
 
   private QueryableDruidServer removeServer(DruidServer server)