elastic · chrisparrinello · Oct 1, 2025 · Oct 2, 2025 · Oct 2, 2025
diff --git a/docs/changelog/135868.yaml b/docs/changelog/135868.yaml
@@ -0,0 +1,5 @@
+pr: 135868
+summary: Add APM metrics for duration of search phases at the coordinating node
+area: Search
+type: enhancement
+issues: []
diff --git a/server/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java b/server/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java
@@ -26,6 +26,7 @@
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.core.Releasable;
+import org.elasticsearch.index.search.stats.CoordinatorSearchPhaseAPMMetrics;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.SearchContextMissingException;
 import org.elasticsearch.search.SearchPhaseResult;
@@ -93,6 +94,8 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
     private final Map<String, PendingExecutions> pendingExecutionsPerNode;
     private final AtomicBoolean requestCancelled = new AtomicBoolean();
     private final int skippedCount;
+    private final CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseMetrics;
+    private long phaseStartTimeNanos;
 
     // protected for tests
     protected final SubscribableListener<Void> doneFuture = new SubscribableListener<>();
@@ -114,7 +117,8 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
         SearchTask task,
         SearchPhaseResults<Result> resultConsumer,
         int maxConcurrentRequestsPerNode,
-        SearchResponse.Clusters clusters
+        SearchResponse.Clusters clusters,
+        CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics
     ) {
         super(name);
         this.namedWriteableRegistry = namedWriteableRegistry;
@@ -155,6 +159,7 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
         // at the end of the search
         addReleasable(resultConsumer);
         this.clusters = clusters;
+        this.coordinatorSearchPhaseMetrics = coordinatorSearchPhaseAPMMetrics;
     }
 
     protected void notifyListShards(
@@ -374,7 +379,9 @@ protected void executeNextPhase(String currentPhase, Supplier<SearchPhase> nextP
 
     private void executePhase(SearchPhase phase) {
         try {
+            phaseStartTimeNanos = timeProvider.relativeCurrentNanosProvider().getAsLong();
             phase.run();
+
         } catch (RuntimeException e) {
             if (logger.isDebugEnabled()) {
                 logger.debug(() -> format("Failed to execute [%s] while moving to [%s] phase", request, phase.getName()), e);
@@ -621,6 +628,7 @@ protected BytesReference buildSearchContextId(ShardSearchFailure[] failures) {
      * @param cause the cause of the phase failure
      */
     public void onPhaseFailure(String phase, String msg, Throwable cause) {
+        recordPhaseTookTime(phase);
         raisePhaseFailure(new SearchPhaseExecutionException(phase, msg, cause, buildShardFailures()));
     }
 
@@ -666,9 +674,17 @@ void sendReleaseSearchContext(ShardSearchContextId contextId, Transport.Connecti
      * @see #onShardResult(SearchPhaseResult)
      */
     private void onPhaseDone() {  // as a tribute to @kimchy aka. finishHim()
+        recordPhaseTookTime(getName());
         executeNextPhase(getName(), this::getNextPhase);
     }
 
+    protected void recordPhaseTookTime(String phaseName) {
+        coordinatorSearchPhaseMetrics.onCoordinatorPhaseDone(
+            phaseName,
+            timeProvider.relativeCurrentNanosProvider().getAsLong() - phaseStartTimeNanos
+        );
+    }
+
     /**
      * Returns a connection to the node if connected otherwise and {@link org.elasticsearch.transport.ConnectTransportException} will be
      * thrown.

diff --git a/server/src/main/java/org/elasticsearch/action/search/CanMatchPreFilterSearchPhase.java b/server/src/main/java/org/elasticsearch/action/search/CanMatchPreFilterSearchPhase.java
@@ -19,6 +19,7 @@
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.query.CoordinatorRewriteContext;
 import org.elasticsearch.index.query.CoordinatorRewriteContextProvider;
+import org.elasticsearch.index.search.stats.CoordinatorSearchPhaseAPMMetrics;
 import org.elasticsearch.search.CanMatchShardResponse;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.SearchShardTarget;
@@ -71,6 +72,7 @@ final class CanMatchPreFilterSearchPhase {
     private final SearchTask task;
     private final Executor executor;
     private final boolean requireAtLeastOneMatch;
+    private final CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics;
 
     private final FixedBitSet possibleMatches;
     private final MinAndMax<?>[] minAndMaxes;
@@ -90,7 +92,8 @@ private CanMatchPreFilterSearchPhase(
         SearchTask task,
         boolean requireAtLeastOneMatch,
         CoordinatorRewriteContextProvider coordinatorRewriteContextProvider,
-        ActionListener<List<SearchShardIterator>> listener
+        ActionListener<List<SearchShardIterator>> listener,
+        CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics
     ) {
         this.logger = logger;
         this.searchTransportService = searchTransportService;
@@ -105,6 +108,7 @@ private CanMatchPreFilterSearchPhase(
         this.requireAtLeastOneMatch = requireAtLeastOneMatch;
         this.coordinatorRewriteContextProvider = coordinatorRewriteContextProvider;
         this.executor = executor;
+        this.coordinatorSearchPhaseAPMMetrics = coordinatorSearchPhaseAPMMetrics;
         final int size = shardsIts.size();
         possibleMatches = new FixedBitSet(size);
         minAndMaxes = new MinAndMax<?>[size];
@@ -136,7 +140,8 @@ public static SubscribableListener<List<SearchShardIterator>> execute(
         TransportSearchAction.SearchTimeProvider timeProvider,
         SearchTask task,
         boolean requireAtLeastOneMatch,
-        CoordinatorRewriteContextProvider coordinatorRewriteContextProvider
+        CoordinatorRewriteContextProvider coordinatorRewriteContextProvider,
+        CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics
     ) {
         if (shardsIts.isEmpty()) {
             return SubscribableListener.newSucceeded(List.of());
@@ -168,7 +173,8 @@ protected void doRun() {
                     task,
                     requireAtLeastOneMatch,
                     coordinatorRewriteContextProvider,
-                    listener
+                    listener,
+                    coordinatorSearchPhaseAPMMetrics
                 ).runCoordinatorRewritePhase();
             }
         });
@@ -184,6 +190,7 @@ private static boolean assertSearchCoordinationThread() {
     private void runCoordinatorRewritePhase() {
         // TODO: the index filter (i.e, `_index:patten`) should be prefiltered on the coordinator
         assert assertSearchCoordinationThread();
+        final long coordinatorStartTimeNanos = timeProvider.relativeCurrentNanosProvider().getAsLong();
         final List<SearchShardIterator> matchedShardLevelRequests = new ArrayList<>();
         for (SearchShardIterator searchShardIterator : shardsIts) {
             final CanMatchNodeRequest canMatchNodeRequest = new CanMatchNodeRequest(
@@ -223,6 +230,10 @@ private void runCoordinatorRewritePhase() {
             checkNoMissingShards(matchedShardLevelRequests);
             new Round(matchedShardLevelRequests).run();
         }
+        coordinatorSearchPhaseAPMMetrics.onCoordinatorPhaseDone(
+            "can_match",
+            timeProvider.relativeCurrentNanosProvider().getAsLong() - coordinatorStartTimeNanos
+        );
     }
 
     private void consumeResult(boolean canMatch, ShardSearchRequest request) {

diff --git a/server/src/main/java/org/elasticsearch/action/search/DfsQueryPhase.java b/server/src/main/java/org/elasticsearch/action/search/DfsQueryPhase.java
@@ -66,6 +66,7 @@ class DfsQueryPhase extends SearchPhase {
 
     // protected for testing
     protected SearchPhase nextPhase(AggregatedDfs dfs) {
+        context.recordPhaseTookTime(getName());
         return SearchQueryThenFetchAsyncAction.nextPhase(client, context, queryResult, dfs);
     }
 

diff --git a/server/src/main/java/org/elasticsearch/action/search/ExpandSearchPhase.java b/server/src/main/java/org/elasticsearch/action/search/ExpandSearchPhase.java
@@ -182,6 +182,7 @@ private static SearchSourceBuilder buildExpandSearchSourceBuilder(InnerHitBuilde
     }
 
     private void onPhaseDone() {
+        context.recordPhaseTookTime(getName());
         context.executeNextPhase(NAME, this::nextPhase);
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/action/search/FetchLookupFieldsPhase.java b/server/src/main/java/org/elasticsearch/action/search/FetchLookupFieldsPhase.java
@@ -144,6 +144,7 @@ public void onFailure(Exception e) {
     }
 
     private void sendResponse() {
+        context.recordPhaseTookTime(getName());
         context.sendSearchResponse(searchResponse, queryResults);
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java b/server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java
@@ -260,6 +260,7 @@ private void moveToNextPhase(
         AtomicArray<? extends SearchPhaseResult> fetchResultsArr,
         SearchPhaseController.ReducedQueryPhase reducedQueryPhase
     ) {
+        context.recordPhaseTookTime(getName());
         context.executeNextPhase(NAME, () -> {
             var resp = SearchPhaseController.merge(context.getRequest().scroll() != null, reducedQueryPhase, fetchResultsArr);
             context.addReleasable(resp);

diff --git a/server/src/main/java/org/elasticsearch/action/search/RankFeaturePhase.java b/server/src/main/java/org/elasticsearch/action/search/RankFeaturePhase.java
@@ -193,6 +193,7 @@ public void onResponse(RankFeatureDoc[] docsWithUpdatedScores) {
                         reducedQueryPhase,
                         topResults
                     );
+                    context.recordPhaseTookTime(getName());
                     moveToNextPhase(rankPhaseResults, reducedRankFeaturePhase);
                 }
 

diff --git a/server/src/main/java/org/elasticsearch/action/search/SearchDfsQueryThenFetchAsyncAction.java b/server/src/main/java/org/elasticsearch/action/search/SearchDfsQueryThenFetchAsyncAction.java
@@ -14,6 +14,7 @@
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.index.search.stats.CoordinatorSearchPhaseAPMMetrics;
 import org.elasticsearch.search.SearchPhaseResult;
 import org.elasticsearch.search.SearchShardTarget;
 import org.elasticsearch.search.dfs.DfsSearchResult;
@@ -47,7 +48,8 @@ final class SearchDfsQueryThenFetchAsyncAction extends AbstractSearchAsyncAction
         ClusterState clusterState,
         SearchTask task,
         SearchResponse.Clusters clusters,
-        Client client
+        Client client,
+        CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics
     ) {
         super(
             "dfs",
@@ -66,7 +68,8 @@ final class SearchDfsQueryThenFetchAsyncAction extends AbstractSearchAsyncAction
             task,
             new ArraySearchPhaseResults<>(shardsIts.size()),
             request.getMaxConcurrentShardRequests(),
-            clusters
+            clusters,
+            coordinatorSearchPhaseAPMMetrics
         );
         this.queryPhaseResultConsumer = queryPhaseResultConsumer;
         addReleasable(queryPhaseResultConsumer);

diff --git a/server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java b/server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java
@@ -34,6 +34,7 @@
 import org.elasticsearch.core.RefCounted;
 import org.elasticsearch.core.SimpleRefCounted;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.search.stats.CoordinatorSearchPhaseAPMMetrics;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.search.SearchPhaseResult;
 import org.elasticsearch.search.SearchService;
@@ -108,7 +109,8 @@ public class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<S
         SearchTask task,
         SearchResponse.Clusters clusters,
         Client client,
-        boolean batchQueryPhase
+        boolean batchQueryPhase,
+        CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics
     ) {
         super(
             "query",
@@ -127,7 +129,8 @@ public class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<S
             task,
             resultConsumer,
             request.getMaxConcurrentShardRequests(),
-            clusters
+            clusters,
+            coordinatorSearchPhaseAPMMetrics
         );
         this.topDocsSize = getTopDocsSize(request);
         this.trackTotalHitsUpTo = request.resolveTrackTotalHitsUpTo();

diff --git a/server/src/main/java/org/elasticsearch/action/search/TransportOpenPointInTimeAction.java b/server/src/main/java/org/elasticsearch/action/search/TransportOpenPointInTimeAction.java
@@ -31,6 +31,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.search.stats.CoordinatorSearchPhaseAPMMetrics;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.rest.RestStatus;
@@ -40,6 +41,7 @@
 import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.search.internal.ShardSearchContextId;
 import org.elasticsearch.tasks.Task;
+import org.elasticsearch.telemetry.TelemetryProvider;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.AbstractTransportRequest;
 import org.elasticsearch.transport.Transport;
@@ -70,6 +72,8 @@ public class TransportOpenPointInTimeAction extends HandledTransportAction<OpenP
     private final TransportService transportService;
     private final SearchService searchService;
     private final ClusterService clusterService;
+    private final TelemetryProvider telemetryProvider;
+    private final CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics;
 
     @Inject
     public TransportOpenPointInTimeAction(
@@ -79,7 +83,9 @@ public TransportOpenPointInTimeAction(
         TransportSearchAction transportSearchAction,
         SearchTransportService searchTransportService,
         NamedWriteableRegistry namedWriteableRegistry,
-        ClusterService clusterService
+        ClusterService clusterService,
+        TelemetryProvider telemetryProvider,
+        CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics
     ) {
         super(TYPE.name(), transportService, actionFilters, OpenPointInTimeRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.transportService = transportService;
@@ -88,6 +94,8 @@ public TransportOpenPointInTimeAction(
         this.searchTransportService = searchTransportService;
         this.namedWriteableRegistry = namedWriteableRegistry;
         this.clusterService = clusterService;
+        this.telemetryProvider = telemetryProvider;
+        this.coordinatorSearchPhaseAPMMetrics = coordinatorSearchPhaseAPMMetrics;
         transportService.registerRequestHandler(
             OPEN_SHARD_READER_CONTEXT_NAME,
             EsExecutors.DIRECT_EXECUTOR_SERVICE,
@@ -174,7 +182,8 @@ public void runNewSearchPhase(
                     timeProvider,
                     task,
                     false,
-                    searchService.getCoordinatorRewriteContextProvider(timeProvider::absoluteStartMillis)
+                    searchService.getCoordinatorRewriteContextProvider(timeProvider::absoluteStartMillis),
+                    coordinatorSearchPhaseAPMMetrics
                 )
                     .addListener(
                         listener.delegateFailureAndWrap(
@@ -240,7 +249,8 @@ void runOpenPointInTimePhase(
                 task,
                 new ArraySearchPhaseResults<>(shardIterators.size()),
                 searchRequest.getMaxConcurrentShardRequests(),
-                clusters
+                clusters,
+                coordinatorSearchPhaseAPMMetrics
             ) {
                 @Override
                 protected void executePhaseOnShard(

diff --git a/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java b/server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java
@@ -67,6 +67,7 @@
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.Rewriteable;
+import org.elasticsearch.index.search.stats.CoordinatorSearchPhaseAPMMetrics;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.shard.ShardNotFoundException;
 import org.elasticsearch.indices.ExecutorSelector;
@@ -170,6 +171,7 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
     private final UsageService usageService;
     private final boolean collectCCSTelemetry;
     private final TimeValue forceConnectTimeoutSecs;
+    private final CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics;
 
     @Inject
     public TransportSearchAction(
@@ -188,7 +190,8 @@ public TransportSearchAction(
         ExecutorSelector executorSelector,
         SearchResponseMetrics searchResponseMetrics,
         Client client,
-        UsageService usageService
+        UsageService usageService,
+        CoordinatorSearchPhaseAPMMetrics coordinatorSearchPhaseAPMMetrics
     ) {
         super(TYPE.name(), transportService, actionFilters, SearchRequest::new, EsExecutors.DIRECT_EXECUTOR_SERVICE);
         this.threadPool = threadPool;
@@ -218,6 +221,7 @@ public TransportSearchAction(
         this.searchResponseMetrics = searchResponseMetrics;
         this.client = client;
         this.usageService = usageService;
+        this.coordinatorSearchPhaseAPMMetrics = coordinatorSearchPhaseAPMMetrics;
         forceConnectTimeoutSecs = settings.getAsTime("search.ccs.force_connect_timeout", null);
     }
 
@@ -1623,7 +1627,8 @@ public void runNewSearchPhase(
                     timeProvider,
                     task,
                     requireAtLeastOneMatch,
-                    searchService.getCoordinatorRewriteContextProvider(timeProvider::absoluteStartMillis)
+                    searchService.getCoordinatorRewriteContextProvider(timeProvider::absoluteStartMillis),
+                    coordinatorSearchPhaseAPMMetrics
                 )
                     .addListener(
                         listener.delegateFailureAndWrap(
@@ -1682,7 +1687,8 @@ public void runNewSearchPhase(
                         clusterState,
                         task,
                         clusters,
-                        client
+                        client,
+                        coordinatorSearchPhaseAPMMetrics
                     );
                 } else {
                     assert searchRequest.searchType() == QUERY_THEN_FETCH : searchRequest.searchType();
@@ -1703,7 +1709,8 @@ public void runNewSearchPhase(
                         task,
                         clusters,
                         client,
-                        searchService.batchQueryPhase()
+                        searchService.batchQueryPhase(),
+                        coordinatorSearchPhaseAPMMetrics
                     );
                 }
                 success = true;