Enable Shard Search-load rate metrics for injection

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

@@ -53,6 +53,8 @@
 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.search.stats.ShardSearchLoadRateProvider;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.translog.TestTranslog;
@@ -638,7 +640,9 @@ public static final IndexShard newIndexShard(
             System::nanoTime,
             null,
             MapperMetrics.NOOP,
-            new IndexingStatsSettings(ClusterSettings.createBuiltInClusterSettings())
+            new IndexingStatsSettings(ClusterSettings.createBuiltInClusterSettings()),
+            new SearchStatsSettings(ClusterSettings.createBuiltInClusterSettings()),
+            ShardSearchLoadRateProvider.DEFAULT
         );
     }
 

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;
@@ -635,6 +636,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,8 @@
 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.search.stats.ShardSearchLoadRateProvider;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
 import org.elasticsearch.index.shard.GlobalCheckpointSyncer;
 import org.elasticsearch.index.shard.IndexEventListener;
@@ -169,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(
@@ -206,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
@@ -292,6 +296,7 @@ public IndexService(
             this.retentionLeaseSyncTask = new AsyncRetentionLeaseSyncTask(this);
         }
         this.indexingStatsSettings = indexingStatsSettings;
+        this.searchStatsSettings = searchStatsSettings;
         updateFsyncTaskIfNecessary();
     }
 
@@ -461,7 +466,8 @@ private long getAvgShardSizeInBytes() throws IOException {
     public synchronized IndexShard createShard(
         final ShardRouting routing,
         final GlobalCheckpointSyncer globalCheckpointSyncer,
-        final RetentionLeaseSyncer retentionLeaseSyncer
+        final RetentionLeaseSyncer retentionLeaseSyncer,
+        final ShardSearchLoadRateProvider shardSearchLoadRateProvider
     ) throws IOException {
         Objects.requireNonNull(retentionLeaseSyncer);
         /*
@@ -577,7 +583,9 @@ public synchronized IndexShard createShard(
                 System::nanoTime,
                 indexCommitListener,
                 mapperMetrics,
-                indexingStatsSettings
+                indexingStatsSettings,
+                searchStatsSettings,
+                shardSearchLoadRateProvider
             );
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);

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

@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Container for cluster settings
+ */
+public class SearchStatsSettings {
+
+    public static final TimeValue RECENT_READ_LOAD_HALF_LIFE_DEFAULT = TimeValue.timeValueSeconds(15); // TODO this is set to seconds for
+                                                                                                       // debugging
+    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 final AtomicReference<TimeValue> recentReadLoadHalfLifeForNewShards = new AtomicReference<>(
+        RECENT_READ_LOAD_HALF_LIFE_SETTING.getDefault(Settings.EMPTY)
+    );
+
+    public SearchStatsSettings(ClusterSettings clusterSettings) {
+        clusterSettings.initializeAndWatch(RECENT_READ_LOAD_HALF_LIFE_SETTING, recentReadLoadHalfLifeForNewShards::set);
+    }
+
+    public TimeValue getRecentReadLoadHalfLifeForNewShards() {
+        return recentReadLoadHalfLifeForNewShards.get();
+    }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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 java.util.function.Supplier;
+
+/**
+ * A factory interface for creating instances of {@link ShardSearchLoadRateService} using configurable settings
+ * and a time source.
+ * <p>
+ * This abstraction allows for flexible instantiation of load rate tracking services, potentially with
+ * different strategies based on provided settings or runtime context.
+ */
+public interface ShardSearchLoadRateProvider {
+
+    /**
+     * The default no-op implementation of the provider, which always returns {@link ShardSearchLoadRateService#NOOP}.
+     * <p>
+     * Useful as a fallback or when search load rate tracking is disabled or not required.
+     */
+    ShardSearchLoadRateProvider DEFAULT = (settings, timeProvider) -> ShardSearchLoadRateService.NOOP;
+
+    /**
+     * Creates a new instance of {@link ShardSearchLoadRateService} using the given settings and time provider.
+     *
+     * @param settings     the search statistics configuration settings that may influence the service behavior
+     * @param timeProvider a supplier of the current time, typically in milliseconds or nanoseconds
+     * @return a {@code ShardSearchLoadRateService} instance configured with the provided context
+     */
+    ShardSearchLoadRateService create(SearchStatsSettings settings, Supplier<Long> timeProvider);
+}

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

@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+/**
+ * Service interface for estimating the search load rate of a shard using provided search statistics.
+ * <p>
+ * Implementations may apply various heuristics or models, such as exponentially weighted moving averages,
+ * to track and estimate the current load on a shard based on its search activity.
+ */
+public interface ShardSearchLoadRateService {
+
+    /**
+     * A no-op implementation of {@code ShardSearchLoadRateService} that always returns {@link SearchLoadRate#NO_OP}.
+     * This can be used as a fallback or default when no actual load tracking is required.
+     */
+    ShardSearchLoadRateService NOOP = (stats) -> SearchLoadRate.NO_OP;
+
+    /**
+     * Computes the search load rate based on the provided shard-level search statistics.
+     *
+     * @param stats the search statistics for the shard, typically including metrics like query count, latency, etc.
+     * @return the {@link SearchLoadRate} representing the current estimated load on the shard
+     */
+    SearchLoadRate getSearchLoadRate(SearchStats.Stats stats);
+
+    /**
+     * Represents the search load rate as computed over time using an exponentially weighted moving average (EWM).
+     * <p>
+     * This record captures the timing of the last update, the delta since the last observation,
+     * and the computed rate itself.
+     *
+     * @param lastTrackedTime the timestamp (e.g., in milliseconds or nanoseconds) of the last update
+     * @param delta the elapsed time since the previous update
+     * @param ewmRate the current exponentially weighted moving average rate
+     */
+    record SearchLoadRate(long lastTrackedTime, long delta, double ewmRate) {
+
+        /**
+         * A static no-op instance representing a default or zeroed state of {@code SearchLoadRate}.
+         * All numeric values are initialized to zero.
+         */
+        public static final SearchLoadRate NO_OP = new SearchLoadRate(0, 0, 0.0);
+    }
+}

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -117,7 +117,10 @@
 import org.elasticsearch.index.refresh.RefreshStats;
 import org.elasticsearch.index.search.stats.FieldUsageStats;
 import org.elasticsearch.index.search.stats.SearchStats;
+import org.elasticsearch.index.search.stats.SearchStatsSettings;
 import org.elasticsearch.index.search.stats.ShardFieldUsageTracker;
+import org.elasticsearch.index.search.stats.ShardSearchLoadRateProvider;
+import org.elasticsearch.index.search.stats.ShardSearchLoadRateService;
 import org.elasticsearch.index.search.stats.ShardSearchStats;
 import org.elasticsearch.index.seqno.ReplicationTracker;
 import org.elasticsearch.index.seqno.RetentionLease;
@@ -204,6 +207,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     private final Store store;
     private final InternalIndexingStats internalIndexingStats;
     private final ShardSearchStats searchStats = new ShardSearchStats();
+    private final ShardSearchLoadRateService shardSearchLoadRateService;
     private final ShardFieldUsageTracker fieldUsageTracker;
     private final String shardUuid = UUIDs.randomBase64UUID();
     private final long shardCreationTime;
@@ -341,7 +345,9 @@ public IndexShard(
         final LongSupplier relativeTimeInNanosSupplier,
         final Engine.IndexCommitListener indexCommitListener,
         final MapperMetrics mapperMetrics,
-        final IndexingStatsSettings indexingStatsSettings
+        final IndexingStatsSettings indexingStatsSettings,
+        final SearchStatsSettings searchStatsSettings,
+        final ShardSearchLoadRateProvider shardSearchLoadRateProvider
     ) throws IOException {
         super(shardRouting.shardId(), indexSettings);
         assert shardRouting.initializing();
@@ -366,6 +372,7 @@ public IndexShard(
             CollectionUtils.appendToCopyNoNullElements(listeners, internalIndexingStats, indexingFailuresDebugListener),
             logger
         );
+        this.shardSearchLoadRateService = shardSearchLoadRateProvider.create(searchStatsSettings, System::currentTimeMillis);
         this.bulkOperationListener = new ShardBulkStats();
         this.globalCheckpointSyncer = globalCheckpointSyncer;
         this.retentionLeaseSyncer = Objects.requireNonNull(retentionLeaseSyncer);
@@ -1414,6 +1421,13 @@ public SearchStats searchStats(String... groups) {
         return searchStats.stats(groups);
     }
 
+    /**
+     * Returns the search load rate stats for this shard.
+     */
+    public ShardSearchLoadRateService.SearchLoadRate getSearchLoadRate() {
+        return shardSearchLoadRateService.getSearchLoadRate(searchStats.stats().getTotal());
+    }
+
     public FieldUsageStats fieldUsageStats(String... fields) {
         return fieldUsageTracker.stats(fields);
     }