Add ForceMergeActionIT

Author: nielsbauman

test/framework/src/main/java/org/elasticsearch/test/ClusterServiceUtils.java

@@ -239,8 +239,7 @@ public static void setAllElapsedMillis(ClusterStatePublicationEvent clusterState
         clusterStatePublicationEvent.setMasterApplyElapsedMillis(0L);
     }
 
-    public static void awaitClusterState(Logger logger, Predicate<ClusterState> statePredicate, ClusterService clusterService)
-        throws Exception {
+    public static void awaitClusterState(Logger logger, Predicate<ClusterState> statePredicate, ClusterService clusterService) {
         final var listener = addTemporaryStateListener(clusterService, statePredicate, ESTestCase.TEST_REQUEST_TIMEOUT);
         ESTestCase.safeAwait(listener, ESTestCase.TEST_REQUEST_TIMEOUT);
     }

test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java

@@ -85,6 +85,7 @@
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
@@ -1219,8 +1220,8 @@ public static PendingClusterTasksResponse getClusterPendingTasks(Client client)
         }
     }
 
-    protected void awaitClusterState(Predicate<ClusterState> statePredicate) throws Exception {
-        awaitClusterState(logger, internalCluster().getMasterName(), statePredicate);
+    protected static void awaitClusterState(Predicate<ClusterState> statePredicate) {
+        awaitClusterState(null, internalCluster().getMasterName(), statePredicate);
     }
 
     protected void awaitClusterState(String viaNode, Predicate<ClusterState> statePredicate) throws Exception {
@@ -1231,7 +1232,7 @@ public static void awaitClusterState(Logger logger, Predicate<ClusterState> stat
         awaitClusterState(logger, internalCluster().getMasterName(), statePredicate);
     }
 
-    public static void awaitClusterState(Logger logger, String viaNode, Predicate<ClusterState> statePredicate) throws Exception {
+    public static void awaitClusterState(Logger logger, String viaNode, Predicate<ClusterState> statePredicate) {
         ClusterServiceUtils.awaitClusterState(logger, statePredicate, internalCluster().getInstance(ClusterService.class, viaNode));
     }
 
@@ -1853,6 +1854,19 @@ public static void awaitIndexExists(String index, Client client, TimeValue timeo
         );
     }
 
+    /**
+     *
+     * Waits until the specified index no longer exists in the cluster.
+     * This method blocks until the index is deleted or times out.
+     * Note that this method waits by listening to cluster state updates <i>on the master node</i>.
+     * Meaning that if this method returns, all other nodes are aware that that index is deleted from the cluster state as well.
+     *
+     * @param index the name of the index to wait for deletion
+     */
+    public static void awaitIndexNotExists(String index) {
+        awaitClusterState(state -> state.metadata().getProject(ProjectId.DEFAULT).hasIndex(index) == false);
+    }
+
     /**
      * Syntactic sugar for enabling allocation for <code>indices</code>
      */

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

@@ -45,16 +45,16 @@ public class ForceMergeAction implements LifecycleAction {
         .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
         .build();
     private static final Settings CLONE_SETTINGS_WITH_CODEC = Settings.builder()
-        .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
-        .put(EngineConfig.INDEX_CODEC_SETTING.getKey(), CodecService.BEST_COMPRESSION_CODEC)
+        .put(CLONE_SETTINGS_WITHOUT_CODEC)
+        .put(BEST_COMPRESSION_SETTINGS)
         .build();
 
     public static final String NAME = "forcemerge";
     public static final ParseField MAX_NUM_SEGMENTS_FIELD = new ParseField("max_num_segments");
     public static final ParseField CODEC = new ParseField("index_codec");
 
-    public static final String FORCE_MERGED_INDEX_PREFIX = "force-merged-";
-    public static final BiFunction<String, LifecycleExecutionState, String> FORCE_MERGED_INDEX_NAME_SUPPLIER = (indexName, state) -> state
+    public static final String FORCE_MERGE_INDEX_PREFIX = "force-merge-";
+    public static final BiFunction<String, LifecycleExecutionState, String> FORCE_MERGE_INDEX_NAME_SUPPLIER = (indexName, state) -> state
         .forceMergeIndexName();
 
     public static final String CONDITIONAL_SKIP_FORCE_MERGE_STEP = BranchingStep.NAME + "-forcemerge-check-prerequisites";
@@ -167,6 +167,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
         StepKey replaceDataStreamIndexKey = new StepKey(phase, NAME, ReplaceDataStreamBackingIndexStep.NAME);
         StepKey deleteSourceIndexKey = new StepKey(phase, NAME, DeleteStep.NAME);
 
+        // If the index is mounted as a searchable snapshot, skip the whole force-merge action
         BranchingStep conditionalSkipForceMergeStep = new BranchingStep(
             conditionalSkipForceMergeKey,
             checkNotWriteIndexKey,
@@ -189,30 +190,33 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
             }
         );
 
-        // Indices in this step key can skip the no-op step and jump directly to the step with closeIndexKey/forcemergeKey key
+        // Wait for the index to not be the write index of a data stream
         CheckNotDataStreamWriteIndexStep checkNotWriteIndexStep = new CheckNotDataStreamWriteIndexStep(
             checkNotWriteIndexKey,
             waitUntilTimeSeriesEndTimeKey
         );
 
+        // If the index is a time series index, wait until its end time has passed
         WaitUntilTimeSeriesEndTimePassesStep waitUntilTimeSeriesEndTimeStep = new WaitUntilTimeSeriesEndTimePassesStep(
             waitUntilTimeSeriesEndTimeKey,
             conditionalSkipCloneKey,
             Instant::now
         );
 
+        // If the index already has 0 replicas, we can skip the clone steps. If the action is configured to change the codec, we must
+        // first update the codec of the original index before proceeding to the force-merge step.
         BranchingStep conditionalSkipCloneStep = new BranchingStep(
             conditionalSkipCloneKey,
             cleanupClonedIndexKey,
             codecChange ? closeIndexKey : forceMergeKey,
             (index, project) -> {
                 IndexMetadata indexMetadata = project.index(index);
                 assert indexMetadata != null : "index " + index.getName() + " must exist in the cluster state";
-                // Returns true if the index already has zero replicas, which means we can avoid the "clone with 0 replicas" dance
-                return indexMetadata.getSettings().getAsInt(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1) == 0;
+                return indexMetadata.getNumberOfReplicas() == 0;
             }
         );
 
+        // Closing the index is required to change the index codec.
         CloseIndexStep closeIndexStep = new CloseIndexStep(closeIndexKey, updateBestCompressionKey, client);
         UpdateSettingsStep updateBestCompressionStep = new UpdateSettingsStep(
             updateBestCompressionKey,
@@ -227,6 +231,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
             ClusterHealthStatus.GREEN
         );
 
+        // If a previous force-merge action created a clone index but the action did not complete, we need to clean up the old clone index.
         CleanupClonedIndexStep cleanupClonedIndexStep = new CleanupClonedIndexStep(cleanupClonedIndexKey, readOnlyKey, client);
         // The readOnlyKey used to exist for BwC reasons (as the step was removed at some point). It has now been reintroduced with the
         // changes to make the force-merge action use a cloned index. Therefore, we intentionally put this step before the
@@ -236,81 +241,86 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
         GenerateUniqueIndexNameStep generateCloneIndexNameStep = new GenerateUniqueIndexNameStep(
             generateCloneIndexNameKey,
             cloneIndexKey,
-            FORCE_MERGED_INDEX_PREFIX,
+            FORCE_MERGE_INDEX_PREFIX,
             (generatedIndexName, lifecycleStateBuilder) -> lifecycleStateBuilder.setForceMergeIndexName(generatedIndexName)
         );
         // Clone the index with 0 replicas and the best compression codec if configured.
         CloneIndexStep cloneIndexStep = new CloneIndexStep(
             cloneIndexKey,
             waitForClonedIndexGreenKey,
             client,
-            FORCE_MERGED_INDEX_NAME_SUPPLIER,
+            FORCE_MERGE_INDEX_NAME_SUPPLIER,
             codecChange ? CLONE_SETTINGS_WITH_CODEC : CLONE_SETTINGS_WITHOUT_CODEC
         );
+        // Wait for the cloned index to be green before proceeding with the force-merge. We wrap this with a
+        // ClusterStateWaitUntilThresholdStep to avoid waiting forever if the index cannot be started for some reason. On timeout,
+        // ILM will move back to the cleanup step, remove the cloned index, and retry the clone.
         ClusterStateWaitUntilThresholdStep waitForClonedIndexGreenStep = new ClusterStateWaitUntilThresholdStep(
             new WaitForIndexColorStep(
                 waitForClonedIndexGreenKey,
                 forceMergeKey,
                 ClusterHealthStatus.GREEN,
-                FORCE_MERGED_INDEX_NAME_SUPPLIER
+                FORCE_MERGE_INDEX_NAME_SUPPLIER
             ),
             cleanupClonedIndexKey
         );
 
+        // Execute the force merge (either on the original index or the cloned index).
         ForceMergeStep forceMergeStep = new ForceMergeStep(forceMergeKey, segmentCountKey, client, maxNumSegments);
+        // This step simply logs whether the force-merge resulted in the desired number of segments or not.
         SegmentCountStep segmentCountStep = new SegmentCountStep(
             segmentCountKey,
             conditionalConfigureClonedIndexKey,
             client,
             maxNumSegments
         );
 
+        // If we cloned the index, we need to complete the setup of the cloned index and swap it with the original index.
+        // If we did not clone the index, there's nothing else for us to do.
         BranchingStep conditionalConfigureClonedIndexStep = new BranchingStep(
             conditionalConfigureClonedIndexKey,
-            nextStepKey, // If we didn't have a clone, there's nothing else for us to do
-            reconfigureReplicasKey, // Otherwise, we need to finish the setup of the cloned (and force-merged) index
+            nextStepKey,
+            reconfigureReplicasKey,
             (index, project) -> {
                 IndexMetadata indexMetadata = project.index(index);
                 assert indexMetadata != null : "index " + index.getName() + " must exist in the cluster state";
-                // Returns true if the index has a clone index name set, which means we need to reconfigure the cloned index
                 String cloneIndexName = indexMetadata.getLifecycleExecutionState().forceMergeIndexName();
                 if (cloneIndexName == null) {
                     return false;
                 }
-                // If for some reason the cloned index does not exist, we don't want to fail the next steps, so we skip them.
-                // This should not happen in ordinary circumstances.
+                // If for some reason the cloned index does not exist in the cluster state, we don't want to fail the next steps,
+                // so we skip them. This should not happen in ordinary circumstances.
                 boolean clonedIndexExists = project.index(cloneIndexName) != null;
                 assert clonedIndexExists
                     : "index [" + index.getName() + "] has cloned index name [" + cloneIndexName + "] but it does not exist in the cluster";
                 return clonedIndexExists;
             }
         );
 
-        // Reset the number of replicas to the value of the original index and remove the write block
+        // Reset the number of replicas to the value of the original index and remove the write block.
         UpdateSettingsStep reconfigureReplicasStep = new UpdateSettingsStep(
             reconfigureReplicasKey,
             copyMetadataKey,
             client,
-            FORCE_MERGED_INDEX_NAME_SUPPLIER,
+            FORCE_MERGE_INDEX_NAME_SUPPLIER,
             (indexMetadata) -> Settings.builder()
-                .put(
-                    IndexMetadata.SETTING_NUMBER_OF_REPLICAS,
-                    indexMetadata.getSettings().get(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, "1")
-                )
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, indexMetadata.getNumberOfReplicas())
                 .put(IndexMetadata.INDEX_BLOCKS_WRITE_SETTING.getKey(), (String) null)
                 .build()
         );
+        // Copy the lifecycle execution state from the original index to the cloned index. This also remove the skip setting so that
+        // the cloned index can continue through the ILM phases.
         CopyExecutionStateStep copyMetadata = new CopyExecutionStateStep(
             copyMetadataKey,
             aliasDataStreamBranchingKey,
-            FORCE_MERGED_INDEX_NAME_SUPPLIER,
+            FORCE_MERGE_INDEX_NAME_SUPPLIER,
             nextStepKey
         );
-        // by the time we get to this step we have 2 indices, the source and the cloned one. we now need to choose an index
-        // swapping strategy such that the cloned index takes the place of the source index (which is also deleted).
-        // if the source index is part of a data stream it's a matter of replacing it with the cloned index one in the data stream and
-        // then deleting the source index; otherwise we'll use the alias management api to atomically transfer the aliases from source to
-        // the cloned index and delete the source
+        // By the time we get to this step we have 2 indices, the source and the cloned one. We now need to choose an index
+        // swapping strategy such that the cloned index takes the place of the source index (which should also be deleted).
+        // If the source index is part of a data stream it's a matter of replacing it with the cloned index one in the data stream and
+        // then deleting the source index; otherwise we'll use the alias management API to atomically transfer the aliases from
+        // the source index to the cloned index and delete the source.
         BranchingStep aliasDataStreamBranchingStep = new BranchingStep(
             aliasDataStreamBranchingKey,
             aliasSwapAndDeleteKey,
@@ -325,13 +335,13 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
             aliasSwapAndDeleteKey,
             nextStepKey,
             client,
-            FORCE_MERGED_INDEX_NAME_SUPPLIER,
+            FORCE_MERGE_INDEX_NAME_SUPPLIER,
             true
         );
         ReplaceDataStreamBackingIndexStep replaceDataStreamBackingIndexStep = new ReplaceDataStreamBackingIndexStep(
             replaceDataStreamIndexKey,
             deleteSourceIndexKey,
-            FORCE_MERGED_INDEX_NAME_SUPPLIER
+            FORCE_MERGE_INDEX_NAME_SUPPLIER
         );
         DeleteStep deleteSourceIndexStep = new DeleteStep(deleteSourceIndexKey, nextStepKey, client);
 

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

@@ -51,6 +51,7 @@ public void performAction(
         ClusterStateObserver observer,
         ActionListener<Void> listener
     ) {
+        // Use the cloned index name if we have one, otherwise fall back to the original index name.
         String clonedIndexName = indexMetadata.getLifecycleExecutionState().forceMergeIndexName();
         String indexName = clonedIndexName != null ? clonedIndexName : indexMetadata.getIndex().getName();
         ForceMergeRequest request = new ForceMergeRequest(indexName);

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

@@ -57,6 +57,7 @@ public int getMaxNumSegments() {
 
     @Override
     public void evaluateCondition(ProjectState state, IndexMetadata indexMetadata, Listener listener, TimeValue masterTimeout) {
+        // Use the cloned index name if we have one, otherwise fall back to the original index name.
         String clonedIndexName = indexMetadata.getLifecycleExecutionState().forceMergeIndexName();
         String forceMergedIndexName = clonedIndexName != null ? clonedIndexName : indexMetadata.getIndex().getName();
         getClient(state.projectId()).admin()

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/PhaseCacheManagementTests.java

@@ -260,7 +260,6 @@ public void testReadStepKeys() {
         String phaseDef = Strings.toString(pei);
         logger.info("--> phaseDef: {}", phaseDef);
 
-        assumeFalse("", true);
         assertThat(
             readStepKeys(REGISTRY, client, phaseDef, "phase", null),
             contains(
@@ -269,10 +268,25 @@ public void testReadStepKeys() {
                 new Step.StepKey("phase", "forcemerge", ForceMergeAction.CONDITIONAL_SKIP_FORCE_MERGE_STEP),
                 new Step.StepKey("phase", "forcemerge", CheckNotDataStreamWriteIndexStep.NAME),
                 new Step.StepKey("phase", "forcemerge", WaitUntilTimeSeriesEndTimePassesStep.NAME),
-                // This read-only key is now a noop step but we preserved it for backwards compatibility
-                new Step.StepKey("phase", "forcemerge", ReadOnlyAction.NAME),
-                new Step.StepKey("phase", "forcemerge", ForceMergeAction.NAME),
-                new Step.StepKey("phase", "forcemerge", SegmentCountStep.NAME)
+                new Step.StepKey("phase", "forcemerge", ForceMergeAction.CONDITIONAL_SKIP_CLONE_STEP),
+                new Step.StepKey("phase", "forcemerge", CloseIndexStep.NAME),
+                new Step.StepKey("phase", "forcemerge", ForceMergeAction.UPDATE_COMPRESSION_SETTINGS_STEP),
+                new Step.StepKey("phase", "forcemerge", OpenIndexStep.NAME),
+                new Step.StepKey("phase", "forcemerge", ForceMergeAction.WAIT_FOR_COMPRESSION_SETTINGS_GREEN),
+                new Step.StepKey("phase", "forcemerge", CleanupClonedIndexStep.NAME),
+                new Step.StepKey("phase", "forcemerge", ReadOnlyStep.NAME),
+                new Step.StepKey("phase", "forcemerge", GenerateUniqueIndexNameStep.NAME),
+                new Step.StepKey("phase", "forcemerge", CloneIndexStep.NAME),
+                new Step.StepKey("phase", "forcemerge", ForceMergeAction.WAIT_FOR_CLONED_INDEX_GREEN),
+                new Step.StepKey("phase", "forcemerge", ForceMergeStep.NAME),
+                new Step.StepKey("phase", "forcemerge", SegmentCountStep.NAME),
+                new Step.StepKey("phase", "forcemerge", ForceMergeAction.CONDITIONAL_CONFIGURE_CLONED_INDEX_STEP),
+                new Step.StepKey("phase", "forcemerge", ForceMergeAction.UPDATE_CLONED_INDEX_SETTINGS_STEP),
+                new Step.StepKey("phase", "forcemerge", CopyExecutionStateStep.NAME),
+                new Step.StepKey("phase", "forcemerge", ForceMergeAction.CONDITIONAL_DATA_STREAM_CHECK_STEP),
+                new Step.StepKey("phase", "forcemerge", ShrinkSetAliasStep.NAME),
+                new Step.StepKey("phase", "forcemerge", ReplaceDataStreamBackingIndexStep.NAME),
+                new Step.StepKey("phase", "forcemerge", DeleteStep.NAME)
             )
         );
     }