Add time range bucketing attribute to APM took time latency metrics

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) {