Account for time taken to write index buffers in IndexingMemoryController

docs/changelog/126786.yaml

@@ -0,0 +1,5 @@
+pr: 126786
+summary: Account for time taken to write index buffers in `IndexingMemoryController`
+area: Distributed
+type: enhancement
+issues: []

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

@@ -527,7 +527,9 @@ 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, 0.123, 0.234));
+        stats.indexing = new IndexingStats(
+            new IndexingStats.Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, false, 1, targetWriteLoad, targetWriteLoad, 1, 0.123, 0.234)
+        );
         return new ShardStats(shardRouting, new ShardPath(false, path, path, shardId), stats, null, null, null, false, 0);
     }
 

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

@@ -230,6 +230,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_DOCUMENTS_FOUND_AND_VALUES_LOADED = def(9_056_0_00);
     public static final TransportVersion BATCHED_QUERY_EXECUTION_DELAYABLE_WRITABLE = def(9_057_0_00);
     public static final TransportVersion SEARCH_INCREMENTAL_TOP_DOCS_NULL = def(9_058_0_00);
+    public static final TransportVersion WRITE_LOAD_INCLUDES_BUFFER_WRITES = def(9_059_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -308,6 +308,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     private final LongSupplier relativeTimeInNanosSupplier;
     private volatile long startedRelativeTimeInNanos = -1L; // use -1 to indicate this has not yet been set to its true value
     private volatile long indexingTimeBeforeShardStartedInNanos;
+    private volatile long indexingLoadBeforeShardStartedInNanos;
     private volatile double recentIndexingLoadAtShardStarted;
     private final SubscribableListener<Void> waitForEngineOrClosedShardListeners = new SubscribableListener<>();
 
@@ -569,6 +570,7 @@ public void updateShardState(
                 // unlikely case that getRelativeTimeInNanos() returns exactly -1, we advance by 1ns to avoid that special value.
                 startedRelativeTimeInNanos = (relativeTimeInNanos != -1L) ? relativeTimeInNanos : 0L;
                 indexingTimeBeforeShardStartedInNanos = internalIndexingStats.totalIndexingTimeInNanos();
+                indexingLoadBeforeShardStartedInNanos = internalIndexingStats.totalIndexingLoadInNanos();
                 recentIndexingLoadAtShardStarted = internalIndexingStats.recentIndexingLoad(startedRelativeTimeInNanos);
             } else if (currentRouting.primary()
                 && currentRouting.relocating()
@@ -1401,6 +1403,7 @@ public IndexingStats indexingStats() {
             throttled,
             throttleTimeInMillis,
             indexingTimeBeforeShardStartedInNanos,
+            indexingLoadBeforeShardStartedInNanos,
             timeSinceShardStartedInNanos,
             currentTimeInNanos,
             recentIndexingLoadAtShardStarted
@@ -3235,6 +3238,16 @@ public void noopUpdate() {
         internalIndexingStats.noopUpdate();
     }
 
+    /**
+     * Increment relevant stats when indexing buffers are written to disk using indexing threads,
+     * in order to apply back-pressure on indexing.
+     * @param took  time it took to write the index buffers for this shard
+     * @see org.elasticsearch.indices.IndexingMemoryController
+     */
+    public void writeIndexBuffersOnIndexThreads(long took) {
+        internalIndexingStats.writeIndexingBuffersTime(took);
+    }
+
     public void maybeCheckIndex() {
         recoveryState.setStage(RecoveryState.Stage.VERIFY_INDEX);
         if (Booleans.isTrue(checkIndexOnStartup) || "checksum".equals(checkIndexOnStartup)) {

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

@@ -27,6 +27,7 @@
 
 import static org.elasticsearch.TransportVersions.INDEXING_STATS_INCLUDES_RECENT_WRITE_LOAD;
 import static org.elasticsearch.TransportVersions.INDEX_STATS_AND_METADATA_INCLUDE_PEAK_WRITE_LOAD;
+import static org.elasticsearch.TransportVersions.WRITE_LOAD_INCLUDES_BUFFER_WRITES;
 
 public class IndexingStats implements Writeable, ToXContentFragment {
 
@@ -45,6 +46,7 @@ public static class Stats implements Writeable, ToXContentFragment {
         private long throttleTimeInMillis;
         private boolean isThrottled;
         private long totalIndexingTimeSinceShardStartedInNanos;
+        private long totalIndexingLoadSinceShardStartedInNanos;
         private long totalActiveTimeInNanos;
         private double recentIndexingLoad;
         private double peakIndexingLoad;
@@ -87,6 +89,13 @@ public Stats(StreamInput in) throws IOException {
                     ? (double) totalIndexingTimeSinceShardStartedInNanos / totalActiveTimeInNanos
                     : 0;
             }
+            if (in.getTransportVersion().onOrAfter(WRITE_LOAD_INCLUDES_BUFFER_WRITES)) {
+                totalIndexingLoadSinceShardStartedInNanos = in.readLong();
+            } else {
+                // When getting stats from an older version which doesn't have the more accurate indexing load, better to fall back to the
+                // indexing time, rather that assuming zero load:
+                totalIndexingLoadSinceShardStartedInNanos = totalActiveTimeInNanos > 0 ? totalIndexingTimeSinceShardStartedInNanos : 0;
+            }
         }
 
         public Stats(
@@ -102,6 +111,7 @@ public Stats(
             boolean isThrottled,
             long throttleTimeInMillis,
             long totalIndexingTimeSinceShardStartedInNanos,
+            long totalIndexingLoadSinceShardStartedInNanos,
             long totalActiveTimeInNanos,
             double recentIndexingLoad,
             double peakIndexingLoad
@@ -119,6 +129,7 @@ public Stats(
             this.throttleTimeInMillis = throttleTimeInMillis;
             // We store the raw unweighted write load values in order to avoid losing precision when we combine the shard stats
             this.totalIndexingTimeSinceShardStartedInNanos = totalIndexingTimeSinceShardStartedInNanos;
+            this.totalIndexingLoadSinceShardStartedInNanos = totalIndexingLoadSinceShardStartedInNanos;
             this.totalActiveTimeInNanos = totalActiveTimeInNanos;
             // We store the weighted write load as a double because the calculation is inherently floating point
             this.recentIndexingLoad = recentIndexingLoad;
@@ -141,8 +152,9 @@ public void add(Stats stats) {
             if (isThrottled != stats.isThrottled) {
                 isThrottled = true; // When combining if one is throttled set result to throttled.
             }
-            // N.B. getWriteLoad() returns the ratio of these sums, which is the average of the ratios weighted by active time:
             totalIndexingTimeSinceShardStartedInNanos += stats.totalIndexingTimeSinceShardStartedInNanos;
+            // N.B. getWriteLoad() returns the ratio of these sums, which is the average of the ratios weighted by active time:
+            totalIndexingLoadSinceShardStartedInNanos += stats.totalIndexingLoadSinceShardStartedInNanos;
             totalActiveTimeInNanos += stats.totalActiveTimeInNanos;
             // We want getRecentWriteLoad() and getPeakWriteLoad() for the aggregated stats to also be the average weighted by active time,
             // so we use the updating formula for a weighted mean:
@@ -237,7 +249,7 @@ public long getNoopUpdateCount() {
          * the elapsed time for each shard.
          */
         public double getWriteLoad() {
-            return totalActiveTimeInNanos > 0 ? (double) totalIndexingTimeSinceShardStartedInNanos / totalActiveTimeInNanos : 0;
+            return totalActiveTimeInNanos > 0 ? (double) totalIndexingLoadSinceShardStartedInNanos / totalActiveTimeInNanos : 0;
         }
 
         /**
@@ -296,6 +308,9 @@ public void writeTo(StreamOutput out) throws IOException {
             if (out.getTransportVersion().onOrAfter(INDEX_STATS_AND_METADATA_INCLUDE_PEAK_WRITE_LOAD)) {
                 out.writeDouble(peakIndexingLoad);
             }
+            if (out.getTransportVersion().onOrAfter(WRITE_LOAD_INCLUDES_BUFFER_WRITES)) {
+                out.writeLong(totalIndexingLoadSinceShardStartedInNanos);
+            }
         }
 
         @Override
@@ -338,6 +353,7 @@ public boolean equals(Object o) {
                 && isThrottled == that.isThrottled
                 && throttleTimeInMillis == that.throttleTimeInMillis
                 && totalIndexingTimeSinceShardStartedInNanos == that.totalIndexingTimeSinceShardStartedInNanos
+                && totalIndexingLoadSinceShardStartedInNanos == that.totalIndexingLoadSinceShardStartedInNanos
                 && totalActiveTimeInNanos == that.totalActiveTimeInNanos
                 && recentIndexingLoad == that.recentIndexingLoad
                 && peakIndexingLoad == that.peakIndexingLoad;
@@ -358,6 +374,7 @@ public int hashCode() {
                 isThrottled,
                 throttleTimeInMillis,
                 totalIndexingTimeSinceShardStartedInNanos,
+                totalIndexingLoadSinceShardStartedInNanos,
                 totalActiveTimeInNanos
             );
         }

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

@@ -22,6 +22,7 @@
 
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.function.LongSupplier;
 
 import static org.elasticsearch.core.TimeValue.timeValueNanos;
@@ -51,6 +52,7 @@ IndexingStats stats(
         boolean isThrottled,
         long currentThrottleInMillis,
         long indexingTimeBeforeShardStartedInNanos,
+        long indexingLoadBeforeShardStartedInNanos,
         long timeSinceShardStartedInNanos,
         long currentTimeInNanos,
         double recentIndexingLoadAtShardStarted
@@ -59,6 +61,7 @@ IndexingStats stats(
             isThrottled,
             currentThrottleInMillis,
             indexingTimeBeforeShardStartedInNanos,
+            indexingLoadBeforeShardStartedInNanos,
             timeSinceShardStartedInNanos,
             currentTimeInNanos,
             recentIndexingLoadAtShardStarted
@@ -70,6 +73,10 @@ long totalIndexingTimeInNanos() {
         return totalStats.indexMetric.sum();
     }
 
+    long totalIndexingLoadInNanos() {
+        return totalStats.indexMetric.sum() + totalStats.writeIndexingBufferTime.sum();
+    }
+
     /**
      * Returns an exponentially-weighted moving rate which measures the indexing load, favoring more recent load.
      */
@@ -153,10 +160,25 @@ void noopUpdate() {
         totalStats.noopUpdates.inc();
     }
 
+    /**
+     * Increment relevant stats when indexing buffers are written to disk using indexing threads,
+     * in order to apply back-pressure on indexing.
+     * @param took   time taken to write buffers
+     * @see org.elasticsearch.indices.IndexingMemoryController
+     */
+    void writeIndexingBuffersTime(long took) {
+        totalStats.writeIndexingBufferTime.add(took);
+        totalStats.recentIndexMetric.addIncrement(took, relativeTimeInNanosSupplier.getAsLong());
+    }
+
     static class StatsHolder {
         private final MeanMetric indexMetric = new MeanMetric(); // Used for the count and total 'took' time (in ns) of index operations
-        private final ExponentiallyWeightedMovingRate recentIndexMetric; // An EWMR of the total 'took' time of index operations (in ns)
-        private final AtomicReference<Double> peakIndexMetric; // The peak value of the EWMR observed in any stats() call
+        private final LongAdder writeIndexingBufferTime = new LongAdder();
+        // Used for the total time taken to flush indexing buffers to disk (on indexing threads) (in ns)
+        private final ExponentiallyWeightedMovingRate recentIndexMetric;
+        // An EWMR of the total 'took' time of index operations (in ns) plus the writeIndexingBufferTime
+        private final AtomicReference<Double> peakIndexMetric;
+        // The peak value of the EWMR (recentIndexMetric) observed in any stats() call
         private final MeanMetric deleteMetric = new MeanMetric();
         private final CounterMetric indexCurrent = new CounterMetric();
         private final CounterMetric indexFailed = new CounterMetric();
@@ -179,12 +201,15 @@ IndexingStats.Stats stats(
             boolean isThrottled,
             long currentThrottleMillis,
             long indexingTimeBeforeShardStartedInNanos,
+            long indexingLoadBeforeShardStartedInNanos,
             long timeSinceShardStartedInNanos,
             long currentTimeInNanos,
             double recentIndexingLoadAtShardStarted
         ) {
             final long totalIndexingTimeInNanos = indexMetric.sum();
             final long totalIndexingTimeSinceShardStartedInNanos = totalIndexingTimeInNanos - indexingTimeBeforeShardStartedInNanos;
+            final long totalIndexingLoadInNanos = totalIndexingTimeInNanos + writeIndexingBufferTime.sum();
+            final long totalIndexingLoadSinceShardStartedInNanos = totalIndexingLoadInNanos - indexingLoadBeforeShardStartedInNanos;
             final double recentIndexingLoadSinceShardStarted = recentIndexMetric.calculateRateSince(
                 currentTimeInNanos,
                 recentIndexMetric.getRate(currentTimeInNanos),
@@ -218,6 +243,7 @@ IndexingStats.Stats stats(
                 isThrottled,
                 TimeUnit.MILLISECONDS.toMillis(currentThrottleMillis),
                 totalIndexingTimeSinceShardStartedInNanos,
+                totalIndexingLoadSinceShardStartedInNanos,
                 timeSinceShardStartedInNanos,
                 recentIndexingLoadSinceShardStarted,
                 peakIndexingLoad

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

@@ -205,11 +205,17 @@ protected void enqueueWriteIndexingBuffer(IndexShard shard) {
      */
     private boolean writePendingIndexingBuffers() {
         boolean wrotePendingIndexingBuffer = false;
+        long startTime;
+        long took;
         for (IndexShard shard = pendingWriteIndexingBufferQueue.pollFirst(); shard != null; shard = pendingWriteIndexingBufferQueue
             .pollFirst()) {
             // Remove the shard from the set first, so that multiple threads can run writeIndexingBuffer concurrently on the same shard.
             pendingWriteIndexingBufferSet.remove(shard);
+            // Calculate the time taken to write the indexing buffers so it can be accounted for in the index write load
+            startTime = System.nanoTime();
             shard.writeIndexingBuffer();
+            took = System.nanoTime() - startTime;
+            shard.writeIndexBuffersOnIndexThreads(took);
             wrotePendingIndexingBuffer = true;
         }
         return wrotePendingIndexingBuffer;
@@ -258,6 +264,7 @@ private void postOperation(ShardId shardId, Engine.Operation operation, Engine.R
         // be reclaimed rapidly. This has the downside of increasing the latency of _bulk requests though. Lucene does the same thing in
         // DocumentsWriter#postUpdate, flushing a segment because the size limit on the RAM buffer was reached happens on the call to
         // IndexWriter#addDocument.
+
         while (writePendingIndexingBuffers()) {
             // If we just wrote segments, then run the checker again if not already running to check if we released enough memory.
             if (statusChecker.tryRun() == false) {

server/src/test/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStatsTests.java

@@ -593,6 +593,7 @@ private static CommonStats createShardLevelCommonStats() {
             ++iota,
             ++iota,
             ++iota,
+            ++iota,
             ++iota
         );
         indicesCommonStats.getIndexing().add(new IndexingStats(indexingStats));

server/src/test/java/org/elasticsearch/action/datastreams/autosharding/DataStreamAutoShardingServiceTests.java

@@ -1066,6 +1066,7 @@ private static IndexingStats createIndexingStats(double indexingLoad, double rec
                 false,
                 0,
                 totalIndexingTimeSinceShardStartedInNanos,
+                totalIndexingTimeSinceShardStartedInNanos,
                 totalActiveTimeInNanos,
                 recentIndexingLoad,
                 peakIndexingLoad

server/src/test/java/org/elasticsearch/cluster/metadata/IndexMetadataStatsTests.java

@@ -126,6 +126,7 @@ private ShardStats createShardStats(
                     false,
                     0,
                     totalIndexingTimeSinceShardStartedInNanos,
+                    totalIndexingTimeSinceShardStartedInNanos,
                     totalActiveTimeInNanos,
                     0.0,
                     0.0

server/src/test/java/org/elasticsearch/index/shard/IndexingStatsTests.java

@@ -33,6 +33,7 @@ public void testStatsGetWriteLoad() {
             false,
             10,
             1_800_000_000L, // totalIndexingTimeSinceShardStartedInNanos - 1.8sec
+            1_800_000_000L, // totalIndexingLoadSinceShardStartedInNanos - 1.8sec
             3_000_000_000L, // totalActiveTimeInNanos - 3sec
             0.1357,
             0.2468
@@ -55,6 +56,7 @@ public void testStatsAdd_indexCount() {
             false,
             10,
             11,
+            11,
             12,
             0.1357,
             0.2468
@@ -72,6 +74,7 @@ public void testStatsAdd_indexCount() {
             false,
             10,
             11,
+            11,
             12,
             0.1357,
             0.2468
@@ -81,8 +84,8 @@ public void testStatsAdd_indexCount() {
     }
 
     public void testStatsAdd_throttled() {
-        IndexingStats.Stats statsFalse = new IndexingStats.Stats(1, 2, 3, 4, 5, 6, 7, 8, 9, false, 10, 11, 12, 0.1357, 0.2468);
-        IndexingStats.Stats statsTrue = new IndexingStats.Stats(1, 2, 3, 4, 5, 6, 7, 8, 9, true, 10, 11, 12, 0.1357, 0.2468);
+        IndexingStats.Stats statsFalse = new IndexingStats.Stats(1, 2, 3, 4, 5, 6, 7, 8, 9, false, 10, 11, 11, 12, 0.1357, 0.2468);
+        IndexingStats.Stats statsTrue = new IndexingStats.Stats(1, 2, 3, 4, 5, 6, 7, 8, 9, true, 10, 11, 11, 12, 0.1357, 0.2468);
         assertThat(sumOfStats(statsFalse, statsFalse).isThrottled(), is(false));
         assertThat(sumOfStats(statsFalse, statsTrue).isThrottled(), is(true));
         assertThat(sumOfStats(statsTrue, statsFalse).isThrottled(), is(true));
@@ -103,6 +106,7 @@ public void testStatsAdd_writeLoads() {
             false,
             10,
             1_000_000_000L, // totalIndexingTimeSinceShardStartedInNanos - 1sec
+            1_000_000_000L, // totalIndexingLoadSinceShardStartedInNanos - 1sec
             2_000_000_000L, // totalActiveTimeInNanos - 2sec
             0.1357, // recentWriteLoad
             0.3579 // peakWriteLoad
@@ -120,6 +124,7 @@ public void testStatsAdd_writeLoads() {
             false,
             10,
             2_100_000_000L, // totalIndexingTimeSinceShardStartedInNanos - 2.1sec
+            2_100_000_000L, // totalIndexingTimeSinceShardStartedInNanos - 2.1sec
             3_000_000_000L, // totalActiveTimeInNanos - 3sec
             0.2468, // recentWriteLoad
             0.5791 // peakWriteLoad

server/src/test/java/org/elasticsearch/rest/action/cat/RestShardsActionTests.java

@@ -153,6 +153,7 @@ private void mockShardStats(boolean includeCommonStats) {
                             randomNonNegativeLong(),
                             randomNonNegativeLong(),
                             randomNonNegativeLong(),
+                            randomNonNegativeLong(),
                             randomDoubleBetween(0.0, 1.0, true),
                             randomDoubleBetween(0.0, 1.0, true)
                         )

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/indices/IndexStatsMonitoringDocTests.java

@@ -404,6 +404,7 @@ private static CommonStats mockCommonStats() {
             no,
             no,
             no,
+            no,
             no
         );
         commonStats.getIndexing().add(new IndexingStats(indexingStats));

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/indices/IndicesStatsMonitoringDocTests.java

@@ -183,7 +183,7 @@ private CommonStats mockCommonStats() {
         commonStats.getDocs().add(new DocsStats(1L, 0L, randomNonNegativeLong() >> 8)); // >> 8 to avoid overflow - we add these things up
         commonStats.getStore().add(new StoreStats(2L, 0L, 0L));
 
-        final IndexingStats.Stats indexingStats = new IndexingStats.Stats(3L, 4L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, true, 5L, 0, 0, 0.0, 0.0);
+        final IndexingStats.Stats indexingStats = new IndexingStats.Stats(3L, 4L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, true, 5L, 0, 0, 0, 0.0, 0.0);
         commonStats.getIndexing().add(new IndexingStats(indexingStats));
 
         final SearchStats.Stats searchStats = new SearchStats.Stats(6L, 7L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L);

x-pack/plugin/monitoring/src/test/java/org/elasticsearch/xpack/monitoring/collector/node/NodeStatsMonitoringDocTests.java

@@ -350,6 +350,7 @@ private static NodeStats mockNodeStats() {
             no,
             no,
             no,
+            no,
             no
         );
         indicesCommonStats.getIndexing().add(new IndexingStats(indexingStats));