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;
@@ -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,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;
@@ -170,6 +172,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 +210,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 +297,7 @@ public IndexService(
             this.retentionLeaseSyncTask = new AsyncRetentionLeaseSyncTask(this);
         }
         this.indexingStatsSettings = indexingStatsSettings;
+        this.searchStatsSettings = searchStatsSettings;
         updateFsyncTaskIfNecessary();
     }
 
@@ -462,7 +467,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);
         /*
@@ -583,7 +589,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,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 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/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,33 @@
+/*
+ * 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 rate,
+ * 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  {@code 0.0}
+     * This can be used as a fallback or default when no actual load tracking is required.
+     */
+    ShardSearchLoadRateService NOOP = (stats) -> 0.0;
+
+    /**
+     * 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 {@code double} representing the calculated EWMR on the shard
+     */
+    double getSearchLoadRate(SearchStats.Stats stats);
+}

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 double getSearchLoadRate() {
+        return shardSearchLoadRateService.getSearchLoadRate(searchStats.stats().getTotal());
+    }
+
     public FieldUsageStats fieldUsageStats(String... fields) {
         return fieldUsageTracker.stats(fields);
     }

server/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -118,6 +118,8 @@
 import org.elasticsearch.index.recovery.RecoveryStats;
 import org.elasticsearch.index.refresh.RefreshStats;
 import org.elasticsearch.index.search.stats.SearchStats;
+import org.elasticsearch.index.search.stats.SearchStatsSettings;
+import org.elasticsearch.index.search.stats.ShardSearchLoadRateProvider;
 import org.elasticsearch.index.seqno.RetentionLeaseStats;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
 import org.elasticsearch.index.seqno.SeqNoStats;
@@ -281,6 +283,7 @@ public class IndicesService extends AbstractLifecycleComponent
     private final QueryRewriteInterceptor queryRewriteInterceptor;
     final SlowLogFieldProvider slowLogFieldProvider; // pkg-private for testingå
     private final IndexingStatsSettings indexStatsSettings;
+    private final SearchStatsSettings searchStatsSettings;
 
     @Override
     protected void doStart() {
@@ -406,6 +409,7 @@ public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, lon
         this.searchOperationListeners = builder.searchOperationListener;
         this.slowLogFieldProvider = builder.slowLogFieldProvider;
         this.indexStatsSettings = new IndexingStatsSettings(clusterService.getClusterSettings());
+        this.searchStatsSettings = new SearchStatsSettings(clusterService.getClusterSettings());
     }
 
     private static final String DANGLING_INDICES_UPDATE_THREAD_NAME = "DanglingIndices#updateTask";
@@ -795,7 +799,8 @@ private synchronized IndexService createIndexService(
             slowLogFieldProvider,
             mapperMetrics,
             searchOperationListeners,
-            indexStatsSettings
+            indexStatsSettings,
+            searchStatsSettings
         );
         for (IndexingOperationListener operationListener : indexingOperationListeners) {
             indexModule.addIndexOperationListener(operationListener);
@@ -893,7 +898,8 @@ public synchronized MapperService createIndexMapperServiceForValidation(IndexMet
             slowLogFieldProvider,
             mapperMetrics,
             searchOperationListeners,
-            indexStatsSettings
+            indexStatsSettings,
+            searchStatsSettings
         );
         pluginsService.forEach(p -> p.onIndexModule(indexModule));
         return indexModule.newIndexMapperService(clusterService, parserConfig, mapperRegistry, scriptService);
@@ -954,7 +960,18 @@ public void createShard(
         IndexService indexService = indexService(shardRouting.index());
         assert indexService != null;
         RecoveryState recoveryState = indexService.createRecoveryState(shardRouting, targetNode, sourceNode);
-        IndexShard indexShard = indexService.createShard(shardRouting, globalCheckpointSyncer, retentionLeaseSyncer);
+
+        ShardSearchLoadRateProvider shardSearchLoadRateProvider = pluginsService.loadSingletonServiceProvider(
+            ShardSearchLoadRateProvider.class,
+            () -> ShardSearchLoadRateProvider.DEFAULT
+        );
+
+        IndexShard indexShard = indexService.createShard(
+            shardRouting,
+            globalCheckpointSyncer,
+            retentionLeaseSyncer,
+            shardSearchLoadRateProvider
+        );
         indexShard.addShardFailureCallback(onShardFailure);
         indexShard.startRecovery(
             recoveryState,