DFS search phase per shard duration APM metric (#135652)

* Adds a per shard duration APM metric for the DFS search phase called es.search.shards.phases.query.duration.histogram

Author: chrisparrinello

docs/changelog/135652.yaml

@@ -0,0 +1,5 @@
+pr: 135652
+summary: DFS search phase per shard duration APM metric
+area: Search
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/index/search/stats/ShardSearchPhaseAPMMetrics.java

@@ -22,13 +22,20 @@
 
 public final class ShardSearchPhaseAPMMetrics implements SearchOperationListener {
 
+    public static final String DFS_SEARCH_PHASE_METRIC = "es.search.shards.phases.dfs.duration.histogram";
     public static final String QUERY_SEARCH_PHASE_METRIC = "es.search.shards.phases.query.duration.histogram";
     public static final String FETCH_SEARCH_PHASE_METRIC = "es.search.shards.phases.fetch.duration.histogram";
 
+    private final LongHistogram dfsPhaseMetric;
     private final LongHistogram queryPhaseMetric;
     private final LongHistogram fetchPhaseMetric;
 
     public ShardSearchPhaseAPMMetrics(MeterRegistry meterRegistry) {
+        this.dfsPhaseMetric = meterRegistry.registerLongHistogram(
+            DFS_SEARCH_PHASE_METRIC,
+            "DFS search phase execution times at the shard level, expressed as a histogram",
+            "ms"
+        );
         this.queryPhaseMetric = meterRegistry.registerLongHistogram(
             QUERY_SEARCH_PHASE_METRIC,
             "Query search phase execution times at the shard level, expressed as a histogram",
@@ -41,6 +48,11 @@ public ShardSearchPhaseAPMMetrics(MeterRegistry meterRegistry) {
         );
     }
 
+    @Override
+    public void onDfsPhase(SearchContext searchContext, long tookInNanos) {
+        recordPhaseLatency(dfsPhaseMetric, tookInNanos);
+    }
+
     @Override
     public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
         SearchExecutionContext searchExecutionContext = searchContext.getSearchExecutionContext();
@@ -55,6 +67,10 @@ public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
         recordPhaseLatency(fetchPhaseMetric, tookInNanos, searchContext.request(), rangeTimestampFrom);
     }
 
+    private static void recordPhaseLatency(LongHistogram histogramMetric, long tookInNanos) {
+        histogramMetric.record(TimeUnit.NANOSECONDS.toMillis(tookInNanos));
+    }
+
     private static void recordPhaseLatency(
         LongHistogram histogramMetric,
         long tookInNanos,

server/src/test/java/org/elasticsearch/search/TelemetryMetrics/ShardSearchPhaseAPMMetricsTests.java

@@ -34,6 +34,7 @@
 
 import static org.elasticsearch.action.support.WriteRequest.RefreshPolicy.IMMEDIATE;
 import static org.elasticsearch.index.query.QueryBuilders.simpleQueryStringQuery;
+import static org.elasticsearch.index.search.stats.ShardSearchPhaseAPMMetrics.DFS_SEARCH_PHASE_METRIC;
 import static org.elasticsearch.index.search.stats.ShardSearchPhaseAPMMetrics.FETCH_SEARCH_PHASE_METRIC;
 import static org.elasticsearch.index.search.stats.ShardSearchPhaseAPMMetrics.QUERY_SEARCH_PHASE_METRIC;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
@@ -92,6 +93,8 @@ public void testMetricsDfsQueryThenFetch() {
             client().prepareSearch(indexName).setSearchType(SearchType.DFS_QUERY_THEN_FETCH).setQuery(simpleQueryStringQuery("doc1")),
             "1"
         );
+        final List<Measurement> dfsMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(DFS_SEARCH_PHASE_METRIC);
+        assertEquals(num_primaries, dfsMeasurements.size());
         final List<Measurement> queryMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(QUERY_SEARCH_PHASE_METRIC);
         assertEquals(num_primaries, queryMeasurements.size());
         final List<Measurement> fetchMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(FETCH_SEARCH_PHASE_METRIC);
@@ -106,6 +109,8 @@ public void testMetricsDfsQueryThenFetchSystem() {
                 .setQuery(simpleQueryStringQuery("doc1")),
             "1"
         );
+        final List<Measurement> dfsMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(DFS_SEARCH_PHASE_METRIC);
+        assertEquals(0, dfsMeasurements.size()); // DFS phase not done for index with single shard
         final List<Measurement> queryMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(QUERY_SEARCH_PHASE_METRIC);
         assertEquals(1, queryMeasurements.size());
         final List<Measurement> fetchMeasurements = getTestTelemetryPlugin().getLongHistogramMeasurement(FETCH_SEARCH_PHASE_METRIC);