Add time range bucketing attribute to APM took time latency metrics (#135549)

This is similar to #135524, but adding the time range bucketing attribute to the took time latency metric.

That requires a bit of ceremony as the took time metric is recorded on the coordinating node, while the time range filter is parsed on each shard. We don't have mappings available on the coord node, which are needed to parse dates. Thus we need to rely on date parsing done on the data nodes, which requires sending back the parsed value to the coord node, performing some simple reduction on it, and adding it back to the search response.

This also includes two improvements to the existing mechanism:
- take date precision into account
- track the time range as part of query rewrite, additionally to query execution. This way queries that are rewritten to a match_all still see their original range reported

Author: javanna

docs/changelog/135549.yaml

@@ -0,0 +1,5 @@
+pr: 135549
+summary: Add time range bucketing attribute to APM took time latency metrics
+area: Search
+type: enhancement
+issues: []

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

@@ -86,7 +86,8 @@ public SearchPhaseController.ReducedQueryPhase reduce() throws Exception {
             1,
             0,
             0,
-            results.isEmpty()
+            results.isEmpty(),
+            null
         );
         if (progressListener != SearchProgressListener.NOOP) {
             progressListener.notifyFinalReduce(

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

@@ -251,7 +251,8 @@ private SearchPhaseController.ReducedQueryPhase newReducedQueryPhaseResults(
             reducedQueryPhase.numReducePhases(),
             reducedQueryPhase.size(),
             reducedQueryPhase.from(),
-            reducedQueryPhase.isEmptyResult()
+            reducedQueryPhase.isEmptyResult(),
+            reducedQueryPhase.timeRangeFilterFromMillis()
         );
     }
 

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

@@ -487,7 +487,8 @@ static ReducedQueryPhase reducedQueryPhase(
                 numReducePhases,
                 0,
                 0,
-                true
+                true,
+                null
             );
         }
         final List<QuerySearchResult> nonNullResults = new ArrayList<>();
@@ -516,6 +517,7 @@ static ReducedQueryPhase reducedQueryPhase(
             : Collections.emptyMap();
         int from = 0;
         int size = 0;
+        Long timeRangeFilterFromMillis = null;
         DocValueFormat[] sortValueFormats = null;
         for (QuerySearchResult result : nonNullResults) {
             from = result.from();
@@ -525,6 +527,16 @@ static ReducedQueryPhase reducedQueryPhase(
                 sortValueFormats = result.sortValueFormats();
             }
 
+            if (result.getTimeRangeFilterFromMillis() != null) {
+                if (timeRangeFilterFromMillis == null) {
+                    timeRangeFilterFromMillis = result.getTimeRangeFilterFromMillis();
+                } else {
+                    // all shards should hold the same value, besides edge cases like different mappings
+                    // for event.ingested and @timestamp across indices being searched
+                    timeRangeFilterFromMillis = Math.min(result.getTimeRangeFilterFromMillis(), timeRangeFilterFromMillis);
+                }
+            }
+
             if (hasSuggest) {
                 assert result.suggest() != null;
                 for (Suggestion<? extends Suggestion.Entry<? extends Suggestion.Entry.Option>> suggestion : result.suggest()) {
@@ -579,7 +591,8 @@ static ReducedQueryPhase reducedQueryPhase(
             numReducePhases,
             size,
             from,
-            false
+            false,
+            timeRangeFilterFromMillis
         );
     }
 
@@ -662,7 +675,8 @@ public record ReducedQueryPhase(
         // the offset into the merged top hits
         int from,
         // <code>true</code> iff the query phase had no results. Otherwise <code>false</code>
-        boolean isEmptyResult
+        boolean isEmptyResult,
+        Long timeRangeFilterFromMillis
     ) {
 
         public ReducedQueryPhase {
@@ -683,7 +697,8 @@ public SearchResponseSections buildResponse(SearchHits hits, Collection<? extend
                 timedOut,
                 terminatedEarly,
                 buildSearchProfileResults(fetchResults),
-                numReducePhases
+                numReducePhases,
+                timeRangeFilterFromMillis
             );
         }
 

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

@@ -11,6 +11,8 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.TimeValue;
@@ -53,11 +55,15 @@ public static Map<String, Object> extractAttributes(SearchRequest searchRequest,
     /**
      * Introspects the provided shard search request and extracts metadata from it about some of its characteristics.
      */
-    public static Map<String, Object> extractAttributes(ShardSearchRequest shardSearchRequest, Long rangeTimestampFrom, long nowInMillis) {
+    public static Map<String, Object> extractAttributes(
+        ShardSearchRequest shardSearchRequest,
+        Long timeRangeFilterFromMillis,
+        long nowInMillis
+    ) {
         Map<String, Object> attributes = extractAttributes(
             shardSearchRequest.source(),
             shardSearchRequest.scroll(),
-            rangeTimestampFrom,
+            timeRangeFilterFromMillis,
             nowInMillis,
             shardSearchRequest.shardId().getIndexName()
         );
@@ -69,7 +75,7 @@ public static Map<String, Object> extractAttributes(ShardSearchRequest shardSear
     private static Map<String, Object> extractAttributes(
         SearchSourceBuilder searchSourceBuilder,
         TimeValue scroll,
-        Long rangeTimestampFrom,
+        Long timeRangeFilterFromMillis,
         long nowInMillis,
         String... localIndices
     ) {
@@ -107,9 +113,9 @@ private static Map<String, Object> extractAttributes(
         }
 
         final boolean hasKnn = searchSourceBuilder.knnSearch().isEmpty() == false || queryMetadataBuilder.knnQuery;
-        String timestampRangeFilter = null;
-        if (rangeTimestampFrom != null) {
-            timestampRangeFilter = introspectTimeRange(rangeTimestampFrom, nowInMillis);
+        String timeRangeFilterFrom = null;
+        if (timeRangeFilterFromMillis != null) {
+            timeRangeFilterFrom = introspectTimeRange(timeRangeFilterFromMillis, nowInMillis);
         }
         return buildAttributesMap(
             target,
@@ -119,7 +125,7 @@ private static Map<String, Object> extractAttributes(
             queryMetadataBuilder.rangeOnTimestamp,
             queryMetadataBuilder.rangeOnEventIngested,
             pitOrScroll,
-            timestampRangeFilter
+            timeRangeFilterFrom
         );
     }
 
@@ -131,7 +137,7 @@ private static Map<String, Object> buildAttributesMap(
         boolean rangeOnTimestamp,
         boolean rangeOnEventIngested,
         String pitOrScroll,
-        String timestampRangeFilter
+        String timeRangeFilterFrom
     ) {
         Map<String, Object> attributes = new HashMap<>(5, 1.0f);
         attributes.put(TARGET_ATTRIBUTE, target);
@@ -143,14 +149,18 @@ private static Map<String, Object> buildAttributesMap(
         if (knn) {
             attributes.put(KNN_ATTRIBUTE, knn);
         }
-        if (rangeOnTimestamp) {
-            attributes.put(RANGE_TIMESTAMP_ATTRIBUTE, rangeOnTimestamp);
-        }
-        if (rangeOnEventIngested) {
-            attributes.put(RANGE_EVENT_INGESTED_ATTRIBUTE, rangeOnEventIngested);
+        if (rangeOnTimestamp && rangeOnEventIngested) {
+            attributes.put(
+                TIME_RANGE_FILTER_FIELD_ATTRIBUTE,
+                DataStream.TIMESTAMP_FIELD_NAME + "_AND_" + IndexMetadata.EVENT_INGESTED_FIELD_NAME
+            );
+        } else if (rangeOnEventIngested) {
+            attributes.put(TIME_RANGE_FILTER_FIELD_ATTRIBUTE, IndexMetadata.EVENT_INGESTED_FIELD_NAME);
+        } else if (rangeOnTimestamp) {
+            attributes.put(TIME_RANGE_FILTER_FIELD_ATTRIBUTE, DataStream.TIMESTAMP_FIELD_NAME);
         }
-        if (timestampRangeFilter != null) {
-            attributes.put(TIMESTAMP_RANGE_FILTER_ATTRIBUTE, timestampRangeFilter);
+        if (timeRangeFilterFrom != null) {
+            attributes.put(TIME_RANGE_FILTER_FROM_ATTRIBUTE, timeRangeFilterFrom);
         }
         return attributes;
     }
@@ -166,9 +176,8 @@ private static final class QueryMetadataBuilder {
     static final String QUERY_TYPE_ATTRIBUTE = "query_type";
     static final String PIT_SCROLL_ATTRIBUTE = "pit_scroll";
     static final String KNN_ATTRIBUTE = "knn";
-    static final String RANGE_TIMESTAMP_ATTRIBUTE = "range_timestamp";
-    static final String RANGE_EVENT_INGESTED_ATTRIBUTE = "range_event_ingested";
-    static final String TIMESTAMP_RANGE_FILTER_ATTRIBUTE = "timestamp_range_filter";
+    static final String TIME_RANGE_FILTER_FIELD_ATTRIBUTE = "time_range_filter_field";
+    static final String TIME_RANGE_FILTER_FROM_ATTRIBUTE = "time_range_filter_from";
 
     private static final String TARGET_KIBANA = ".kibana";
     private static final String TARGET_ML = ".ml";
@@ -303,6 +312,10 @@ private static void introspectQueryBuilder(QueryBuilder queryBuilder, QueryMetad
                 introspectQueryBuilder(nested.query(), queryMetadataBuilder, ++level);
                 break;
             case RangeQueryBuilder range:
+                // Note that the outcome of this switch differs depending on whether it is executed on the coord node, or data node.
+                // Data nodes perform query rewrite on each shard. That means that a query that reports a certain time range filter at the
+                // coordinator, may not report the same for all the shards it targets, but rather only for those that do end up executing
+                // a true range query at the shard level.
                 switch (range.fieldName()) {
                     // don't track unbounded ranges, they translate to either match_none if the field does not exist
                     // or match_all if the field is mapped
@@ -343,9 +356,16 @@ private enum TimeRangeBucket {
         }
     }
 
-    static String introspectTimeRange(long timeRangeFrom, long nowInMillis) {
+    public static void addTimeRangeAttribute(Long timeRangeFrom, long nowInMillis, Map<String, Object> attributes) {
+        if (timeRangeFrom != null) {
+            String timestampRangeFilter = introspectTimeRange(timeRangeFrom, nowInMillis);
+            attributes.put(TIME_RANGE_FILTER_FROM_ATTRIBUTE, timestampRangeFilter);
+        }
+    }
+
+    static String introspectTimeRange(long timeRangeFromMillis, long nowInMillis) {
         for (TimeRangeBucket value : TimeRangeBucket.values()) {
-            if (timeRangeFrom >= nowInMillis - value.millis) {
+            if (timeRangeFromMillis >= nowInMillis - value.millis) {
                 return value.bucketName;
             }
         }

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

@@ -87,6 +87,8 @@ public class SearchResponse extends ActionResponse implements ChunkedToXContentO
     private final ShardSearchFailure[] shardFailures;
     private final Clusters clusters;
     private final long tookInMillis;
+    // only used for telemetry purposes on the coordinating node, where the search response gets created
+    private transient Long timeRangeFilterFromMillis;
 
     private final RefCounted refCounted = LeakTracker.wrap(new SimpleRefCounted());
 
@@ -187,6 +189,7 @@ public SearchResponse(
             clusters,
             pointInTimeId
         );
+        this.timeRangeFilterFromMillis = searchResponseSections.timeRangeFilterFromMillis;
     }
 
     public SearchResponse(
@@ -464,6 +467,10 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeOptionalBytesReference(pointInTimeId);
     }
 
+    public Long getTimeRangeFilterFromMillis() {
+        return timeRangeFilterFromMillis;
+    }
+
     @Override
     public String toString() {
         return hasReferences() == false ? "SearchResponse[released]" : Strings.toString(this);

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

@@ -32,7 +32,8 @@ public class SearchResponseSections implements Releasable {
         false,
         null,
         null,
-        1
+        1,
+        null
     );
     public static final SearchResponseSections EMPTY_WITHOUT_TOTAL_HITS = new SearchResponseSections(
         SearchHits.EMPTY_WITHOUT_TOTAL_HITS,
@@ -41,7 +42,8 @@ public class SearchResponseSections implements Releasable {
         false,
         null,
         null,
-        1
+        1,
+        null
     );
     protected final SearchHits hits;
     protected final InternalAggregations aggregations;
@@ -50,6 +52,7 @@ public class SearchResponseSections implements Releasable {
     protected final boolean timedOut;
     protected final Boolean terminatedEarly;
     protected final int numReducePhases;
+    protected final Long timeRangeFilterFromMillis;
 
     public SearchResponseSections(
         SearchHits hits,
@@ -58,7 +61,8 @@ public SearchResponseSections(
         boolean timedOut,
         Boolean terminatedEarly,
         SearchProfileResults profileResults,
-        int numReducePhases
+        int numReducePhases,
+        Long timeRangeFilterFromMillis
     ) {
         this.hits = hits;
         this.aggregations = aggregations;
@@ -67,6 +71,7 @@ public SearchResponseSections(
         this.timedOut = timedOut;
         this.terminatedEarly = terminatedEarly;
         this.numReducePhases = numReducePhases;
+        this.timeRangeFilterFromMillis = timeRangeFilterFromMillis;
     }
 
     public final SearchHits hits() {

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

@@ -430,7 +430,12 @@ public void onFailure(Exception e) {
                     Arrays.stream(resolvedIndices.getConcreteLocalIndices()).map(Index::getName).toArray(String[]::new)
                 );
                 if (collectCCSTelemetry == false || resolvedIndices.getRemoteClusterIndices().isEmpty()) {
-                    searchResponseActionListener = new SearchTelemetryListener(delegate, searchResponseMetrics, searchRequestAttributes);
+                    searchResponseActionListener = new SearchTelemetryListener(
+                        delegate,
+                        searchResponseMetrics,
+                        searchRequestAttributes,
+                        timeProvider.absoluteStartMillis()
+                    );
                 } else {
                     CCSUsage.Builder usageBuilder = new CCSUsage.Builder();
                     usageBuilder.setRemotesCount(resolvedIndices.getRemoteClusterIndices().size());
@@ -459,6 +464,7 @@ public void onFailure(Exception e) {
                         delegate,
                         searchResponseMetrics,
                         searchRequestAttributes,
+                        timeProvider.absoluteStartMillis(),
                         usageService,
                         usageBuilder
                     );
@@ -2046,6 +2052,7 @@ static String[] ignoreBlockedIndices(ProjectState projectState, String[] concret
     private static class SearchTelemetryListener extends DelegatingActionListener<SearchResponse, SearchResponse> {
         private final CCSUsage.Builder usageBuilder;
         private final SearchResponseMetrics searchResponseMetrics;
+        private final long nowInMillis;
         private final UsageService usageService;
         private final boolean collectCCSTelemetry;
         private final Map<String, Object> searchRequestAttributes;
@@ -2054,12 +2061,14 @@ private static class SearchTelemetryListener extends DelegatingActionListener<Se
             ActionListener<SearchResponse> listener,
             SearchResponseMetrics searchResponseMetrics,
             Map<String, Object> searchRequestAttributes,
+            long nowInMillis,
             UsageService usageService,
             CCSUsage.Builder usageBuilder
         ) {
             super(listener);
             this.searchResponseMetrics = searchResponseMetrics;
             this.searchRequestAttributes = searchRequestAttributes;
+            this.nowInMillis = nowInMillis;
             this.collectCCSTelemetry = true;
             this.usageService = usageService;
             this.usageBuilder = usageBuilder;
@@ -2068,11 +2077,13 @@ private static class SearchTelemetryListener extends DelegatingActionListener<Se
         SearchTelemetryListener(
             ActionListener<SearchResponse> listener,
             SearchResponseMetrics searchResponseMetrics,
-            Map<String, Object> searchRequestAttributes
+            Map<String, Object> searchRequestAttributes,
+            long nowInMillis
         ) {
             super(listener);
             this.searchResponseMetrics = searchResponseMetrics;
             this.searchRequestAttributes = searchRequestAttributes;
+            this.nowInMillis = nowInMillis;
             this.collectCCSTelemetry = false;
             this.usageService = null;
             this.usageBuilder = null;
@@ -2081,7 +2092,12 @@ private static class SearchTelemetryListener extends DelegatingActionListener<Se
         @Override
         public void onResponse(SearchResponse searchResponse) {
             try {
-                searchResponseMetrics.recordTookTime(searchResponse.getTookInMillis(), searchRequestAttributes);
+                searchResponseMetrics.recordTookTime(
+                    searchResponse.getTookInMillis(),
+                    searchResponse.getTimeRangeFilterFromMillis(),
+                    nowInMillis,
+                    searchRequestAttributes
+                );
                 SearchResponseMetrics.ResponseCountTotalStatus responseCountTotalStatus =
                     SearchResponseMetrics.ResponseCountTotalStatus.SUCCESS;
                 if (searchResponse.getShardFailures() != null && searchResponse.getShardFailures().length > 0) {