Enable Shard-Level Search-load rate metric (elastic#128660)

Introduces a new search load metric to the stats infrastructure, measured and tracked on a per-shard basis. The metric represents the Exponentially Weighted Moving Rate (EWMR) of search operations, calculated using the "took" time from each completed search phase.

Author: drempapis

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/indices.stats/80_search_load.yml

@@ -0,0 +1,52 @@
+---
+setup:
+  - requires:
+      cluster_features: ["mapper.search_load_per_shard"]
+      reason: Shard search load stats were introduced in 9.1
+---
+"Search load is tracked at shard level":
+  - do:
+      indices.create:
+        index: index
+        body:
+          mappings:
+            properties:
+              name:
+                type: text
+              description:
+                type: text
+              price:
+                type: double
+
+  - do:
+      indices.stats:
+        index: "index"
+        level: shards
+        metric: [ search ]
+
+  - match: { _all.total.search.recent_search_load: 0.0 }
+  - match: { indices.index.total.search.recent_search_load: 0.0 }
+  - match: { indices.index.shards.0.0.search.recent_search_load: 0.0 }
+
+  - do:
+      index:
+        index: index
+        body: { "name": "specialty coffee", "description": "arabica coffee beans", "price": 100 }
+
+  - do:
+      search:
+        index: index
+        body:
+          query:
+            match: { name: "specialty coffee" }
+          size: 1
+
+  - do:
+      indices.stats:
+        index: "index"
+        level: shards
+        metric: [ search ]
+
+  - gte: { _all.total.search.recent_search_load: 0.0 }
+  - gte: { indices.index.total.search.recent_search_load: 0.0 }
+  - gte: { indices.index.shards.0.0.search.recent_search_load: 0.0 }

server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java

@@ -53,6 +53,7 @@
 import org.elasticsearch.index.flush.FlushStats;
 import org.elasticsearch.index.mapper.MapperMetrics;
 import org.elasticsearch.index.mapper.SourceToParse;
+import org.elasticsearch.index.search.stats.SearchStatsSettings;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.translog.TestTranslog;
@@ -638,7 +639,8 @@ public static final IndexShard newIndexShard(
             System::nanoTime,
             null,
             MapperMetrics.NOOP,
-            new IndexingStatsSettings(ClusterSettings.createBuiltInClusterSettings())
+            new IndexingStatsSettings(ClusterSettings.createBuiltInClusterSettings()),
+            new SearchStatsSettings(ClusterSettings.createBuiltInClusterSettings())
         );
     }
 

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -292,6 +292,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion SEARCH_SOURCE_EXCLUDE_VECTORS_PARAM = def(9_092_0_00);
     public static final TransportVersion SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS = def(9_093_0_00);
     public static final TransportVersion ML_INFERENCE_ELASTIC_RERANK = def(9_094_0_00);
+    public static final TransportVersion SEARCH_LOAD_PER_INDEX_STATS = def(9_095_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -90,6 +90,7 @@
 import org.elasticsearch.index.MergePolicyConfig;
 import org.elasticsearch.index.engine.ThreadPoolMergeExecutorService;
 import org.elasticsearch.index.engine.ThreadPoolMergeScheduler;
+import org.elasticsearch.index.search.stats.SearchStatsSettings;
 import org.elasticsearch.index.shard.IndexingStatsSettings;
 import org.elasticsearch.indices.IndexingMemoryController;
 import org.elasticsearch.indices.IndicesQueryCache;
@@ -640,6 +641,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         ShardsAvailabilityHealthIndicatorService.REPLICA_UNASSIGNED_BUFFER_TIME,
         DataStreamFailureStoreSettings.DATA_STREAM_FAILURE_STORED_ENABLED_SETTING,
         IndexingStatsSettings.RECENT_WRITE_LOAD_HALF_LIFE_SETTING,
+        SearchStatsSettings.RECENT_READ_LOAD_HALF_LIFE_SETTING,
         TransportGetAllocationStatsAction.CACHE_TTL_SETTING
     );
 }

server/src/main/java/org/elasticsearch/index/IndexModule.java

@@ -48,6 +48,7 @@
 import org.elasticsearch.index.mapper.MapperMetrics;
 import org.elasticsearch.index.mapper.MapperRegistry;
 import org.elasticsearch.index.mapper.MapperService;
+import org.elasticsearch.index.search.stats.SearchStatsSettings;
 import org.elasticsearch.index.shard.IndexEventListener;
 import org.elasticsearch.index.shard.IndexingOperationListener;
 import org.elasticsearch.index.shard.IndexingStatsSettings;
@@ -179,6 +180,7 @@ public interface DirectoryWrapper {
     private final SetOnce<Engine.IndexCommitListener> indexCommitListener = new SetOnce<>();
     private final MapperMetrics mapperMetrics;
     private final IndexingStatsSettings indexingStatsSettings;
+    private final SearchStatsSettings searchStatsSettings;
 
     /**
      * Construct the index module for the index with the specified index settings. The index module contains extension points for plugins
@@ -200,7 +202,8 @@ public IndexModule(
         final SlowLogFieldProvider slowLogFieldProvider,
         final MapperMetrics mapperMetrics,
         final List<SearchOperationListener> searchOperationListeners,
-        final IndexingStatsSettings indexingStatsSettings
+        final IndexingStatsSettings indexingStatsSettings,
+        final SearchStatsSettings searchStatsSettings
     ) {
         this.indexSettings = indexSettings;
         this.analysisRegistry = analysisRegistry;
@@ -216,6 +219,7 @@ public IndexModule(
         this.recoveryStateFactories = recoveryStateFactories;
         this.mapperMetrics = mapperMetrics;
         this.indexingStatsSettings = indexingStatsSettings;
+        this.searchStatsSettings = searchStatsSettings;
     }
 
     /**
@@ -552,7 +556,8 @@ public IndexService newIndexService(
                 indexCommitListener.get(),
                 mapperMetrics,
                 queryRewriteInterceptor,
-                indexingStatsSettings
+                indexingStatsSettings,
+                searchStatsSettings
             );
             success = true;
             return indexService;

server/src/main/java/org/elasticsearch/index/IndexService.java

@@ -67,6 +67,7 @@
 import org.elasticsearch.index.query.QueryRewriteContext;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.query.SearchIndexNameMatcher;
+import org.elasticsearch.index.search.stats.SearchStatsSettings;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
 import org.elasticsearch.index.shard.GlobalCheckpointSyncer;
 import org.elasticsearch.index.shard.IndexEventListener;
@@ -170,6 +171,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final MapperMetrics mapperMetrics;
     private final QueryRewriteInterceptor queryRewriteInterceptor;
     private final IndexingStatsSettings indexingStatsSettings;
+    private final SearchStatsSettings searchStatsSettings;
 
     @SuppressWarnings("this-escape")
     public IndexService(
@@ -207,7 +209,8 @@ public IndexService(
         Engine.IndexCommitListener indexCommitListener,
         MapperMetrics mapperMetrics,
         QueryRewriteInterceptor queryRewriteInterceptor,
-        IndexingStatsSettings indexingStatsSettings
+        IndexingStatsSettings indexingStatsSettings,
+        SearchStatsSettings searchStatsSettings
     ) {
         super(indexSettings);
         assert indexCreationContext != IndexCreationContext.RELOAD_ANALYZERS
@@ -293,6 +296,7 @@ public IndexService(
             this.retentionLeaseSyncTask = new AsyncRetentionLeaseSyncTask(this);
         }
         this.indexingStatsSettings = indexingStatsSettings;
+        this.searchStatsSettings = searchStatsSettings;
         updateFsyncTaskIfNecessary();
     }
 
@@ -583,7 +587,8 @@ public synchronized IndexShard createShard(
                 System::nanoTime,
                 indexCommitListener,
                 mapperMetrics,
-                indexingStatsSettings
+                indexingStatsSettings,
+                searchStatsSettings
             );
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);

server/src/main/java/org/elasticsearch/index/mapper/MapperFeatures.java

@@ -43,6 +43,7 @@ public class MapperFeatures implements FeatureSpecification {
     static final NodeFeature NPE_ON_DIMS_UPDATE_FIX = new NodeFeature("mapper.npe_on_dims_update_fix");
     static final NodeFeature IVF_FORMAT_CLUSTER_FEATURE = new NodeFeature("mapper.ivf_format_cluster_feature");
     static final NodeFeature IVF_NESTED_SUPPORT = new NodeFeature("mapper.ivf_nested_support");
+    static final NodeFeature SEARCH_LOAD_PER_SHARD = new NodeFeature("mapper.search_load_per_shard");
 
     @Override
     public Set<NodeFeature> getTestFeatures() {
@@ -72,7 +73,8 @@ public Set<NodeFeature> getTestFeatures() {
             RESCORE_ZERO_VECTOR_QUANTIZED_VECTOR_MAPPING,
             USE_DEFAULT_OVERSAMPLE_VALUE_FOR_BBQ,
             IVF_FORMAT_CLUSTER_FEATURE,
-            IVF_NESTED_SUPPORT
+            IVF_NESTED_SUPPORT,
+            SEARCH_LOAD_PER_SHARD
         );
     }
 }

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

@@ -49,6 +49,10 @@ public static class Stats implements Writeable, ToXContentFragment {
         private long queryFailure;
         private long fetchFailure;
 
+        // This tracks the search execution time across different phases (e.g., query, fetch, etc.), favouring more recent
+        // values by assigning them greater significance than older values.
+        private double recentSearchLoad;
+
         private Stats() {
             // for internal use, initializes all counts to 0
         }
@@ -67,7 +71,8 @@ public Stats(
             long scrollCurrent,
             long suggestCount,
             long suggestTimeInMillis,
-            long suggestCurrent
+            long suggestCurrent,
+            double recentSearchLoad
         ) {
             this.queryCount = queryCount;
             this.queryTimeInMillis = queryTimeInMillis;
@@ -86,6 +91,9 @@ public Stats(
             this.suggestCount = suggestCount;
             this.suggestTimeInMillis = suggestTimeInMillis;
             this.suggestCurrent = suggestCurrent;
+
+            this.recentSearchLoad = recentSearchLoad;
+
         }
 
         private Stats(StreamInput in) throws IOException {
@@ -109,6 +117,10 @@ private Stats(StreamInput in) throws IOException {
                 queryFailure = in.readVLong();
                 fetchFailure = in.readVLong();
             }
+
+            if (in.getTransportVersion().onOrAfter(TransportVersions.SEARCH_LOAD_PER_INDEX_STATS)) {
+                recentSearchLoad = in.readDouble();
+            }
         }
 
         @Override
@@ -133,6 +145,10 @@ public void writeTo(StreamOutput out) throws IOException {
                 out.writeVLong(queryFailure);
                 out.writeVLong(fetchFailure);
             }
+
+            if (out.getTransportVersion().onOrAfter(TransportVersions.SEARCH_LOAD_PER_INDEX_STATS)) {
+                out.writeDouble(recentSearchLoad);
+            }
         }
 
         public void add(Stats stats) {
@@ -153,6 +169,8 @@ public void add(Stats stats) {
             suggestCount += stats.suggestCount;
             suggestTimeInMillis += stats.suggestTimeInMillis;
             suggestCurrent += stats.suggestCurrent;
+
+            recentSearchLoad += stats.recentSearchLoad;
         }
 
         public void addForClosingShard(Stats stats) {
@@ -171,6 +189,8 @@ public void addForClosingShard(Stats stats) {
 
             suggestCount += stats.suggestCount;
             suggestTimeInMillis += stats.suggestTimeInMillis;
+
+            recentSearchLoad += stats.recentSearchLoad;
         }
 
         public long getQueryCount() {
@@ -245,6 +265,10 @@ public long getSuggestCurrent() {
             return suggestCurrent;
         }
 
+        public double getSearchLoadRate() {
+            return recentSearchLoad;
+        }
+
         public static Stats readStats(StreamInput in) throws IOException {
             return new Stats(in);
         }
@@ -269,6 +293,8 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.humanReadableField(Fields.SUGGEST_TIME_IN_MILLIS, Fields.SUGGEST_TIME, getSuggestTime());
             builder.field(Fields.SUGGEST_CURRENT, suggestCurrent);
 
+            builder.field(Fields.RECENT_SEARCH_LOAD, recentSearchLoad);
+
             return builder;
         }
 
@@ -290,7 +316,8 @@ public boolean equals(Object o) {
                 && scrollCurrent == that.scrollCurrent
                 && suggestCount == that.suggestCount
                 && suggestTimeInMillis == that.suggestTimeInMillis
-                && suggestCurrent == that.suggestCurrent;
+                && suggestCurrent == that.suggestCurrent
+                && recentSearchLoad == that.recentSearchLoad;
         }
 
         @Override
@@ -309,7 +336,8 @@ public int hashCode() {
                 scrollCurrent,
                 suggestCount,
                 suggestTimeInMillis,
-                suggestCurrent
+                suggestCurrent,
+                recentSearchLoad
             );
         }
     }
@@ -427,6 +455,7 @@ static final class Fields {
         static final String SUGGEST_TIME = "suggest_time";
         static final String SUGGEST_TIME_IN_MILLIS = "suggest_time_in_millis";
         static final String SUGGEST_CURRENT = "suggest_current";
+        static final String RECENT_SEARCH_LOAD = "recent_search_load";
     }
 
     @Override

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

@@ -0,0 +1,50 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.index.search.stats;
+
+import org.elasticsearch.common.settings.ClusterSettings;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.TimeValue;
+
+/**
+ * Container for cluster settings
+ */
+public class SearchStatsSettings {
+
+    public static final TimeValue RECENT_READ_LOAD_HALF_LIFE_DEFAULT = TimeValue.timeValueMinutes(5);
+    static final TimeValue RECENT_READ_LOAD_HALF_LIFE_MIN = TimeValue.timeValueSeconds(1); // A sub-second half-life makes no sense
+    static final TimeValue RECENT_READ_LOAD_HALF_LIFE_MAX = TimeValue.timeValueDays(100_000); // Long.MAX_VALUE nanos, rounded down
+
+    /**
+     * A cluster setting giving the half-life, in seconds, to use for the Exponentially Weighted Moving Rate calculation used for the
+     * recency-weighted read load
+     *
+     * <p>This is dynamic, but changes only apply to newly-opened shards.
+     */
+    public static final Setting<TimeValue> RECENT_READ_LOAD_HALF_LIFE_SETTING = Setting.timeSetting(
+        "indices.stats.recent_read_load.half_life",
+        RECENT_READ_LOAD_HALF_LIFE_DEFAULT,
+        RECENT_READ_LOAD_HALF_LIFE_MIN,
+        RECENT_READ_LOAD_HALF_LIFE_MAX,
+        Setting.Property.Dynamic,
+        Setting.Property.NodeScope
+    );
+
+    private volatile TimeValue recentReadLoadHalfLifeForNewShards = RECENT_READ_LOAD_HALF_LIFE_SETTING.getDefault(Settings.EMPTY);
+
+    public SearchStatsSettings(ClusterSettings clusterSettings) {
+        clusterSettings.initializeAndWatch(RECENT_READ_LOAD_HALF_LIFE_SETTING, value -> recentReadLoadHalfLifeForNewShards = value);
+    }
+
+    public TimeValue getRecentReadLoadHalfLifeForNewShards() {
+        return recentReadLoadHalfLifeForNewShards;
+    }
+}