ILM don't rollover empty indices (#89557)

docs/changelog/89557.yaml

@@ -0,0 +1,42 @@
+pr: 89557
+summary: ILM don't rollover empty indices
+area: ILM+SLM
+type: enhancement
+issues:
+ - 86203
+highlight:
+  title: ILM no longer rolls over empty indices
+  body: |-
+    For both new and existing Index Lifecycle Management (ILM) policies,
+    the rollover action will only execute if an index has at least one document.
+
+    For indices with a `max_age` condition that are no longer being written
+    to, this will mean that they will no longer roll over every time their
+    `max_age` is reached.
+
+    A policy can override this behavior, and explicitly opt in to rolling over
+    empty indices, by adding a `"min_docs": 0` condition:
+
+    [source,console]
+    ----
+    PUT _ilm/policy/allow_empty_rollover_policy
+    {
+      "policy": {
+        "phases": {
+          "hot": {
+            "actions": {
+              "rollover" : {
+                "max_age": "7d",
+                "max_size": "100gb",
+                "min_docs": 0
+              }
+            }
+          }
+        }
+      }
+    }
+    ----
+
+    This can also be disabled on a cluster-wide basis by setting
+    `indices.lifecycle.rollover.only_if_has_documents` to `false`.
+  notable: true

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

@@ -50,7 +50,8 @@ A rollover action must specify at least one max_* condition, it may include zero
 or more min_* conditions. An empty rollover action is invalid.
 
 The index will rollover once any max_* condition is satisfied and all
-min_* conditions are satisfied.
+min_* conditions are satisfied. Note, however, that empty indices are not rolled
+over by default.
 
 // tag::rollover-conditions[]
 `max_age`::
@@ -122,6 +123,12 @@ See notes on `max_primary_shard_docs`.
 
 // end::rollover-conditions[]
 
+IMPORTANT: Empty indices will not be rolled over, even if they have an associated `max_age` that
+would otherwise result in a roll over occurring. A policy can override this behavior, and explicitly
+opt in to rolling over empty indices, by adding a `"min_docs": 0` condition. This can also be
+disabled on a cluster-wide basis by setting `indices.lifecycle.rollover.only_if_has_documents` to
+`false`.
+
 [[ilm-rollover-ex]]
 ==== Example
 
@@ -246,7 +253,7 @@ PUT _ilm/policy/my_policy
 When you specify multiple rollover conditions,
 the index is rolled over when _any_ of the max_* and _all_ of the min_* conditions are met.
 This example rolls the index over if it is at least 7 days old or at least 100 gigabytes,
-but only as long as the index is not empty.
+but only as long as the index contains at least 1000 documents.
 
 [source,console]
 --------------------------------------------------
@@ -259,7 +266,7 @@ PUT _ilm/policy/my_policy
           "rollover" : {
             "max_age": "7d",
             "max_size": "100gb",
-            "min_docs": 1
+            "min_docs": 1000
           }
         }
       }

docs/reference/settings/ilm-settings.asciidoc

@@ -26,6 +26,12 @@ indices. Defaults to `true`.
 (<<dynamic-cluster-setting,Dynamic>>, <<time-units, time unit value>>)
 How often {ilm} checks for indices that meet policy criteria. Defaults to `10m`.
 
+[[indices-lifecycle-rollover-only-if-has-documents]]
+`indices.lifecycle.rollover.only_if_has_documents`::
+(<<dynamic-cluster-setting,Dynamic>>, Boolean)
+Whether ILM will only roll over non-empty indices. If enabled, ILM will only roll over indices
+as long as they contain at least one document. Defaults to `true`.
+
 ==== Index level settings
 These index-level {ilm-init} settings are typically configured through index
 templates. For more information, see <<ilm-gs-create-policy>>.

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/LifecycleSettings.java

@@ -21,9 +21,10 @@ public class LifecycleSettings {
     public static final String LIFECYCLE_INDEXING_COMPLETE = "index.lifecycle.indexing_complete";
     public static final String LIFECYCLE_ORIGINATION_DATE = "index.lifecycle.origination_date";
     public static final String LIFECYCLE_PARSE_ORIGINATION_DATE = "index.lifecycle.parse_origination_date";
-    public static final String LIFECYCLE_STEP_WAIT_TIME_THRESHOLD = "index.lifecycle.step.wait_time_threshold";
     public static final String LIFECYCLE_HISTORY_INDEX_ENABLED = "indices.lifecycle.history_index_enabled";
     public static final String LIFECYCLE_STEP_MASTER_TIMEOUT = "indices.lifecycle.step.master_timeout";
+    public static final String LIFECYCLE_STEP_WAIT_TIME_THRESHOLD = "index.lifecycle.step.wait_time_threshold";
+    public static final String LIFECYCLE_ROLLOVER_ONLY_IF_HAS_DOCUMENTS = "indices.lifecycle.rollover.only_if_has_documents";
 
     public static final String SLM_HISTORY_INDEX_ENABLED = "slm.history_index_enabled";
     public static final String SLM_RETENTION_SCHEDULE = "slm.retention_schedule";
@@ -89,6 +90,13 @@ public class LifecycleSettings {
         Setting.Property.Dynamic,
         Setting.Property.IndexScope
     );
+    public static final Setting<Boolean> LIFECYCLE_ROLLOVER_ONLY_IF_HAS_DOCUMENTS_SETTING = Setting.boolSetting(
+        LIFECYCLE_ROLLOVER_ONLY_IF_HAS_DOCUMENTS,
+        true,
+        Setting.Property.Dynamic,
+        Setting.Property.NodeScope,
+        Setting.Property.DeprecatedWarning
+    );
 
     public static final Setting<Boolean> SLM_HISTORY_INDEX_ENABLED_SETTING = Setting.boolSetting(
         SLM_HISTORY_INDEX_ENABLED,

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/WaitForRolloverReadyStep.java

@@ -196,6 +196,36 @@ public void evaluateCondition(Metadata metadata, Index index, Listener listener,
             rolloverTarget = rolloverAlias;
         }
 
+        // if we should only rollover if not empty, *and* if neither an explicit min_docs nor an explicit min_primary_shard_docs
+        // has been specified on this policy, then inject a default min_docs: 1 condition so that we do not rollover empty indices
+        boolean rolloverOnlyIfHasDocuments = LifecycleSettings.LIFECYCLE_ROLLOVER_ONLY_IF_HAS_DOCUMENTS_SETTING.get(metadata.settings());
+        RolloverRequest rolloverRequest = createRolloverRequest(rolloverTarget, masterTimeout, rolloverOnlyIfHasDocuments);
+
+        getClient().admin()
+            .indices()
+            .rolloverIndex(
+                rolloverRequest,
+                ActionListener.wrap(
+                    response -> listener.onResponse(rolloverRequest.areConditionsMet(response.getConditionStatus()), EmptyInfo.INSTANCE),
+                    listener::onFailure
+                )
+            );
+    }
+
+    /**
+     * Builds a RolloverRequest that captures the various max_* and min_* conditions of this {@link WaitForRolloverReadyStep}.
+     *
+     * To prevent empty indices from rolling over, a `min_docs: 1` condition will be injected if `rolloverOnlyIfHasDocuments` is true
+     * and the request doesn't already have an associated min_docs or min_primary_shard_docs condition.
+     *
+     * @param rolloverTarget the index to rollover
+     * @param masterTimeout the master timeout to use with the request
+     * @param rolloverOnlyIfHasDocuments whether to inject a min_docs 1 condition if there is not already a min_docs
+     *                                   (or min_primary_shard_docs) condition
+     * @return A RolloverRequest suitable for passing to {@code rolloverIndex(...) }.
+     */
+    // visible for testing
+    RolloverRequest createRolloverRequest(String rolloverTarget, TimeValue masterTimeout, boolean rolloverOnlyIfHasDocuments) {
         RolloverRequest rolloverRequest = new RolloverRequest(rolloverTarget, null).masterNodeTimeout(masterTimeout);
         rolloverRequest.dryRun(true);
         if (maxSize != null) {
@@ -228,15 +258,11 @@ public void evaluateCondition(Metadata metadata, Index index, Listener listener,
         if (minPrimaryShardDocs != null) {
             rolloverRequest.addMinPrimaryShardDocsCondition(minPrimaryShardDocs);
         }
-        getClient().admin()
-            .indices()
-            .rolloverIndex(
-                rolloverRequest,
-                ActionListener.wrap(
-                    response -> listener.onResponse(rolloverRequest.areConditionsMet(response.getConditionStatus()), EmptyInfo.INSTANCE),
-                    listener::onFailure
-                )
-            );
+
+        if (rolloverOnlyIfHasDocuments && (minDocs == null && minPrimaryShardDocs == null)) {
+            rolloverRequest.addMinIndexDocsCondition(1L);
+        }
+        return rolloverRequest;
     }
 
     ByteSizeValue getMaxSize() {