Rollover add max_primary_shard_docs condition

docs/changelog/80981.yaml

@@ -0,0 +1,5 @@
+pr: 80981
+summary: Rollover add max_primary_shard_docs condition
+area: ILM+SLM
+type: enhancement
+issues: []

docs/reference/ilm/actions/ilm-rollover.asciidoc

@@ -81,6 +81,15 @@ replicas are ignored.
 TIP: To see the current shard size, use the <<cat-shards, _cat shards>> API.
 The `store` value shows the size each shard, and `prirep` indicates whether a
 shard is a primary (`p`) or a replica (`r`).
+
+`max_primary_shard_docs`::
+(Optional, integer)
+Triggers rollover when the largest primary shard in the index reaches a certain number of documents.
+This is the maximum docs of the primary shards in the index. As with `max_docs`,
+replicas are ignored.
++
+TIP: To see the current shard docs, use the <<cat-shards, _cat shards>> API.
+The `docs` value shows the number of documents each shard.
 // end::rollover-conditions[]
 
 [[ilm-rollover-ex]]
@@ -155,6 +164,29 @@ PUT _ilm/policy/my_policy
 }
 --------------------------------------------------
 
+[ilm-rollover-documents-ex]]
+===== Roll over based on document count of the largest primary shard
+
+This example rolls the index over when it contains at least ten million documents of the largest primary shard.
+
+[source,console]
+--------------------------------------------------
+PUT _ilm/policy/my_policy
+{
+  "policy": {
+    "phases": {
+      "hot": {
+        "actions": {
+          "rollover" : {
+            "max_primary_shard_docs": 10000000
+          }
+        }
+      }
+    }
+  }
+}
+--------------------------------------------------
+
 [ilm-rollover-age-ex]]
 ===== Roll over based on index age
 

docs/reference/indices/rollover-index.asciidoc

@@ -79,7 +79,7 @@ If you use an index alias for time series data, you can use
 date. For example, you can create an alias that points to an index named
 `<my-index-{now/d}-000001>`. If you create the index on May 6, 2099, the index's
 name is `my-index-2099.05.06-000001`. If you roll over the alias on May 7, 2099,
-the new index's name is `my-index-2099.05.07-000002`. For an example, see 
+the new index's name is `my-index-2099.05.07-000002`. For an example, see
 <<roll-over-index-alias-with-write-index>>.
 ****
 
@@ -233,7 +233,8 @@ POST my-data-stream/_rollover
   "conditions": {
     "max_age": "7d",
     "max_docs": 1000,
-    "max_primary_shard_size": "50gb"
+    "max_primary_shard_size": "50gb",
+    "max_primary_shard_docs": "2000"
   }
 }
 ----
@@ -255,7 +256,8 @@ The API returns:
   "conditions": {
     "[max_age: 7d]": false,
     "[max_docs: 1000]": true,
-    "[max_primary_shard_size: 50gb]": false
+    "[max_primary_shard_size: 50gb]": false,
+    "[max_primary_shard_docs: 2000]": false
   }
 }
 ----
@@ -302,7 +304,8 @@ POST my-alias/_rollover
   "conditions": {
     "max_age": "7d",
     "max_docs": 1000,
-    "max_primary_shard_size": "50gb"
+    "max_primary_shard_size": "50gb",
+    "max_primary_shard_docs": "2000"
   }
 }
 ----
@@ -324,7 +327,8 @@ The API returns:
   "conditions": {
     "[max_age: 7d]": false,
     "[max_docs: 1000]": true,
-    "[max_primary_shard_size: 50gb]": false
+    "[max_primary_shard_size: 50gb]": false,
+    "[max_primary_shard_docs: 2000]": false
   }
 }
 ----
@@ -371,7 +375,8 @@ POST my-write-alias/_rollover
   "conditions": {
     "max_age": "7d",
     "max_docs": 1000,
-    "max_primary_shard_size": "50gb"
+    "max_primary_shard_size": "50gb",
+    "max_primary_shard_docs": "2000"
   }
 }
 ----
@@ -393,7 +398,8 @@ The API returns:
   "conditions": {
     "[max_age: 7d]": false,
     "[max_docs: 1000]": true,
-    "[max_primary_shard_size: 50gb]": false
+    "[max_primary_shard_size: 50gb]": false,
+    "[max_primary_shard_docs: 2000]": false
   }
 }
 ----

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/indices.rollover/25_max_shard_doc_condition.yml

@@ -0,0 +1,59 @@
+---
+"Rollover with max_primary_shard_docs condition":
+  - skip:
+      version: " - 8.1.99"
+      reason: introduced in 8.2.0
+
+  # create index with alias and replica
+  - do:
+      indices.create:
+        index: logs-1
+        wait_for_active_shards: 1
+        body:
+          settings:
+            index:
+              number_of_shards: 1
+              number_of_replicas: 0
+          aliases:
+            logs_search: {}
+
+  # index first document and wait for refresh
+  - do:
+      index:
+        index: logs-1
+        id:    "1"
+        body:  { "foo": "hello world" }
+        refresh: true
+
+  # perform alias rollover with no result
+  - do:
+      indices.rollover:
+        alias: "logs_search"
+        wait_for_active_shards: 1
+        body:
+          conditions:
+            max_primary_shard_docs: 2
+
+  - match: { conditions: { "[max_primary_shard_docs: 2]": false } }
+  - match: { rolled_over: false }
+
+  # index second document and wait for refresh
+  - do:
+      index:
+        index: logs-1
+        id:    "2"
+        body:  { "foo": "hello world" }
+        refresh: true
+
+  # perform alias rollover
+  - do:
+      indices.rollover:
+        alias: "logs_search"
+        wait_for_active_shards: 1
+        body:
+          conditions:
+            max_primary_shard_docs: 2
+
+  - match: { conditions: { "[max_primary_shard_docs: 2]": true } }
+  - match: { rolled_over: true }
+

server/src/internalClusterTest/java/org/elasticsearch/action/admin/indices/rollover/RolloverIT.java

@@ -542,6 +542,71 @@ public void testRolloverMaxPrimaryShardSize() throws Exception {
         }
     }
 
+    public void testRolloverMaxPrimaryShardDocs() throws Exception {
+        assertAcked(
+            prepareCreate("test-1").setSettings(Settings.builder().put("index.number_of_shards", 1)).addAlias(new Alias("test_alias")).get()
+        );
+        int numDocs = randomIntBetween(10, 20);
+        for (int i = 0; i < numDocs; i++) {
+            indexDoc("test-1", Integer.toString(i), "field", "foo-" + i);
+        }
+        flush("test-1");
+        refresh("test_alias");
+
+        // A large max_primary_shard_docs
+        {
+            final RolloverResponse response = client().admin()
+                .indices()
+                .prepareRolloverIndex("test_alias")
+                .addMaxPrimaryShardDocsCondition(randomIntBetween(21, 30))
+                .get();
+            assertThat(response.getOldIndex(), equalTo("test-1"));
+            assertThat(response.getNewIndex(), equalTo("test-000002"));
+            assertThat("No rollover with a large max_primary_shard_docs condition", response.isRolledOver(), equalTo(false));
+            final IndexMetadata oldIndex = client().admin().cluster().prepareState().get().getState().metadata().index("test-1");
+            assertThat(oldIndex.getRolloverInfos().size(), equalTo(0));
+        }
+
+        // A small max_primary_shard_docs
+        {
+            MaxPrimaryShardDocsCondition maxPrimaryShardDocsCondition = new MaxPrimaryShardDocsCondition(randomLongBetween(1, 9));
+            long beforeTime = client().threadPool().absoluteTimeInMillis() - 1000L;
+            final RolloverResponse response = client().admin()
+                .indices()
+                .prepareRolloverIndex("test_alias")
+                .addMaxPrimaryShardDocsCondition(maxPrimaryShardDocsCondition.value)
+                .get();
+            assertThat(response.getOldIndex(), equalTo("test-1"));
+            assertThat(response.getNewIndex(), equalTo("test-000002"));
+            assertThat("Should rollover with a small max_primary_shard_docs condition", response.isRolledOver(), equalTo(true));
+            final IndexMetadata oldIndex = client().admin().cluster().prepareState().get().getState().metadata().index("test-1");
+            List<Condition<?>> metConditions = oldIndex.getRolloverInfos().get("test_alias").getMetConditions();
+            assertThat(metConditions.size(), equalTo(1));
+            assertThat(
+                metConditions.get(0).toString(),
+                equalTo(new MaxPrimaryShardDocsCondition(maxPrimaryShardDocsCondition.value).toString())
+            );
+            assertThat(
+                oldIndex.getRolloverInfos().get("test_alias").getTime(),
+                is(both(greaterThanOrEqualTo(beforeTime)).and(lessThanOrEqualTo(client().threadPool().absoluteTimeInMillis() + 1000L)))
+            );
+        }
+
+        // An empty index
+        {
+            final RolloverResponse response = client().admin()
+                .indices()
+                .prepareRolloverIndex("test_alias")
+                .addMaxPrimaryShardDocsCondition(randomNonNegativeLong())
+                .get();
+            assertThat(response.getOldIndex(), equalTo("test-000002"));
+            assertThat(response.getNewIndex(), equalTo("test-000003"));
+            assertThat("No rollover with an empty index", response.isRolledOver(), equalTo(false));
+            final IndexMetadata oldIndex = client().admin().cluster().prepareState().get().getState().metadata().index("test-000002");
+            assertThat(oldIndex.getRolloverInfos().size(), equalTo(0));
+        }
+    }
+
     public void testRejectIfAliasFoundInTemplate() throws Exception {
         client().admin()
             .indices()

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/Condition.java

@@ -70,7 +70,13 @@ public String name() {
     /**
      * Holder for index stats used to evaluate conditions
      */
-    public record Stats(long numDocs, long indexCreated, ByteSizeValue indexSize, ByteSizeValue maxPrimaryShardSize) {}
+    public record Stats(
+        long numDocs,
+        long indexCreated,
+        ByteSizeValue indexSize,
+        ByteSizeValue maxPrimaryShardSize,
+        long maxPrimaryShardDocs
+    ) {}
 
     /**
      * Holder for evaluated condition result

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/MaxPrimaryShardDocsCondition.java

@@ -0,0 +1,64 @@
+/*
+ * 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 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 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.action.admin.indices.rollover;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParser;
+
+import java.io.IOException;
+
+public class MaxPrimaryShardDocsCondition extends Condition<Long> {
+    public static final String NAME = "max_primary_shard_docs";
+
+    public MaxPrimaryShardDocsCondition(Long value) {
+        super(NAME);
+        this.value = value;
+    }
+
+    public MaxPrimaryShardDocsCondition(StreamInput in) throws IOException {
+        super(NAME);
+        this.value = in.readLong();
+    }
+
+    @Override
+    public Result evaluate(Stats stats) {
+        return new Result(this, this.value <= stats.maxPrimaryShardDocs());
+    }
+
+    @Override
+    public String getWriteableName() {
+        return NAME;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeLong(value);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        return builder.field(NAME, value);
+    }
+
+    public static MaxPrimaryShardDocsCondition fromXContent(XContentParser parser) throws IOException {
+        if (parser.nextToken() == XContentParser.Token.VALUE_NUMBER) {
+            return new MaxPrimaryShardDocsCondition(parser.longValue());
+        } else {
+            throw new IllegalArgumentException("invalid token: " + parser.currentToken());
+        }
+    }
+
+    @Override
+    boolean includedInVersion(Version version) {
+        return version.onOrAfter(Version.V_8_2_0);
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/RolloverRequest.java

@@ -46,6 +46,7 @@ public class RolloverRequest extends AcknowledgedRequest<RolloverRequest> implem
     private static final ParseField MAX_DOCS_CONDITION = new ParseField(MaxDocsCondition.NAME);
     private static final ParseField MAX_SIZE_CONDITION = new ParseField(MaxSizeCondition.NAME);
     private static final ParseField MAX_PRIMARY_SHARD_SIZE_CONDITION = new ParseField(MaxPrimaryShardSizeCondition.NAME);
+    private static final ParseField MAX_PRIMARY_SHARD_DOCS_CONDITION = new ParseField(MaxPrimaryShardDocsCondition.NAME);
 
     static {
         CONDITION_PARSER.declareString(
@@ -70,6 +71,10 @@ public class RolloverRequest extends AcknowledgedRequest<RolloverRequest> implem
             ),
             MAX_PRIMARY_SHARD_SIZE_CONDITION
         );
+        CONDITION_PARSER.declareLong(
+            (conditions, value) -> conditions.put(MaxPrimaryShardDocsCondition.NAME, new MaxPrimaryShardDocsCondition(value)),
+            MAX_PRIMARY_SHARD_DOCS_CONDITION
+        );
 
         PARSER.declareField(
             (parser, request, context) -> CONDITION_PARSER.parse(parser, request.conditions, null),
@@ -256,6 +261,17 @@ public void addMaxPrimaryShardSizeCondition(ByteSizeValue size) {
         this.conditions.put(maxPrimaryShardSizeCondition.name, maxPrimaryShardSizeCondition);
     }
 
+    /**
+     * Adds a size-based condition to check if the docs of the largest primary shard has at least <code>numDocs</code>
+     */
+    public void addMaxPrimaryShardDocsCondition(long numDocs) {
+        MaxPrimaryShardDocsCondition maxPrimaryShardDocsCondition = new MaxPrimaryShardDocsCondition(numDocs);
+        if (this.conditions.containsKey(maxPrimaryShardDocsCondition.name)) {
+            throw new IllegalArgumentException(maxPrimaryShardDocsCondition.name + " condition is already set");
+        }
+        this.conditions.put(maxPrimaryShardDocsCondition.name, maxPrimaryShardDocsCondition);
+    }
+
     public boolean isDryRun() {
         return dryRun;
     }

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/RolloverRequestBuilder.java

@@ -50,6 +50,11 @@ public RolloverRequestBuilder addMaxPrimaryShardSizeCondition(ByteSizeValue size
         return this;
     }
 
+    public RolloverRequestBuilder addMaxPrimaryShardDocsCondition(long docs) {
+        this.request.addMaxPrimaryShardDocsCondition(docs);
+        return this;
+    }
+
     public RolloverRequestBuilder dryRun(boolean dryRun) {
         this.request.dryRun(dryRun);
         return this;