ILM: Force merge on zero-replica cloned index before snapshot (elastic#133954)

When performing a searchable snapshot action with force merge enabled,
if the source index has one or more replicas, ILM now clones the index
with zero replicas and performs the force merge on the clone. The
snapshot is then taken from the force-merged clone instead of the source
index, ensuring only primary shards are force-merged. The cloned index
is deleted after the snapshot is mounted, and all references and step
logic have been updated accordingly. Test coverage was added for the new
flow, including handling retries and cleanup of failed clones.

Key changes:
- Execution state: Track the force-merged clone index in ILM state and
propagate through relevant APIs.
- SearchableSnapshotAction: Add conditional steps to clone the index
with 0 replicas, force-merge, and delete the clone as needed.
- Steps: Update ForceMerge, SegmentCount, Snapshot, and Delete steps to
operate on the correct index (source or clone).
- Tests/QA: Add and enhance tests to verify force-merge and snapshot
behavior with and without replicas, including retry/cleanup paths and
configuration for stable force-merges.

Resolves elastic#75478

Author: nielsbauman

docs/changelog/133954.yaml

@@ -0,0 +1,6 @@
+pr: 133954
+summary: "ILM: Force merge on zero-replica cloned index before snapshotting for searchable snapshots"
+area: ILM+SLM
+type: enhancement
+issues:
+ - 75478

server/src/main/java/org/elasticsearch/cluster/metadata/LifecycleExecutionState.java

@@ -40,7 +40,8 @@ public record LifecycleExecutionState(
     String snapshotName,
     String shrinkIndexName,
     String snapshotIndexName,
-    String downsampleIndexName
+    String downsampleIndexName,
+    String forceMergeCloneIndexName
 ) {
 
     public static final String ILM_CUSTOM_METADATA_KEY = "ilm";
@@ -64,6 +65,7 @@ public record LifecycleExecutionState(
     private static final String SNAPSHOT_INDEX_NAME = "snapshot_index_name";
     private static final String SHRINK_INDEX_NAME = "shrink_index_name";
     private static final String DOWNSAMPLE_INDEX_NAME = "rollup_index_name";
+    private static final String FORCE_MERGE_CLONE_INDEX_NAME = "force_merge_clone_index_name";
 
     public static final LifecycleExecutionState EMPTY_STATE = LifecycleExecutionState.builder().build();
 
@@ -89,7 +91,8 @@ public static Builder builder(LifecycleExecutionState state) {
             .setShrinkIndexName(state.shrinkIndexName)
             .setSnapshotIndexName(state.snapshotIndexName)
             .setDownsampleIndexName(state.downsampleIndexName)
-            .setStepTime(state.stepTime);
+            .setStepTime(state.stepTime)
+            .setForceMergeCloneIndexName(state.forceMergeCloneIndexName);
     }
 
     public static LifecycleExecutionState fromCustomMetadata(Map<String, String> customData) {
@@ -202,6 +205,10 @@ public static LifecycleExecutionState fromCustomMetadata(Map<String, String> cus
         if (downsampleIndexName != null) {
             builder.setDownsampleIndexName(downsampleIndexName);
         }
+        String forceMergeCloneIndexName = customData.get(FORCE_MERGE_CLONE_INDEX_NAME);
+        if (forceMergeCloneIndexName != null) {
+            builder.setForceMergeCloneIndexName(forceMergeCloneIndexName);
+        }
         return builder.build();
     }
 
@@ -274,6 +281,9 @@ public Map<String, String> asMap() {
         if (downsampleIndexName != null) {
             result.put(DOWNSAMPLE_INDEX_NAME, downsampleIndexName);
         }
+        if (forceMergeCloneIndexName != null) {
+            result.put(FORCE_MERGE_CLONE_INDEX_NAME, forceMergeCloneIndexName);
+        }
         return Collections.unmodifiableMap(result);
     }
 
@@ -307,6 +317,7 @@ public static class Builder {
         private String shrinkIndexName;
         private String snapshotIndexName;
         private String downsampleIndexName;
+        private String forceMergeCloneIndexName;
 
         public Builder setPhase(String phase) {
             this.phase = phase;
@@ -398,6 +409,11 @@ public Builder setDownsampleIndexName(String downsampleIndexName) {
             return this;
         }
 
+        public Builder setForceMergeCloneIndexName(String forceMergeCloneIndexName) {
+            this.forceMergeCloneIndexName = forceMergeCloneIndexName;
+            return this;
+        }
+
         public LifecycleExecutionState build() {
             return new LifecycleExecutionState(
                 phase,
@@ -417,7 +433,8 @@ public LifecycleExecutionState build() {
                 snapshotName,
                 shrinkIndexName,
                 snapshotIndexName,
-                downsampleIndexName
+                downsampleIndexName,
+                forceMergeCloneIndexName
             );
         }
     }

test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

@@ -2087,6 +2087,10 @@ protected static void awaitIndexExists(String index, TimeValue timeout) throws I
         ensureHealth(client(), index, request -> request.addParameter("timeout", timeout.toString()));
     }
 
+    protected static void awaitIndexDoesNotExist(String index) throws Exception {
+        awaitIndexDoesNotExist(index, TimeValue.timeValueSeconds(10));
+    }
+
     protected static void awaitIndexDoesNotExist(String index, TimeValue timeout) throws Exception {
         assertBusy(() -> assertFalse(indexExists(index)), timeout.millis(), TimeUnit.MILLISECONDS);
     }

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

@@ -101,8 +101,11 @@ void createSnapshot(ProjectId projectId, IndexMetadata indexMetadata, ActionList
             );
             return;
         }
+        // If we performed the force merge step on the cloned index, we need to snapshot that index instead of the original.
+        final String clonedIndexName = lifecycleState.forceMergeCloneIndexName();
+        final String forceMergedIndexName = clonedIndexName != null ? clonedIndexName : indexName;
         CreateSnapshotRequest request = new CreateSnapshotRequest(TimeValue.MAX_VALUE, snapshotRepository, snapshotName);
-        request.indices(indexName);
+        request.indices(forceMergedIndexName);
         // this is safe as the snapshot creation will still be async, it's just that the listener will be notified when the snapshot is
         // complete
         request.waitForCompletion(true);
@@ -112,7 +115,7 @@ void createSnapshot(ProjectId projectId, IndexMetadata indexMetadata, ActionList
             logger.debug(
                 "create snapshot response for policy [{}] and index [{}] is: {}",
                 policyName,
-                indexName,
+                forceMergedIndexName,
                 Strings.toString(response)
             );
             final SnapshotInfo snapInfo = response.getSnapshotInfo();
@@ -128,7 +131,7 @@ void createSnapshot(ProjectId projectId, IndexMetadata indexMetadata, ActionList
                         snapshotRepository,
                         snapshotName,
                         policyName,
-                        indexName,
+                        forceMergedIndexName,
                         snapInfo.failedShards(),
                         snapInfo.totalShards()
                     )

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

@@ -16,26 +16,56 @@
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.LifecycleExecutionState;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.Index;
 
+import java.util.function.BiFunction;
+
 /**
  * Deletes a single index.
  */
 public class DeleteStep extends AsyncRetryDuringSnapshotActionStep {
+
     public static final String NAME = "delete";
     private static final Logger logger = LogManager.getLogger(DeleteStep.class);
+    private static final BiFunction<String, LifecycleExecutionState, String> DEFAULT_TARGET_INDEX_NAME_SUPPLIER = (
+        indexName,
+        lifecycleState) -> indexName;
+
+    private final BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier;
+    private final boolean indexSurvives;
 
+    /**
+     * Use this constructor to delete the index that ILM is currently operating on.
+     */
     public DeleteStep(StepKey key, StepKey nextStepKey, Client client) {
+        this(key, nextStepKey, client, DEFAULT_TARGET_INDEX_NAME_SUPPLIER, false);
+    }
+
+    /**
+     * Use this constructor to delete a specific index, potentially different from the one that ILM is currently operating on. The parameter
+     * {@code indexSurvives} indicates whether the index that ILM runs on will survive (i.e. not get deleted) this step.
+     * Look at the callers of {@link AsyncActionStep#indexSurvives()} for more details.
+     */
+    public DeleteStep(
+        StepKey key,
+        StepKey nextStepKey,
+        Client client,
+        BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier,
+        boolean indexSurvives
+    ) {
         super(key, nextStepKey, client);
+        this.targetIndexNameSupplier = targetIndexNameSupplier;
+        this.indexSurvives = indexSurvives;
     }
 
     @Override
     public void performDuringNoSnapshot(IndexMetadata indexMetadata, ProjectMetadata currentProject, ActionListener<Void> listener) {
         String policyName = indexMetadata.getLifecyclePolicyName();
-        String indexName = indexMetadata.getIndex().getName();
+        String indexName = targetIndexNameSupplier.apply(indexMetadata.getIndex().getName(), indexMetadata.getLifecycleExecutionState());
         IndexAbstraction indexAbstraction = currentProject.getIndicesLookup().get(indexName);
         assert indexAbstraction != null : "invalid cluster metadata. index [" + indexName + "] was not found";
         DataStream dataStream = indexAbstraction.getParentDataStream();
@@ -88,7 +118,7 @@ public void performDuringNoSnapshot(IndexMetadata indexMetadata, ProjectMetadata
 
     @Override
     public boolean indexSurvives() {
-        return false;
+        return indexSurvives;
     }
 
     @Override

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

@@ -16,10 +16,12 @@
 import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.ProjectState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.LifecycleExecutionState;
 import org.elasticsearch.common.Strings;
 
 import java.util.Arrays;
 import java.util.Objects;
+import java.util.function.BiFunction;
 
 /**
  * Invokes a force merge on a single index.
@@ -28,11 +30,33 @@ public class ForceMergeStep extends AsyncActionStep {
 
     public static final String NAME = "forcemerge";
     private static final Logger logger = LogManager.getLogger(ForceMergeStep.class);
+    private static final BiFunction<String, LifecycleExecutionState, String> DEFAULT_TARGET_INDEX_NAME_SUPPLIER = (
+        indexName,
+        lifecycleState) -> indexName;
+
     private final int maxNumSegments;
+    private final BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier;
 
+    /**
+     * Creates a new {@link ForceMergeStep} that will perform a force merge on the index that ILM is currently operating on.
+     */
     public ForceMergeStep(StepKey key, StepKey nextStepKey, Client client, int maxNumSegments) {
+        this(key, nextStepKey, client, maxNumSegments, DEFAULT_TARGET_INDEX_NAME_SUPPLIER);
+    }
+
+    /**
+     * Creates a new {@link ForceMergeStep} that will perform a force merge on the index name returned by the supplier.
+     */
+    public ForceMergeStep(
+        StepKey key,
+        StepKey nextStepKey,
+        Client client,
+        int maxNumSegments,
+        BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier
+    ) {
         super(key, nextStepKey, client);
         this.maxNumSegments = maxNumSegments;
+        this.targetIndexNameSupplier = targetIndexNameSupplier;
     }
 
     @Override
@@ -51,7 +75,8 @@ public void performAction(
         ClusterStateObserver observer,
         ActionListener<Void> listener
     ) {
-        String indexName = indexMetadata.getIndex().getName();
+        String indexName = targetIndexNameSupplier.apply(indexMetadata.getIndex().getName(), indexMetadata.getLifecycleExecutionState());
+        assert indexName != null : "target index name supplier must not return null";
         ForceMergeRequest request = new ForceMergeRequest(indexName);
         request.maxNumSegments(maxNumSegments);
         getClient(currentState.projectId()).admin().indices().forceMerge(request, listener.delegateFailureAndWrap((l, response) -> {

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

@@ -21,6 +21,7 @@
 
 import java.util.Locale;
 import java.util.Objects;
+import java.util.function.BiFunction;
 
 /**
  * Generates a snapshot name for the given index and records it in the index metadata along with the provided snapshot repository.
@@ -35,10 +36,17 @@ public class GenerateSnapshotNameStep extends ClusterStateActionStep {
     private static final Logger logger = LogManager.getLogger(GenerateSnapshotNameStep.class);
 
     private final String snapshotRepository;
-
-    public GenerateSnapshotNameStep(StepKey key, StepKey nextStepKey, String snapshotRepository) {
+    private final BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier;
+
+    public GenerateSnapshotNameStep(
+        StepKey key,
+        StepKey nextStepKey,
+        String snapshotRepository,
+        BiFunction<String, LifecycleExecutionState, String> targetIndexNameSupplier
+    ) {
         super(key, nextStepKey);
         this.snapshotRepository = snapshotRepository;
+        this.targetIndexNameSupplier = targetIndexNameSupplier;
     }
 
     public String getSnapshotRepository() {
@@ -72,9 +80,11 @@ public ProjectState performAction(Index index, ProjectState projectState) {
                     + "] cannot continue until the repository is created or the policy is changed"
             );
         }
+        final String indexName = targetIndexNameSupplier.apply(index.getName(), lifecycleState);
+        assert indexName != null : "target index name supplier must not return null";
 
         LifecycleExecutionState.Builder newLifecycleState = LifecycleExecutionState.builder(lifecycleState);
-        newLifecycleState.setSnapshotIndexName(index.getName());
+        newLifecycleState.setSnapshotIndexName(indexName);
         newLifecycleState.setSnapshotRepository(snapshotRepository);
         if (lifecycleState.snapshotName() == null) {
             // generate and validate the snapshotName