Calculate recent write load in indexing stats

This uses the recently-added `ExponentiallyWeightedMovingRate` class
to calculate a write load which favours more recent load and include
this alongside the existing unweighted all-time write load in
`IndexingStats.Stats`.

As of this change, the new load metric is not used anywhere, although
it can be retrieved with the index stats or node stats APIs.

Author: PeteGillinElastic

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/DataStreamAutoshardingIT.java

@@ -527,7 +527,7 @@ private static ShardStats getShardStats(IndexMetadata indexMeta, int shardIndex,
         CommonStats stats = new CommonStats();
         stats.docs = new DocsStats(100, 0, randomByteSizeValue().getBytes());
         stats.store = new StoreStats();
-        stats.indexing = new IndexingStats(new IndexingStats.Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, false, 1, targetWriteLoad, 1));
+        stats.indexing = new IndexingStats(new IndexingStats.Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, false, 1, targetWriteLoad, 1, 0.123));
         return new ShardStats(shardRouting, new ShardPath(false, path, path, shardId), stats, null, null, null, false, 0);
     }
 

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

@@ -31,6 +31,7 @@
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.lucene.uid.Versions;
+import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -635,7 +636,8 @@ public static final IndexShard newIndexShard(
             IndexModule.DEFAULT_SNAPSHOT_COMMIT_SUPPLIER,
             System::nanoTime,
             null,
-            MapperMetrics.NOOP
+            MapperMetrics.NOOP,
+            new IndexingStatsSettings(ClusterSettings.createBuiltInClusterSettings())
         );
     }
 

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

@@ -186,6 +186,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INFERENCE_CONTEXT = def(9_028_0_00);
     public static final TransportVersion ML_INFERENCE_DEEPSEEK = def(9_029_00_0);
     public static final TransportVersion ESQL_FAILURE_FROM_REMOTE = def(9_030_00_0);
+    public static final TransportVersion INDEXING_STATS_INCLUDES_RECENT_WRITE_LOAD = def(9_031_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/common/metrics/ExponentiallyWeightedMovingRate.java

@@ -11,6 +11,7 @@
 
 import static java.lang.Math.exp;
 import static java.lang.Math.expm1;
+import static java.lang.Math.log;
 
 /**
  * Implements a version of an exponentially weighted moving rate (EWMR). This is a calculation over a finite time series of increments to
@@ -41,7 +42,8 @@ public class ExponentiallyWeightedMovingRate {
     private final double lambda;
     private final long startTime;
     private double rate;
-    long lastTime;
+    private long lastTime;
+    private boolean waitingForFirstIncrement;
 
     /**
      * Constructor.
@@ -57,14 +59,12 @@ public ExponentiallyWeightedMovingRate(double lambda, long startTime) {
         if (lambda < 0.0) {
             throw new IllegalArgumentException("lambda must be non-negative but was " + lambda);
         }
-        if (startTime <= 0.0) {
-            throw new IllegalArgumentException("startTime must be non-negative but was " + startTime);
-        }
         synchronized (this) {
             this.lambda = lambda;
             this.rate = Double.NaN; // should never be used
             this.startTime = startTime;
-            this.lastTime = 0; // after an increment, this must be positive, so a zero value indicates we're waiting for the first
+            this.lastTime = 0; // should never be used
+            this.waitingForFirstIncrement = true;
         }
     }
 
@@ -80,7 +80,7 @@ public ExponentiallyWeightedMovingRate(double lambda, long startTime) {
      */
     public double getRate(long time) {
         synchronized (this) {
-            if (lastTime == 0) { // indicates that no increment has happened yet
+            if (waitingForFirstIncrement) {
                 return 0.0;
             } else if (time <= lastTime) {
                 return rate;
@@ -104,6 +104,9 @@ public double getRate(long time) {
      * instance. It is only non-static because it uses this instance's {@code lambda} and {@code startTime}.
      */
     public double calculateRateSince(long currentTime, double currentRate, long oldTime, double oldRate) {
+        if (oldTime < startTime) {
+            oldTime = startTime;
+        }
         if (currentTime <= oldTime) {
             return 0.0;
         }
@@ -127,12 +130,13 @@ public double calculateRateSince(long currentTime, double currentRate, long oldT
      */
     public void addIncrement(double increment, long time) {
         synchronized (this) {
-            if (lastTime == 0) { // indicates that this is the first increment
+            if (waitingForFirstIncrement) {
                 if (time <= startTime) {
                     time = startTime + 1;
                 }
                 // This is the formula for R(t_1) given in subsection 2.6 of the document referenced above:
                 rate = increment / expHelper(time - startTime);
+                waitingForFirstIncrement = false;
             } else {
                 if (time < lastTime) {
                     time = lastTime;
@@ -165,4 +169,12 @@ private double expHelper(double time) {
             return time * (1.0 - 0.5 * lambdaTime);
         }
     }
+
+    /**
+     * Returns the configured half-life of this instance. The units are the same as all other times in API calls, and the inverse of the
+     * units used for the {@code lambda} constructor parameter. If {@code lambda} is {@code 0.0}, returns {@link Double#POSITIVE_INFINITY}.
+     */
+    public double getHalfLife() {
+        return log(2.0) / lambda;
+    }
 }

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

@@ -88,6 +88,7 @@
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.IndexingPressure;
 import org.elasticsearch.index.MergePolicyConfig;
+import org.elasticsearch.index.shard.IndexingStatsSettings;
 import org.elasticsearch.indices.IndexingMemoryController;
 import org.elasticsearch.indices.IndicesQueryCache;
 import org.elasticsearch.indices.IndicesRequestCache;
@@ -629,6 +630,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         DataStreamGlobalRetentionSettings.DATA_STREAMS_DEFAULT_RETENTION_SETTING,
         DataStreamGlobalRetentionSettings.DATA_STREAMS_MAX_RETENTION_SETTING,
         ShardsAvailabilityHealthIndicatorService.REPLICA_UNASSIGNED_BUFFER_TIME,
-        DataStream.isFailureStoreFeatureFlagEnabled() ? DataStreamFailureStoreSettings.DATA_STREAM_FAILURE_STORED_ENABLED_SETTING : null
+        DataStream.isFailureStoreFeatureFlagEnabled() ? DataStreamFailureStoreSettings.DATA_STREAM_FAILURE_STORED_ENABLED_SETTING : null,
+        IndexingStatsSettings.RECENT_WRITE_LOAD_HALF_LIFE_SETTING
     ).filter(Objects::nonNull).collect(toSet());
 }

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

@@ -49,6 +49,7 @@
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.shard.IndexEventListener;
 import org.elasticsearch.index.shard.IndexingOperationListener;
+import org.elasticsearch.index.shard.IndexingStatsSettings;
 import org.elasticsearch.index.shard.SearchOperationListener;
 import org.elasticsearch.index.shard.ShardPath;
 import org.elasticsearch.index.similarity.SimilarityService;
@@ -176,6 +177,7 @@ public interface DirectoryWrapper {
     private final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories;
     private final SetOnce<Engine.IndexCommitListener> indexCommitListener = new SetOnce<>();
     private final MapperMetrics mapperMetrics;
+    private final IndexingStatsSettings indexingStatsSettings;
 
     /**
      * Construct the index module for the index with the specified index settings. The index module contains extension points for plugins
@@ -196,7 +198,8 @@ public IndexModule(
         final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories,
         final SlowLogFieldProvider slowLogFieldProvider,
         final MapperMetrics mapperMetrics,
-        final List<SearchOperationListener> searchOperationListeners
+        final List<SearchOperationListener> searchOperationListeners,
+        final IndexingStatsSettings indexingStatsSettings
     ) {
         this.indexSettings = indexSettings;
         this.analysisRegistry = analysisRegistry;
@@ -211,6 +214,7 @@ public IndexModule(
         this.expressionResolver = expressionResolver;
         this.recoveryStateFactories = recoveryStateFactories;
         this.mapperMetrics = mapperMetrics;
+        this.indexingStatsSettings = indexingStatsSettings;
     }
 
     /**
@@ -544,7 +548,8 @@ public IndexService newIndexService(
                 snapshotCommitSupplier,
                 indexCommitListener.get(),
                 mapperMetrics,
-                queryRewriteInterceptor
+                queryRewriteInterceptor,
+                indexingStatsSettings
             );
             success = true;
             return indexService;

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

@@ -72,6 +72,7 @@
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.IndexShardClosedException;
 import org.elasticsearch.index.shard.IndexingOperationListener;
+import org.elasticsearch.index.shard.IndexingStatsSettings;
 import org.elasticsearch.index.shard.SearchOperationListener;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.shard.ShardNotFoundException;
@@ -164,6 +165,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final ValuesSourceRegistry valuesSourceRegistry;
     private final MapperMetrics mapperMetrics;
     private final QueryRewriteInterceptor queryRewriteInterceptor;
+    private final IndexingStatsSettings indexingStatsSettings;
 
     @SuppressWarnings("this-escape")
     public IndexService(
@@ -199,7 +201,8 @@ public IndexService(
         IndexStorePlugin.SnapshotCommitSupplier snapshotCommitSupplier,
         Engine.IndexCommitListener indexCommitListener,
         MapperMetrics mapperMetrics,
-        QueryRewriteInterceptor queryRewriteInterceptor
+        QueryRewriteInterceptor queryRewriteInterceptor,
+        IndexingStatsSettings indexingStatsSettings
     ) {
         super(indexSettings);
         assert indexCreationContext != IndexCreationContext.RELOAD_ANALYZERS
@@ -283,6 +286,7 @@ public IndexService(
             this.globalCheckpointTask = new AsyncGlobalCheckpointTask(this);
             this.retentionLeaseSyncTask = new AsyncRetentionLeaseSyncTask(this);
         }
+        this.indexingStatsSettings = indexingStatsSettings;
         updateFsyncTaskIfNecessary();
     }
 
@@ -566,7 +570,8 @@ public synchronized IndexShard createShard(
                 snapshotCommitSupplier,
                 System::nanoTime,
                 indexCommitListener,
-                mapperMetrics
+                mapperMetrics,
+                indexingStatsSettings
             );
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);

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

@@ -297,6 +297,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     private final LongSupplier relativeTimeInNanosSupplier;
     private volatile long startedRelativeTimeInNanos;
     private volatile long indexingTimeBeforeShardStartedInNanos;
+    private volatile double recentIndexingLoadAtShardStarted;
     private final SubscribableListener<Void> waitForEngineOrClosedShardListeners = new SubscribableListener<>();
 
     // the translog keeps track of the GCP, but unpromotable shards have no translog so we need to track the GCP here instead
@@ -326,7 +327,8 @@ public IndexShard(
         final IndexStorePlugin.SnapshotCommitSupplier snapshotCommitSupplier,
         final LongSupplier relativeTimeInNanosSupplier,
         final Engine.IndexCommitListener indexCommitListener,
-        final MapperMetrics mapperMetrics
+        final MapperMetrics mapperMetrics,
+        final IndexingStatsSettings indexingStatsSettings
     ) throws IOException {
         super(shardRouting.shardId(), indexSettings);
         assert shardRouting.initializing();
@@ -344,7 +346,7 @@ public IndexShard(
         this.threadPool = threadPool;
         this.mapperService = mapperService;
         this.indexCache = indexCache;
-        this.internalIndexingStats = new InternalIndexingStats();
+        this.internalIndexingStats = new InternalIndexingStats(relativeTimeInNanosSupplier, indexingStatsSettings);
         var indexingFailuresDebugListener = new IndexingFailuresDebugListener(this);
         this.indexingOperationListeners = new IndexingOperationListener.CompositeListener(
             CollectionUtils.appendToCopyNoNullElements(listeners, internalIndexingStats, indexingFailuresDebugListener),
@@ -552,6 +554,7 @@ public void updateShardState(
                 changeState(IndexShardState.STARTED, "global state is [" + newRouting.state() + "]");
                 startedRelativeTimeInNanos = getRelativeTimeInNanos();
                 indexingTimeBeforeShardStartedInNanos = internalIndexingStats.totalIndexingTimeInNanos();
+                recentIndexingLoadAtShardStarted = internalIndexingStats.recentIndexingLoad(startedRelativeTimeInNanos);
             } else if (currentRouting.primary()
                 && currentRouting.relocating()
                 && replicationTracker.isRelocated()
@@ -1361,11 +1364,14 @@ public IndexingStats indexingStats() {
             throttleTimeInMillis = engine.getIndexThrottleTimeInMillis();
         }
 
+        long currentTimeInNanos = getRelativeTimeInNanos();
         return internalIndexingStats.stats(
             throttled,
             throttleTimeInMillis,
             indexingTimeBeforeShardStartedInNanos,
-            getRelativeTimeInNanos() - startedRelativeTimeInNanos
+            currentTimeInNanos - startedRelativeTimeInNanos,
+            currentTimeInNanos,
+            recentIndexingLoadAtShardStarted
         );
     }