Retry downsample ILM action using a new target index (#94965)

csoulios · web-flow · commit e9cfd815070a · 2023-05-15T11:10:40.000+03:00
Currently, when the ILM downample step is being retried, the same target index is used. This can cause the subsequent downsample API invocation to index rolled up data into shards of the target index that already exists and while the previous downsample API invocation is still partially running (and also rolling up data into the same target shard). Note that, the downsample step may fail in case a cluster is being restarted in a rolling manner (for example for an upgrade) or when the elected master node fails (the downsample action is coordinated from the elected master node). This PR modfies the ILM DownsampleAction so that when DownsampleStep fails, it will retry by going performing the following steps 1. Cleanup existing target index, 2. Generate a new index name for the target index 3. Downsample using the new target index name. Note 1: This change may leave some garbage indices that we must find another way to cleanup. However, the downsample process will become more resilient. Note 2: A similar approach is used by the searchable_snapshot ILM action Closes #93580
diff --git a/docs/changelog/94965.yaml b/docs/changelog/94965.yaml
@@ -0,0 +1,6 @@
+pr: 94965
+summary: Retry downsample ILM action using a new target index
+area: "ILM+SLM"
+type: bug
+issues:
+ - 93580
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/LifecycleExecutionState.java b/server/src/main/java/org/elasticsearch/cluster/metadata/LifecycleExecutionState.java
@@ -81,7 +81,7 @@ public static Builder builder(LifecycleExecutionState state) {
             .setSnapshotName(state.snapshotName)
             .setShrinkIndexName(state.shrinkIndexName)
             .setSnapshotIndexName(state.snapshotIndexName)
-            .setRollupIndexName(state.downsampleIndexName)
+            .setDownsampleIndexName(state.downsampleIndexName)
             .setStepTime(state.stepTime);
     }
 
@@ -187,9 +187,9 @@ public static LifecycleExecutionState fromCustomMetadata(Map<String, String> cus
         if (snapshotIndexName != null) {
             builder.setSnapshotIndexName(snapshotIndexName);
         }
-        String rollupIndexName = customData.get(DOWNSAMPLE_INDEX_NAME);
-        if (rollupIndexName != null) {
-            builder.setRollupIndexName(rollupIndexName);
+        String downsampleIndexName = customData.get(DOWNSAMPLE_INDEX_NAME);
+        if (downsampleIndexName != null) {
+            builder.setDownsampleIndexName(downsampleIndexName);
         }
         return builder.build();
     }
@@ -273,7 +273,7 @@ public static class Builder {
         private String snapshotRepository;
         private String shrinkIndexName;
         private String snapshotIndexName;
-        private String rollupIndexName;
+        private String downsampleIndexName;
 
         public Builder setPhase(String phase) {
             this.phase = phase;
@@ -355,8 +355,8 @@ public Builder setSnapshotIndexName(String snapshotIndexName) {
             return this;
         }
 
-        public Builder setRollupIndexName(String rollupIndexName) {
-            this.rollupIndexName = rollupIndexName;
+        public Builder setDownsampleIndexName(String downsampleIndexName) {
+            this.downsampleIndexName = downsampleIndexName;
             return this;
         }
 
@@ -378,7 +378,7 @@ public LifecycleExecutionState build() {
                 snapshotName,
                 shrinkIndexName,
                 snapshotIndexName,
-                rollupIndexName
+                downsampleIndexName
             );
         }
     }
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleAction.java
@@ -106,10 +106,10 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
         StepKey checkNotWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
         StepKey waitForNoFollowerStepKey = new StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
         StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyStep.NAME);
-        StepKey cleanupRollupIndexKey = new StepKey(phase, NAME, CleanupTargetIndexStep.NAME);
-        StepKey generateRollupIndexNameKey = new StepKey(phase, NAME, GENERATE_DOWNSAMPLE_STEP_NAME);
-        StepKey rollupKey = new StepKey(phase, NAME, DownsampleStep.NAME);
-        StepKey waitForRollupIndexKey = new StepKey(phase, NAME, WaitForIndexColorStep.NAME);
+        StepKey cleanupDownsampleIndexKey = new StepKey(phase, NAME, CleanupTargetIndexStep.NAME);
+        StepKey generateDownsampleIndexNameKey = new StepKey(phase, NAME, GENERATE_DOWNSAMPLE_STEP_NAME);
+        StepKey downsampleKey = new StepKey(phase, NAME, DownsampleStep.NAME);
+        StepKey waitForDownsampleIndexKey = new StepKey(phase, NAME, WaitForIndexColorStep.NAME);
         StepKey copyMetadataKey = new StepKey(phase, NAME, CopyExecutionStateStep.NAME);
         StepKey dataStreamCheckBranchingKey = new StepKey(phase, NAME, CONDITIONAL_DATASTREAM_CHECK_KEY);
         StepKey replaceDataStreamIndexKey = new StepKey(phase, NAME, ReplaceDataStreamBackingIndexStep.NAME);
@@ -132,44 +132,51 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
             checkNotWriteIndex,
             waitForNoFollowerStepKey
         );
-        WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(waitForNoFollowerStepKey, cleanupRollupIndexKey, client);
-
-        // We generate a unique rollup index name, but we also retry if the allocation of the rollup index is not possible, so we want to
-        // delete the "previously generated" rollup index (this is a no-op if it's the first run of the action, and we haven't generated a
-        // rollup index name)
-        CleanupTargetIndexStep cleanupRollupIndexStep = new CleanupTargetIndexStep(
-            cleanupRollupIndexKey,
-            readOnlyKey,
+        WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(waitForNoFollowerStepKey, readOnlyKey, client);
+
+        // Mark source index as read-only
+        ReadOnlyStep readOnlyStep = new ReadOnlyStep(readOnlyKey, cleanupDownsampleIndexKey, client);
+
+        // We generate a unique downsample index name, but we also retry if the allocation of the downsample index
+        // is not possible, so we want to delete the "previously generated" downsample index (this is a no-op if it's
+        // the first run of the action, and we haven't generated a downsample index name)
+        CleanupTargetIndexStep cleanupDownsampleIndexStep = new CleanupTargetIndexStep(
+            cleanupDownsampleIndexKey,
+            generateDownsampleIndexNameKey,
             client,
             (indexMetadata) -> IndexMetadata.INDEX_DOWNSAMPLE_SOURCE_NAME.get(indexMetadata.getSettings()),
             (indexMetadata) -> indexMetadata.getLifecycleExecutionState().downsampleIndexName()
         );
-        // Mark source index as read-only
-        ReadOnlyStep readOnlyStep = new ReadOnlyStep(readOnlyKey, generateRollupIndexNameKey, client);
 
-        // Generate a unique rollup index name and store it in the ILM execution state
-        GenerateUniqueIndexNameStep generateRollupIndexNameStep = new GenerateUniqueIndexNameStep(
-            generateRollupIndexNameKey,
-            rollupKey,
+        // Generate a unique downsample index name and store it in the ILM execution state
+        GenerateUniqueIndexNameStep generateDownsampleIndexNameStep = new GenerateUniqueIndexNameStep(
+            generateDownsampleIndexNameKey,
+            downsampleKey,
             DOWNSAMPLED_INDEX_PREFIX,
-            (rollupIndexName, lifecycleStateBuilder) -> lifecycleStateBuilder.setRollupIndexName(rollupIndexName)
+            (downsampleIndexName, lifecycleStateBuilder) -> lifecycleStateBuilder.setDownsampleIndexName(downsampleIndexName)
         );
 
-        // Here is where the actual rollup action takes place
-        DownsampleStep rollupStep = new DownsampleStep(rollupKey, waitForRollupIndexKey, client, fixedInterval);
+        // Here is where the actual downsample action takes place
+        DownsampleStep downsampleStep = new DownsampleStep(
+            downsampleKey,
+            waitForDownsampleIndexKey,
+            cleanupDownsampleIndexKey,
+            client,
+            fixedInterval
+        );
 
         // Wait until the downsampled index is recovered. We again wait until the configured threshold is breached and
-        // if the downsampled index has not successfully recovered until then, we rewind to the "cleanup-rollup-index"
-        // step to delete this unsuccessful downsampled index and retry the operation by generating a new downsampled index
+        // if the downsampled index has not successfully recovered until then, we rewind to the "cleanup-downsample-index"
+        // step to delete this unsuccessful downsampled index and retry the operation by generating a new downsample index
         // name and attempting to downsample again.
-        ClusterStateWaitUntilThresholdStep rollupAllocatedStep = new ClusterStateWaitUntilThresholdStep(
+        ClusterStateWaitUntilThresholdStep downsampleAllocatedStep = new ClusterStateWaitUntilThresholdStep(
             new WaitForIndexColorStep(
-                waitForRollupIndexKey,
+                waitForDownsampleIndexKey,
                 copyMetadataKey,
                 ClusterHealthStatus.YELLOW,
                 (indexName, lifecycleState) -> lifecycleState.downsampleIndexName()
             ),
-            cleanupRollupIndexKey
+            cleanupDownsampleIndexKey
         );
 
         CopyExecutionStateStep copyExecutionStateStep = new CopyExecutionStateStep(
@@ -213,11 +220,11 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
             isTimeSeriesIndexBranchingStep,
             checkNotWriteIndexStep,
             waitForNoFollowersStep,
-            cleanupRollupIndexStep,
             readOnlyStep,
-            generateRollupIndexNameStep,
-            rollupStep,
-            rollupAllocatedStep,
+            cleanupDownsampleIndexStep,
+            generateDownsampleIndexNameStep,
+            downsampleStep,
+            downsampleAllocatedStep,
             copyExecutionStateStep,
             isDataStreamBranchingStep,
             replaceDataStreamBackingIndex,
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleStep.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/DownsampleStep.java
@@ -9,7 +9,6 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateObserver;
@@ -35,9 +34,20 @@ public class DownsampleStep extends AsyncActionStep {
     private static final Logger logger = LogManager.getLogger(DownsampleStep.class);
 
     private final DateHistogramInterval fixedInterval;
+    private final StepKey nextStepOnSuccess;
+    private final StepKey nextStepOnFailure;
+    private volatile boolean downsampleFailed;
 
-    public DownsampleStep(StepKey key, StepKey nextStepKey, Client client, DateHistogramInterval fixedInterval) {
-        super(key, nextStepKey, client);
+    public DownsampleStep(
+        StepKey key,
+        StepKey nextStepOnSuccess,
+        StepKey nextStepOnFailure,
+        Client client,
+        DateHistogramInterval fixedInterval
+    ) {
+        super(key, null, client);
+        this.nextStepOnSuccess = nextStepOnSuccess;
+        this.nextStepOnFailure = nextStepOnFailure;
         this.fixedInterval = fixedInterval;
     }
 
@@ -62,72 +72,63 @@ public void performAction(
         final String indexName = indexMetadata.getIndex().getName();
         final String downsampleIndexName = lifecycleState.downsampleIndexName();
         if (Strings.hasText(downsampleIndexName) == false) {
+            downsampleFailed = true;
             listener.onFailure(
                 new IllegalStateException(
-                    "rollup index name was not generated for policy [" + policyName + "] and index [" + indexName + "]"
+                    "downsample index name was not generated for policy [" + policyName + "] and index [" + indexName + "]"
                 )
             );
             return;
         }
 
-        IndexMetadata rollupIndexMetadata = currentState.metadata().index(downsampleIndexName);
-        if (rollupIndexMetadata != null) {
-            IndexMetadata.DownsampleTaskStatus rollupIndexStatus = IndexMetadata.INDEX_DOWNSAMPLE_STATUS.get(
-                rollupIndexMetadata.getSettings()
+        IndexMetadata downsampleIndexMetadata = currentState.metadata().index(downsampleIndexName);
+        if (downsampleIndexMetadata != null) {
+            IndexMetadata.DownsampleTaskStatus downsampleIndexStatus = IndexMetadata.INDEX_DOWNSAMPLE_STATUS.get(
+                downsampleIndexMetadata.getSettings()
             );
-            // Rollup index has already been created with the generated name and its status is "success".
-            // So we skip index rollup creation.
-            if (IndexMetadata.DownsampleTaskStatus.SUCCESS.equals(rollupIndexStatus)) {
+            if (IndexMetadata.DownsampleTaskStatus.SUCCESS.equals(downsampleIndexStatus)) {
+                // Downsample index has already been created with the generated name and its status is "success".
+                // So we skip index downsample creation.
                 logger.warn(
-                    "skipping [{}] step for index [{}] as part of policy [{}] as the rollup index [{}] already exists",
+                    "skipping [{}] step for index [{}] as part of policy [{}] as the downsample index [{}] already exists",
                     DownsampleStep.NAME,
                     indexName,
                     policyName,
                     downsampleIndexName
                 );
                 listener.onResponse(null);
             } else {
-                logger.warn(
-                    "[{}] step for index [{}] as part of policy [{}] found the rollup index [{}] already exists. Deleting it.",
-                    DownsampleStep.NAME,
-                    indexName,
-                    policyName,
-                    downsampleIndexName
+                // Downsample index has already been created with the generated name but its status is not "success".
+                // So we fail this step so that we go back to cleaning up the index and try again with a new downsample
+                // index name.
+                downsampleFailed = true;
+                listener.onFailure(
+                    new IllegalStateException(
+                        "failing ["
+                            + DownsampleStep.NAME
+                            + "] step for index ["
+                            + indexName
+                            + "] as part of policy ["
+                            + policyName
+                            + "] because the downsample index ["
+                            + downsampleIndexName
+                            + "] already exists with downsample status ["
+                            + downsampleIndexStatus
+                            + "]"
+                    )
                 );
-                // Rollup index has already been created with the generated name but its status is not "success".
-                // So we delete the index and proceed with executing the rollup step.
-                DeleteIndexRequest deleteRequest = new DeleteIndexRequest(downsampleIndexName);
-                getClient().admin().indices().delete(deleteRequest, ActionListener.wrap(response -> {
-                    if (response.isAcknowledged()) {
-                        performDownsampleIndex(indexName, downsampleIndexName, listener);
-                    } else {
-                        listener.onFailure(
-                            new IllegalStateException(
-                                "failing ["
-                                    + DownsampleStep.NAME
-                                    + "] step for index ["
-                                    + indexName
-                                    + "] as part of policy ["
-                                    + policyName
-                                    + "] because the rollup index ["
-                                    + downsampleIndexName
-                                    + "] already exists with rollup status ["
-                                    + rollupIndexStatus
-                                    + "]"
-                            )
-                        );
-                    }
-                }, listener::onFailure));
             }
-            return;
+        } else {
+            performDownsampleIndex(indexName, downsampleIndexName, ActionListener.wrap(listener::onResponse, e -> {
+                downsampleFailed = true;
+                listener.onFailure(e);
+            }));
         }
-
-        performDownsampleIndex(indexName, downsampleIndexName, listener);
     }
 
-    private void performDownsampleIndex(String indexName, String rollupIndexName, ActionListener<Void> listener) {
+    void performDownsampleIndex(String indexName, String downsampleIndexName, ActionListener<Void> listener) {
         DownsampleConfig config = new DownsampleConfig(fixedInterval);
-        DownsampleAction.Request request = new DownsampleAction.Request(indexName, rollupIndexName, config).masterNodeTimeout(
+        DownsampleAction.Request request = new DownsampleAction.Request(indexName, downsampleIndexName, config).masterNodeTimeout(
             TimeValue.MAX_VALUE
         );
         // Currently, DownsampleAction always acknowledges action was complete when no exceptions are thrown.
@@ -138,24 +139,35 @@ private void performDownsampleIndex(String indexName, String rollupIndexName, Ac
         );
     }
 
+    @Override
+    public final StepKey getNextStepKey() {
+        return downsampleFailed ? nextStepOnFailure : nextStepOnSuccess;
+    }
+
     public DateHistogramInterval getFixedInterval() {
         return fixedInterval;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(super.hashCode(), fixedInterval);
+        return Objects.hash(super.hashCode(), fixedInterval, nextStepOnSuccess, nextStepOnFailure);
     }
 
     @Override
     public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
         if (obj == null) {
             return false;
         }
         if (getClass() != obj.getClass()) {
             return false;
         }
         DownsampleStep other = (DownsampleStep) obj;
-        return super.equals(obj) && Objects.equals(fixedInterval, other.fixedInterval);
+        return super.equals(obj)
+            && Objects.equals(fixedInterval, other.fixedInterval)
+            && Objects.equals(nextStepOnSuccess, other.nextStepOnSuccess)
+            && Objects.equals(nextStepOnFailure, other.nextStepOnFailure);
     }
 }
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/DownsampleActionTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/DownsampleActionTests.java
@@ -76,14 +76,14 @@ public void testToSteps() {
 
         assertTrue(steps.get(2) instanceof WaitForNoFollowersStep);
         assertThat(steps.get(2).getKey().name(), equalTo(WaitForNoFollowersStep.NAME));
-        assertThat(steps.get(2).getNextStepKey().name(), equalTo(CleanupTargetIndexStep.NAME));
+        assertThat(steps.get(2).getNextStepKey().name(), equalTo(ReadOnlyStep.NAME));
 
-        assertTrue(steps.get(3) instanceof CleanupTargetIndexStep);
-        assertThat(steps.get(3).getKey().name(), equalTo(CleanupTargetIndexStep.NAME));
-        assertThat(steps.get(3).getNextStepKey().name(), equalTo(ReadOnlyStep.NAME));
+        assertTrue(steps.get(3) instanceof ReadOnlyStep);
+        assertThat(steps.get(3).getKey().name(), equalTo(ReadOnlyStep.NAME));
+        assertThat(steps.get(3).getNextStepKey().name(), equalTo(CleanupTargetIndexStep.NAME));
 
-        assertTrue(steps.get(4) instanceof ReadOnlyStep);
-        assertThat(steps.get(4).getKey().name(), equalTo(ReadOnlyStep.NAME));
+        assertTrue(steps.get(4) instanceof CleanupTargetIndexStep);
+        assertThat(steps.get(4).getKey().name(), equalTo(CleanupTargetIndexStep.NAME));
         assertThat(steps.get(4).getNextStepKey().name(), equalTo(GENERATE_DOWNSAMPLE_STEP_NAME));
 
         assertTrue(steps.get(5) instanceof GenerateUniqueIndexNameStep);
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/DownsampleStepTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/DownsampleStepTests.java
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/LifecycleExecutionStateTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ilm/LifecycleExecutionStateTests.java

Original file line number	Diff line number	Diff line change
`@@ -81,7 +81,7 @@ public static Builder builder(LifecycleExecutionState state) {`
`81`	`81`	`.setSnapshotName(state.snapshotName)`
`82`	`82`	`.setShrinkIndexName(state.shrinkIndexName)`
`83`	`83`	`.setSnapshotIndexName(state.snapshotIndexName)`
`84`		`- .setRollupIndexName(state.downsampleIndexName)`
	`84`	`+ .setDownsampleIndexName(state.downsampleIndexName)`
`85`	`85`	`.setStepTime(state.stepTime);`
`86`	`86`	`}`
`87`	`87`
`@@ -187,9 +187,9 @@ public static LifecycleExecutionState fromCustomMetadata(Map<String, String> cus`
`187`	`187`	`if (snapshotIndexName != null) {`
`188`	`188`	`builder.setSnapshotIndexName(snapshotIndexName);`
`189`	`189`	`}`
`190`		`- String rollupIndexName = customData.get(DOWNSAMPLE_INDEX_NAME);`
`191`		`- if (rollupIndexName != null) {`
`192`		`- builder.setRollupIndexName(rollupIndexName);`
	`190`	`+ String downsampleIndexName = customData.get(DOWNSAMPLE_INDEX_NAME);`
	`191`	`+ if (downsampleIndexName != null) {`
	`192`	`+ builder.setDownsampleIndexName(downsampleIndexName);`
`193`	`193`	`}`
`194`	`194`	`return builder.build();`
`195`	`195`	`}`
`@@ -273,7 +273,7 @@ public static class Builder {`
`273`	`273`	`private String snapshotRepository;`
`274`	`274`	`private String shrinkIndexName;`
`275`	`275`	`private String snapshotIndexName;`
`276`		`- private String rollupIndexName;`
	`276`	`+ private String downsampleIndexName;`
`277`	`277`
`278`	`278`	`public Builder setPhase(String phase) {`
`279`	`279`	`this.phase = phase;`
`@@ -355,8 +355,8 @@ public Builder setSnapshotIndexName(String snapshotIndexName) {`
`355`	`355`	`return this;`
`356`	`356`	`}`
`357`	`357`
`358`		`- public Builder setRollupIndexName(String rollupIndexName) {`
`359`		`- this.rollupIndexName = rollupIndexName;`
	`358`	`+ public Builder setDownsampleIndexName(String downsampleIndexName) {`
	`359`	`+ this.downsampleIndexName = downsampleIndexName;`
`360`	`360`	`return this;`
`361`	`361`	`}`
`362`	`362`
`@@ -378,7 +378,7 @@ public LifecycleExecutionState build() {`
`378`	`378`	`snapshotName,`
`379`	`379`	`shrinkIndexName,`
`380`	`380`	`snapshotIndexName,`
`381`		`- rollupIndexName`
	`381`	`+ downsampleIndexName`
`382`	`382`	`);`
`383`	`383`	`}`
`384`	`384`	`}`