Fetch search phase coordinator duration APM metric (#136547)

Adds the following APM metric to track the duration of the fetch phase at the coordinator node:
- es.search_response.took_durations.fetch.histogram

Author: chrisparrinello

docs/changelog/136547.yaml

@@ -0,0 +1,5 @@
+pr: 136547
+summary: Fetch search phase coordinator duration APM metric
+area: Search
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java

@@ -92,6 +92,7 @@ public void onFailure(Exception e) {
 
     private void innerRun() throws Exception {
         assert this.reducedQueryPhase == null ^ this.resultConsumer == null;
+        long phaseStartTimeInNanos = System.nanoTime();
         // depending on whether we executed the RankFeaturePhase we may or may not have the reduced query result computed already
         final var reducedQueryPhase = this.reducedQueryPhase == null ? resultConsumer.reduce() : this.reducedQueryPhase;
         final int numShards = context.getNumShards();
@@ -104,22 +105,27 @@ private void innerRun() throws Exception {
         if (queryAndFetchOptimization) {
             assert assertConsistentWithQueryAndFetchOptimization();
             // query AND fetch optimization
-            moveToNextPhase(searchPhaseShardResults, reducedQueryPhase);
+            moveToNextPhase(searchPhaseShardResults, reducedQueryPhase, phaseStartTimeInNanos);
         } else {
             ScoreDoc[] scoreDocs = reducedQueryPhase.sortedTopDocs().scoreDocs();
             // no docs to fetch -- sidestep everything and return
             if (scoreDocs.length == 0) {
                 // we have to release contexts here to free up resources
                 searchPhaseShardResults.asList()
                     .forEach(searchPhaseShardResult -> releaseIrrelevantSearchContext(searchPhaseShardResult, context));
-                moveToNextPhase(new AtomicArray<>(0), reducedQueryPhase);
+                moveToNextPhase(new AtomicArray<>(0), reducedQueryPhase, phaseStartTimeInNanos);
             } else {
-                innerRunFetch(scoreDocs, numShards, reducedQueryPhase);
+                innerRunFetch(scoreDocs, numShards, reducedQueryPhase, phaseStartTimeInNanos);
             }
         }
     }
 
-    private void innerRunFetch(ScoreDoc[] scoreDocs, int numShards, SearchPhaseController.ReducedQueryPhase reducedQueryPhase) {
+    private void innerRunFetch(
+        ScoreDoc[] scoreDocs,
+        int numShards,
+        SearchPhaseController.ReducedQueryPhase reducedQueryPhase,
+        long phaseStartTimeInNanos
+    ) {
         ArraySearchPhaseResults<FetchSearchResult> fetchResults = new ArraySearchPhaseResults<>(numShards);
         final List<Map<Integer, RankDoc>> rankDocsPerShard = false == shouldExplainRankScores(context.getRequest())
             ? null
@@ -133,7 +139,7 @@ private void innerRunFetch(ScoreDoc[] scoreDocs, int numShards, SearchPhaseContr
             docIdsToLoad.length, // we count down every shard in the result no matter if we got any results or not
             () -> {
                 try (fetchResults) {
-                    moveToNextPhase(fetchResults.getAtomicArray(), reducedQueryPhase);
+                    moveToNextPhase(fetchResults.getAtomicArray(), reducedQueryPhase, phaseStartTimeInNanos);
                 }
             },
             context
@@ -258,8 +264,10 @@ public void onFailure(Exception e) {
 
     private void moveToNextPhase(
         AtomicArray<? extends SearchPhaseResult> fetchResultsArr,
-        SearchPhaseController.ReducedQueryPhase reducedQueryPhase
+        SearchPhaseController.ReducedQueryPhase reducedQueryPhase,
+        long phaseStartTimeInNanos
     ) {
+        context.getSearchResponseMetrics().recordSearchPhaseDuration(getName(), System.nanoTime() - phaseStartTimeInNanos);
         context.executeNextPhase(NAME, () -> {
             var resp = SearchPhaseController.merge(context.getRequest().scroll() != null, reducedQueryPhase, fetchResultsArr);
             context.addReleasable(resp);

server/src/main/java/org/elasticsearch/rest/action/search/SearchResponseMetrics.java

@@ -78,6 +78,12 @@ public SearchResponseMetrics(MeterRegistry meterRegistry) {
                 "The search phase dfs_query duration in milliseconds at the coordinator, expressed as a histogram",
                 "millis"
             ),
+            "fetch",
+            meterRegistry.registerLongHistogram(
+                String.format(Locale.ROOT, SEARCH_PHASE_METRIC_FORMAT, "fetch"),
+                "The search phase fetch duration in milliseconds at the coordinator, expressed as a histogram",
+                "millis"
+            ),
             "open_pit",
             meterRegistry.registerLongHistogram(
                 String.format(Locale.ROOT, SEARCH_PHASE_METRIC_FORMAT, "open_pit"),

server/src/test/java/org/elasticsearch/rest/action/search/SearchPhaseCoordinatorAPMMetricsTests.java

@@ -41,10 +41,11 @@ public class SearchPhaseCoordinatorAPMMetricsTests extends ESSingleNodeTestCase
     private static final String indexName = "test_coordinator_search_phase_metrics";
     private final int num_primaries = randomIntBetween(2, 7);
 
-    private static final String QUERY_SEARCH_PHASE_METRIC = "es.search_response.took_durations.query.histogram";
+    private static final String DFS_QUERY_SEARCH_PHASE_METRIC = "es.search_response.took_durations.dfs_query.histogram";
     private static final String DFS_SEARCH_PHASE_METRIC = "es.search_response.took_durations.dfs.histogram";
+    private static final String FETCH_SEARCH_PHASE_METRIC = "es.search_response.took_durations.fetch.histogram";
     private static final String OPEN_PIT_SEARCH_PHASE_METRIC = "es.search_response.took_durations.open_pit.histogram";
-    private static final String DFS_QUERY_SEARCH_PHASE_METRIC = "es.search_response.took_durations.dfs_query.histogram";
+    private static final String QUERY_SEARCH_PHASE_METRIC = "es.search_response.took_durations.query.histogram";
 
     @Override
     protected boolean resetNodeAfterTest() {
@@ -81,15 +82,15 @@ public void testSearchQueryThenFetch() {
             client().prepareSearch(indexName).setSearchType(SearchType.QUERY_THEN_FETCH).setQuery(simpleQueryStringQuery("doc1")),
             "1"
         );
-        assertMeasurements(List.of(QUERY_SEARCH_PHASE_METRIC));
+        assertMeasurements(List.of(QUERY_SEARCH_PHASE_METRIC, FETCH_SEARCH_PHASE_METRIC));
     }
 
     public void testDfsSearch() {
         assertSearchHitsWithoutFailures(
             client().prepareSearch(indexName).setSearchType(SearchType.DFS_QUERY_THEN_FETCH).setQuery(simpleQueryStringQuery("doc1")),
             "1"
         );
-        assertMeasurements(List.of(DFS_SEARCH_PHASE_METRIC, DFS_QUERY_SEARCH_PHASE_METRIC));
+        assertMeasurements(List.of(DFS_SEARCH_PHASE_METRIC, DFS_QUERY_SEARCH_PHASE_METRIC, FETCH_SEARCH_PHASE_METRIC));
     }
 
     public void testPointInTime() {
@@ -105,7 +106,7 @@ public void testPointInTime() {
                     .setQuery(simpleQueryStringQuery("doc1")),
                 "1"
             );
-            assertMeasurements(List.of(OPEN_PIT_SEARCH_PHASE_METRIC, QUERY_SEARCH_PHASE_METRIC));
+            assertMeasurements(List.of(OPEN_PIT_SEARCH_PHASE_METRIC, QUERY_SEARCH_PHASE_METRIC, FETCH_SEARCH_PHASE_METRIC));
         } finally {
             client().execute(TransportClosePointInTimeAction.TYPE, new ClosePointInTimeRequest(pointInTimeId)).actionGet();
         }