Add configured indexing memory limit to node stats (elastic#60342)

This commit adds the configured memory limit to the node stats API.

build.gradle

@@ -182,8 +182,8 @@ tasks.register("verifyVersions") {
  * after the backport of the backcompat code is complete.
  */
 
-boolean bwc_tests_enabled = true
-final String bwc_tests_disabled_issue = "" /* place a PR link here when committing bwc changes */
+boolean bwc_tests_enabled = false
+final String bwc_tests_disabled_issue = "https://github.com/elastic/elasticsearch/pull/60342" /* place a PR link here when committing bwc changes */
 if (bwc_tests_enabled == false) {
   if (bwc_tests_disabled_issue.isEmpty()) {
     throw new GradleException("bwc_tests_disabled_issue must be set when bwc_tests_enabled == false")

docs/reference/cluster/nodes-stats.asciidoc

@@ -2227,6 +2227,15 @@ Number of indexing requests rejected in the primary stage.
 (integer)
 Number of indexing requests rejected in the replica stage.
 ========
+`limit`::
+(<<byte-units,byte value>>)
+Configured memory limit for the indexing requests. Replica requests have an
+automatic limit that is 1.5x this value.
+
+`limit_in_bytes`::
+(integer)
+Configured memory limit, in bytes, for the indexing requests. Replica requests
+have an automatic limit that is 1.5x this value.
 =======
 ======
 

rest-api-spec/src/main/resources/rest-api-spec/test/nodes.stats/50_indexing_pressure.yml

@@ -29,3 +29,21 @@
   - gte:  { nodes.$node_id.indexing_pressure.memory.total.coordinating_rejections: 0 }
   - gte:  { nodes.$node_id.indexing_pressure.memory.total.primary_rejections: 0 }
   - gte:  { nodes.$node_id.indexing_pressure.memory.total.replica_rejections: 0 }
+---
+"Indexing pressure memory limit":
+  - skip:
+      # Change to 7.9.99 on backport
+      version: " - 7.10.99"
+      reason: "memory limit was added in 7.10"
+      features: [arbitrary_key]
+
+  - do:
+      nodes.info: {}
+  - set:
+      nodes._arbitrary_key_: node_id
+
+  - do:
+      nodes.stats:
+        metric: [ indexing_pressure ]
+
+  - gte:  { nodes.$node_id.indexing_pressure.memory.limit_in_bytes: 0 }

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

@@ -146,6 +146,6 @@ public IndexingPressureStats stats() {
         return new IndexingPressureStats(totalCombinedCoordinatingAndPrimaryBytes.get(), totalCoordinatingBytes.get(),
             totalPrimaryBytes.get(), totalReplicaBytes.get(), currentCombinedCoordinatingAndPrimaryBytes.get(),
             currentCoordinatingBytes.get(), currentPrimaryBytes.get(), currentReplicaBytes.get(), coordinatingRejections.get(),
-            primaryRejections.get(), replicaRejections.get());
+            primaryRejections.get(), replicaRejections.get(), primaryAndCoordinatingLimits);
     }
 }

server/src/main/java/org/elasticsearch/index/stats/IndexingPressureStats.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.index.stats;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -43,6 +44,7 @@ public class IndexingPressureStats implements Writeable, ToXContentFragment {
     private final long coordinatingRejections;
     private final long primaryRejections;
     private final long replicaRejections;
+    private final long memoryLimit;
 
     public IndexingPressureStats(StreamInput in) throws IOException {
         totalCombinedCoordinatingAndPrimaryBytes = in.readVLong();
@@ -58,12 +60,19 @@ public IndexingPressureStats(StreamInput in) throws IOException {
         coordinatingRejections = in.readVLong();
         primaryRejections = in.readVLong();
         replicaRejections = in.readVLong();
+
+        // TODO: Change to 7.10 after backport
+        if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+            memoryLimit = in.readVLong();
+        } else {
+            memoryLimit = -1L;
+        }
     }
 
     public IndexingPressureStats(long totalCombinedCoordinatingAndPrimaryBytes, long totalCoordinatingBytes, long totalPrimaryBytes,
                                  long totalReplicaBytes, long currentCombinedCoordinatingAndPrimaryBytes, long currentCoordinatingBytes,
                                  long currentPrimaryBytes, long currentReplicaBytes, long coordinatingRejections, long primaryRejections,
-                                 long replicaRejections) {
+                                 long replicaRejections, long memoryLimit) {
         this.totalCombinedCoordinatingAndPrimaryBytes = totalCombinedCoordinatingAndPrimaryBytes;
         this.totalCoordinatingBytes = totalCoordinatingBytes;
         this.totalPrimaryBytes = totalPrimaryBytes;
@@ -75,6 +84,7 @@ public IndexingPressureStats(long totalCombinedCoordinatingAndPrimaryBytes, long
         this.coordinatingRejections = coordinatingRejections;
         this.primaryRejections = primaryRejections;
         this.replicaRejections = replicaRejections;
+        this.memoryLimit = memoryLimit;
     }
 
     @Override
@@ -92,6 +102,11 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeVLong(coordinatingRejections);
         out.writeVLong(primaryRejections);
         out.writeVLong(replicaRejections);
+
+        // TODO: Change to 7.10 after backport
+        if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+            out.writeVLong(memoryLimit);
+        }
     }
 
     public long getTotalCombinedCoordinatingAndPrimaryBytes() {
@@ -151,6 +166,8 @@ public long getReplicaRejections() {
     private static final String COORDINATING_REJECTIONS = "coordinating_rejections";
     private static final String PRIMARY_REJECTIONS = "primary_rejections";
     private static final String REPLICA_REJECTIONS = "replica_rejections";
+    private static final String LIMIT = "limit";
+    private static final String LIMIT_IN_BYTES = "limit_in_bytes";
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
@@ -173,6 +190,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.field(PRIMARY_REJECTIONS, primaryRejections);
         builder.field(REPLICA_REJECTIONS, replicaRejections);
         builder.endObject();
+        builder.humanReadableField(LIMIT_IN_BYTES, LIMIT, new ByteSizeValue(memoryLimit));
         builder.endObject();
         return builder.endObject();
     }