Enhance memory accounting for document expansion and introduce max document size limit (#123543)

This commit improves memory accounting by incorporating document
expansion during shard bulk execution. Additionally, it introduces a new
limit on the maximum document size, which defaults to 5% of the
available heap.

This limit can be configured using the new setting:

indexing_pressure.memory.max_operation_size
These changes help prevent excessive memory consumption and
improve indexing stability.

Closes ES-10777

Author: fcofdez

docs/changelog/123543.yaml

@@ -0,0 +1,6 @@
+pr: 123543
+summary: Enhance memory accounting for document expansion and introduce max document
+  size limit
+area: CRUD
+type: enhancement
+issues: []

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

@@ -112,7 +112,7 @@ public void testShortCircuitFailure() throws Exception {
             IndexingPressure primaryPressure = internalCluster().getInstance(IndexingPressure.class, node);
             long memoryLimit = primaryPressure.stats().getMemoryLimit();
             long primaryRejections = primaryPressure.stats().getPrimaryRejections();
-            try (Releasable ignored = primaryPressure.markPrimaryOperationStarted(10, memoryLimit, false)) {
+            try (Releasable ignored = primaryPressure.validateAndMarkPrimaryOperationStarted(10, memoryLimit, 0, false, false)) {
                 while (primaryPressure.stats().getPrimaryRejections() == primaryRejections) {
                     while (nextRequested.get()) {
                         nextRequested.set(false);

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/IncrementalBulkIT.java

@@ -399,7 +399,7 @@ public void testShortCircuitShardLevelFailure() throws Exception {
         IndexingPressure primaryPressure = internalCluster().getInstance(IndexingPressure.class, node);
         long memoryLimit = primaryPressure.stats().getMemoryLimit();
         long primaryRejections = primaryPressure.stats().getPrimaryRejections();
-        try (Releasable releasable = primaryPressure.markPrimaryOperationStarted(10, memoryLimit, false)) {
+        try (Releasable releasable = primaryPressure.validateAndMarkPrimaryOperationStarted(10, memoryLimit, 0, false, false)) {
             while (primaryPressure.stats().getPrimaryRejections() == primaryRejections) {
                 while (nextRequested.get()) {
                     nextRequested.set(false);
@@ -497,7 +497,7 @@ public void testShortCircuitShardLevelFailureWithIngestNodeHop() throws Exceptio
         assertThat(node, equalTo(dataOnlyNode));
         IndexingPressure primaryPressure = internalCluster().getInstance(IndexingPressure.class, node);
         long memoryLimit = primaryPressure.stats().getMemoryLimit();
-        try (Releasable releasable = primaryPressure.markPrimaryOperationStarted(10, memoryLimit, false)) {
+        try (Releasable releasable = primaryPressure.validateAndMarkPrimaryOperationStarted(10, memoryLimit, 0, false, false)) {
             while (nextRequested.get()) {
                 nextRequested.set(false);
                 refCounted.incRef();

server/src/internalClusterTest/java/org/elasticsearch/index/IndexingPressureIT.java

@@ -90,10 +90,12 @@ private static void assertThatRetentionLeaseSyncCompletesSuccessfully(String pri
      */
     private static Releasable fullyAllocatePrimaryIndexingCapacityOnNode(String targetNode) {
         return internalCluster().getInstance(IndexingPressure.class, targetNode)
-            .markPrimaryOperationStarted(
+            .validateAndMarkPrimaryOperationStarted(
                 1,
                 IndexingPressure.MAX_PRIMARY_BYTES.get(internalCluster().getInstance(Settings.class, targetNode)).getBytes() + 1,
-                true
+                0,
+                true,
+                false
             );
     }
 }

server/src/internalClusterTest/java/org/elasticsearch/index/seqno/RetentionLeaseSyncActionIT.java

@@ -185,6 +185,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion RE_REMOVE_MIN_COMPATIBLE_SHARD_NODE = def(9_021_0_00);
     public static final TransportVersion UNASSIGENEDINFO_RESHARD_ADDED = def(9_022_0_00);
     public static final TransportVersion INCLUDE_INDEX_MODE_IN_GET_DATA_STREAM = def(9_023_0_00);
+    public static final TransportVersion MAX_OPERATION_SIZE_REJECTIONS_ADDED = def(9_024_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -795,6 +795,8 @@ static class IndexPressureStats implements ToXContentFragment {
             long currentReplicaOps = 0;
             long lowWaterMarkSplits = 0;
             long highWaterMarkSplits = 0;
+            long largeOpsRejections = 0;
+            long totalLargeRejectedOpsBytes = 0;
             for (NodeStats nodeStat : nodeStats) {
                 IndexingPressureStats nodeStatIndexingPressureStats = nodeStat.getIndexingPressureStats();
                 if (nodeStatIndexingPressureStats != null) {
@@ -820,6 +822,8 @@ static class IndexPressureStats implements ToXContentFragment {
                     totalCoordinatingRequests += nodeStatIndexingPressureStats.getTotalCoordinatingRequests();
                     lowWaterMarkSplits += nodeStatIndexingPressureStats.getLowWaterMarkSplits();
                     highWaterMarkSplits += nodeStatIndexingPressureStats.getHighWaterMarkSplits();
+                    largeOpsRejections += nodeStatIndexingPressureStats.getLargeOpsRejections();
+                    totalLargeRejectedOpsBytes += nodeStatIndexingPressureStats.getTotalLargeRejectedOpsBytes();
                 }
             }
             indexingPressureStats = new IndexingPressureStats(
@@ -844,7 +848,9 @@ static class IndexPressureStats implements ToXContentFragment {
                 primaryDocumentRejections,
                 totalCoordinatingRequests,
                 lowWaterMarkSplits,
-                highWaterMarkSplits
+                highWaterMarkSplits,
+                largeOpsRejections,
+                totalLargeRejectedOpsBytes
             );
         }
 

server/src/main/java/org/elasticsearch/action/admin/cluster/stats/ClusterStatsNodes.java

@@ -103,6 +103,24 @@ public long totalSizeInBytes() {
         return totalSizeInBytes;
     }
 
+    public long maxOperationSizeInBytes() {
+        long maxOperationSizeInBytes = 0;
+        for (int i = 0; i < items.length; i++) {
+            DocWriteRequest<?> request = items[i].request();
+            if (request instanceof IndexRequest) {
+                if (((IndexRequest) request).source() != null) {
+                    maxOperationSizeInBytes = Math.max(maxOperationSizeInBytes, ((IndexRequest) request).source().length());
+                }
+            } else if (request instanceof UpdateRequest) {
+                IndexRequest doc = ((UpdateRequest) request).doc();
+                if (doc != null && doc.source() != null) {
+                    maxOperationSizeInBytes = Math.max(maxOperationSizeInBytes, ((UpdateRequest) request).doc().source().length());
+                }
+            }
+        }
+        return maxOperationSizeInBytes;
+    }
+
     public BulkItemRequest[] items() {
         return items;
     }
@@ -199,6 +217,14 @@ public long ramBytesUsed() {
         return sum;
     }
 
+    public long largestOperationSize() {
+        long maxOperationSize = 0;
+        for (BulkItemRequest item : items) {
+            maxOperationSize = Math.max(maxOperationSize, item.ramBytesUsed());
+        }
+        return maxOperationSize;
+    }
+
     public boolean isSimulated() {
         return isSimulated;
     }

server/src/main/java/org/elasticsearch/action/bulk/BulkShardRequest.java

@@ -87,6 +87,13 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
 
     private static final Logger logger = LogManager.getLogger(TransportShardBulkAction.class);
 
+    // Represents the maximum memory overhead factor for an operation when processed for indexing.
+    // This accounts for potential increases in memory usage due to document expansion, including:
+    // 1. If the document source is not stored in a contiguous byte array, it will be copied to ensure contiguity.
+    // 2. If the document contains strings, Jackson uses char arrays (2 bytes per character) to parse string fields, doubling memory usage.
+    // 3. Parsed string fields create new copies of their data, further increasing memory consumption.
+    private static final int MAX_EXPANDED_OPERATION_MEMORY_OVERHEAD_FACTOR = 4;
+
     private final UpdateHelper updateHelper;
     private final MappingUpdatedAction mappingUpdatedAction;
     private final Consumer<Runnable> postWriteAction;
@@ -161,8 +168,16 @@ protected void shardOperationOnPrimary(
     protected void dispatchedShardOperationOnPrimary(
         BulkShardRequest request,
         IndexShard primary,
-        ActionListener<PrimaryResult<BulkShardRequest, BulkShardResponse>> listener
+        ActionListener<PrimaryResult<BulkShardRequest, BulkShardResponse>> outerListener
     ) {
+        var listener = ActionListener.releaseBefore(
+            indexingPressure.trackPrimaryOperationExpansion(
+                primaryOperationCount(request),
+                getMaxOperationMemoryOverhead(request),
+                force(request)
+            ),
+            outerListener
+        );
         ClusterStateObserver observer = new ClusterStateObserver(clusterService, request.timeout(), logger, threadPool.getThreadContext());
         performOnPrimary(request, primary, updateHelper, threadPool::absoluteTimeInMillis, (update, shardId, mappingListener) -> {
             assert update != null;
@@ -200,6 +215,16 @@ protected int primaryOperationCount(BulkShardRequest request) {
         return request.items().length;
     }
 
+    @Override
+    protected long primaryLargestOperationSize(BulkShardRequest request) {
+        return request.largestOperationSize();
+    }
+
+    @Override
+    protected boolean primaryAllowsOperationsBeyondSizeLimit(BulkShardRequest request) {
+        return false;
+    }
+
     public static void performOnPrimary(
         BulkShardRequest request,
         IndexShard primary,
@@ -638,7 +663,15 @@ private static BulkItemResponse processUpdateResponse(
     }
 
     @Override
-    protected void dispatchedShardOperationOnReplica(BulkShardRequest request, IndexShard replica, ActionListener<ReplicaResult> listener) {
+    protected void dispatchedShardOperationOnReplica(
+        BulkShardRequest request,
+        IndexShard replica,
+        ActionListener<ReplicaResult> outerListener
+    ) {
+        var listener = ActionListener.releaseBefore(
+            indexingPressure.trackReplicaOperationExpansion(getMaxOperationMemoryOverhead(request), force(request)),
+            outerListener
+        );
         ActionListener.completeWith(listener, () -> {
             final long startBulkTime = System.nanoTime();
             final Translog.Location location = performOnReplica(request, replica);
@@ -647,6 +680,10 @@ protected void dispatchedShardOperationOnReplica(BulkShardRequest request, Index
         });
     }
 
+    private static long getMaxOperationMemoryOverhead(BulkShardRequest request) {
+        return request.maxOperationSizeInBytes() * MAX_EXPANDED_OPERATION_MEMORY_OVERHEAD_FACTOR;
+    }
+
     @Override
     protected long replicaOperationSize(BulkShardRequest request) {
         return request.ramBytesUsed();

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -138,6 +138,11 @@ protected int primaryOperationCount(ResyncReplicationRequest request) {
         return request.getOperations().length;
     }
 
+    @Override
+    protected long primaryLargestOperationSize(ResyncReplicationRequest request) {
+        return Stream.of(request.getOperations()).mapToLong(Translog.Operation::estimateSize).max().orElse(0);
+    }
+
     public static ResyncReplicationRequest performOnPrimary(ResyncReplicationRequest request) {
         return request;
     }