Enable Shard-Level Search-load rate metric

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

@@ -284,6 +284,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion JOIN_ON_ALIASES = def(9_088_0_00);
     public static final TransportVersion ILM_ADD_SKIP_SETTING = def(9_089_0_00);
     public static final TransportVersion ML_INFERENCE_MISTRAL_CHAT_COMPLETION_ADDED = def(9_090_0_00);
+    public static final TransportVersion SEARCH_LOAD_PER_INDEX_STATS = def(9_091_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -89,6 +89,7 @@
 import org.elasticsearch.index.IndexingPressure;
 import org.elasticsearch.index.MergePolicyConfig;
 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;
@@ -636,6 +637,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/search/stats/SearchStats.java

@@ -49,6 +49,8 @@ public static class Stats implements Writeable, ToXContentFragment {
         private long queryFailure;
         private long fetchFailure;
 
+        private double recentSearchLoad;
+
         private Stats() {
             // for internal use, initializes all counts to 0
         }
@@ -67,7 +69,8 @@ public Stats(
             long scrollCurrent,
             long suggestCount,
             long suggestTimeInMillis,
-            long suggestCurrent
+            long suggestCurrent,
+            double recentSearchLoad
         ) {
             this.queryCount = queryCount;
             this.queryTimeInMillis = queryTimeInMillis;
@@ -86,6 +89,9 @@ public Stats(
             this.suggestCount = suggestCount;
             this.suggestTimeInMillis = suggestTimeInMillis;
             this.suggestCurrent = suggestCurrent;
+
+            this.recentSearchLoad = recentSearchLoad;
+
         }
 
         private Stats(StreamInput in) throws IOException {
@@ -109,6 +115,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 +143,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 +167,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 +187,8 @@ public void addForClosingShard(Stats stats) {
 
             suggestCount += stats.suggestCount;
             suggestTimeInMillis += stats.suggestTimeInMillis;
+
+            recentSearchLoad += stats.recentSearchLoad;
         }
 
         public long getQueryCount() {
@@ -245,6 +263,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 +291,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.EMW_RATE, recentSearchLoad);
+
             return builder;
         }
 
@@ -290,7 +314,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 +334,8 @@ public int hashCode() {
                 scrollCurrent,
                 suggestCount,
                 suggestTimeInMillis,
-                suggestCurrent
+                suggestCurrent,
+                recentSearchLoad
             );
         }
     }
@@ -427,6 +453,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 EMW_RATE = "ewm_rate";
     }
 
     @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;
+    }
+}

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

@@ -10,6 +10,7 @@
 package org.elasticsearch.index.search.stats;
 
 import org.elasticsearch.common.metrics.CounterMetric;
+import org.elasticsearch.common.metrics.ExponentiallyWeightedMovingRate;
 import org.elasticsearch.common.metrics.MeanMetric;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.util.CollectionUtils;
@@ -26,9 +27,15 @@
 
 public final class ShardSearchStats implements SearchOperationListener {
 
-    private final StatsHolder totalStats = new StatsHolder();
+    private final StatsHolder totalStats;
     private final CounterMetric openContexts = new CounterMetric();
     private volatile Map<String, StatsHolder> groupsStats = emptyMap();
+    private final SearchStatsSettings searchStatsSettings;
+
+    public ShardSearchStats(SearchStatsSettings searchStatsSettings) {
+        this.searchStatsSettings = searchStatsSettings;
+        this.totalStats = new StatsHolder(searchStatsSettings);
+    }
 
     /**
      * Returns the stats, including group specific stats. If the groups are null/0 length, then nothing
@@ -78,9 +85,11 @@ public void onFailedQueryPhase(SearchContext searchContext) {
     @Override
     public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
         computeStats(searchContext, searchContext.hasOnlySuggest() ? statsHolder -> {
+            statsHolder.recentSearchLoad.addIncrement(tookInNanos, System.nanoTime());
             statsHolder.suggestMetric.inc(tookInNanos);
             statsHolder.suggestCurrent.dec();
         } : statsHolder -> {
+            statsHolder.recentSearchLoad.addIncrement(tookInNanos, System.nanoTime());
             statsHolder.queryMetric.inc(tookInNanos);
             statsHolder.queryCurrent.dec();
         });
@@ -102,6 +111,7 @@ public void onFailedFetchPhase(SearchContext searchContext) {
     @Override
     public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
         computeStats(searchContext, statsHolder -> {
+            statsHolder.recentSearchLoad.addIncrement(tookInNanos, System.nanoTime());
             statsHolder.fetchMetric.inc(tookInNanos);
             statsHolder.fetchCurrent.dec();
         });
@@ -123,7 +133,7 @@ private StatsHolder groupStats(String group) {
             synchronized (this) {
                 stats = groupsStats.get(group);
                 if (stats == null) {
-                    stats = new StatsHolder();
+                    stats = new StatsHolder(searchStatsSettings);
                     groupsStats = Maps.copyMapWithAddedEntry(groupsStats, group, stats);
                 }
             }
@@ -173,6 +183,13 @@ static final class StatsHolder {
         final CounterMetric queryFailure = new CounterMetric();
         final CounterMetric fetchFailure = new CounterMetric();
 
+        final ExponentiallyWeightedMovingRate recentSearchLoad;
+
+        StatsHolder(SearchStatsSettings searchStatsSettings) {
+            double lambdaInInverseNanos = Math.log(2.0) / searchStatsSettings.getRecentReadLoadHalfLifeForNewShards().nanos();
+            this.recentSearchLoad = new ExponentiallyWeightedMovingRate(lambdaInInverseNanos, System.nanoTime());
+        }
+
         SearchStats.Stats stats() {
             return new SearchStats.Stats(
                 queryMetric.count(),
@@ -188,7 +205,8 @@ SearchStats.Stats stats() {
                 scrollCurrent.count(),
                 suggestMetric.count(),
                 TimeUnit.NANOSECONDS.toMillis(suggestMetric.sum()),
-                suggestCurrent.count()
+                suggestCurrent.count(),
+                recentSearchLoad.getRate(System.nanoTime())
             );
         }
     }