Add relevant attributes to shard search latency APM metrics

We already record latency of the query and fetch phase as APM metrics.
We'd like to be able to slice such latencies based on some recurring
categories of the request:

- does it have agg or hit only?
- is it sorted by field or by score?
- does it have a time range filter?
- does it target user data or internal indices?

This commit introduces introspection for a shard search request and stores the
extracted attributes together with the shard phase latency metrics.

This builds on top of elastic#134232 to use the same infra and store the same attributes
for shard level latency metrics.

Author: javanna

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

@@ -12,6 +12,8 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.BoostingQueryBuilder;
 import org.elasticsearch.index.query.ConstantScoreQueryBuilder;
@@ -20,6 +22,7 @@
 import org.elasticsearch.index.query.RangeQueryBuilder;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.internal.ShardSearchRequest;
 import org.elasticsearch.search.sort.FieldSortBuilder;
 import org.elasticsearch.search.sort.ScoreSortBuilder;
 import org.elasticsearch.search.sort.SortBuilder;
@@ -42,17 +45,37 @@ private SearchRequestAttributesExtractor() {}
 
     /**
      * Introspects the provided search request and extracts metadata from it about some of its characteristics.
-     *
      */
     public static Map<String, Object> extractAttributes(SearchRequest searchRequest, String[] localIndices) {
+        return extractAttributes(searchRequest.source(), searchRequest.scroll(), localIndices);
+    }
+
+    /**
+     * Introspects the provided shard search request and extracts metadata from it about some of its characteristics.
+     */
+    public static Map<String, Object> extractAttributes(ShardSearchRequest shardSearchRequest) {
+        Map<String, Object> attributes = extractAttributes(
+            shardSearchRequest.source(),
+            shardSearchRequest.scroll(),
+            shardSearchRequest.shardId().getIndexName()
+        );
+        boolean isSystem = ((EsExecutors.EsThread) Thread.currentThread()).isSystem();
+        attributes.put(SYSTEM_THREAD_ATTRIBUTE_NAME, isSystem);
+        return attributes;
+    }
+
+    private static Map<String, Object> extractAttributes(
+        SearchSourceBuilder searchSourceBuilder,
+        TimeValue scroll,
+        String... localIndices
+    ) {
         String target = extractIndices(localIndices);
 
         String pitOrScroll = null;
-        if (searchRequest.scroll() != null) {
+        if (scroll != null) {
             pitOrScroll = SCROLL;
         }
 
-        SearchSourceBuilder searchSourceBuilder = searchRequest.source();
         if (searchSourceBuilder == null) {
             return buildAttributesMap(target, ScoreSortBuilder.NAME, HITS_ONLY, false, false, false, pitOrScroll);
         }
@@ -144,7 +167,7 @@ private static final class QueryMetadataBuilder {
     private static final String TARGET_USER = "user";
     private static final String ERROR = "error";
 
-    static String extractIndices(String[] indices) {
+    static String extractIndices(String... indices) {
         try {
             // Note that indices are expected to be resolved, meaning wildcards are not handled on purpose
             // If indices resolve to data streams, the name of the data stream is returned as opposed to its backing indices
@@ -213,6 +236,7 @@ static String extractPrimarySort(SortBuilder<?> primarySortBuilder) {
     private static final String PIT = "pit";
     private static final String SCROLL = "scroll";
 
+    public static final String SYSTEM_THREAD_ATTRIBUTE_NAME = "system_thread";
     public static final Map<String, Object> SEARCH_SCROLL_ATTRIBUTES = Map.of(QUERY_TYPE_ATTRIBUTE, SCROLL);
 
     static String extractQueryType(SearchSourceBuilder searchSourceBuilder) {

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

@@ -9,13 +9,13 @@
 
 package org.elasticsearch.index.search.stats;
 
-import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.action.search.SearchRequestAttributesExtractor;
 import org.elasticsearch.index.shard.SearchOperationListener;
 import org.elasticsearch.search.internal.SearchContext;
+import org.elasticsearch.search.internal.ShardSearchRequest;
 import org.elasticsearch.telemetry.metric.LongHistogram;
 import org.elasticsearch.telemetry.metric.MeterRegistry;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
@@ -24,14 +24,9 @@ public final class ShardSearchPhaseAPMMetrics implements SearchOperationListener
     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";
 
-    public static final String SYSTEM_THREAD_ATTRIBUTE_NAME = "system_thread";
-
     private final LongHistogram queryPhaseMetric;
     private final LongHistogram fetchPhaseMetric;
 
-    // Avoid allocating objects in the search path and multithreading clashes
-    private static final ThreadLocal<Map<String, Object>> THREAD_LOCAL_ATTRS = ThreadLocal.withInitial(() -> new HashMap<>(1));
-
     public ShardSearchPhaseAPMMetrics(MeterRegistry meterRegistry) {
         this.queryPhaseMetric = meterRegistry.registerLongHistogram(
             QUERY_SEARCH_PHASE_METRIC,
@@ -47,18 +42,16 @@ public ShardSearchPhaseAPMMetrics(MeterRegistry meterRegistry) {
 
     @Override
     public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
-        recordPhaseLatency(queryPhaseMetric, tookInNanos);
+        recordPhaseLatency(queryPhaseMetric, tookInNanos, searchContext.request());
     }
 
     @Override
     public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
-        recordPhaseLatency(fetchPhaseMetric, tookInNanos);
+        recordPhaseLatency(fetchPhaseMetric, tookInNanos, searchContext.request());
     }
 
-    private static void recordPhaseLatency(LongHistogram histogramMetric, long tookInNanos) {
-        Map<String, Object> attrs = ShardSearchPhaseAPMMetrics.THREAD_LOCAL_ATTRS.get();
-        boolean isSystem = ((EsExecutors.EsThread) Thread.currentThread()).isSystem();
-        attrs.put(SYSTEM_THREAD_ATTRIBUTE_NAME, isSystem);
-        histogramMetric.record(TimeUnit.NANOSECONDS.toMillis(tookInNanos), attrs);
+    private static void recordPhaseLatency(LongHistogram histogramMetric, long tookInNanos, ShardSearchRequest request) {
+        Map<String, Object> attributes = SearchRequestAttributesExtractor.extractAttributes(request);
+        histogramMetric.record(TimeUnit.NANOSECONDS.toMillis(tookInNanos), attributes);
     }
 }