Retry ILM steps that fail due to SnapshotInProgressException

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/AbstractUnfollowIndexStep.java

@@ -7,6 +7,7 @@
 
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 
 import java.util.Map;
@@ -20,7 +21,8 @@ abstract class AbstractUnfollowIndexStep extends AsyncActionStep {
     }
 
     @Override
-    public final void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState, Listener listener) {
+    public final void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState,
+                                    ClusterStateObserver observer, Listener listener) {
         String followerIndex = indexMetaData.getIndex().getName();
         Map<String, String> customIndexMetadata = indexMetaData.getCustomData(CCR_METADATA_KEY);
         if (customIndexMetadata == null) {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/AsyncActionStep.java

@@ -7,6 +7,7 @@
 
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 
 /**
@@ -29,7 +30,8 @@ public boolean indexSurvives() {
         return true;
     }
 
-    public abstract void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState, Listener listener);
+    public abstract void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState,
+                                       ClusterStateObserver observer, Listener listener);
 
     public interface Listener {
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/AsyncRetryDuringSnapshotActionStep.java

@@ -0,0 +1,165 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.core.indexlifecycle;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
+import org.elasticsearch.cluster.SnapshotsInProgress;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.repositories.IndexId;
+import org.elasticsearch.snapshots.SnapshotInProgressException;
+
+import java.util.function.Consumer;
+
+/**
+ * This is an abstract AsyncActionStep that wraps the performed action listener, checking to see
+ * if the action fails due to a snapshot being in progress. If a snapshot is in progress, it
+ * registers an observer and waits to try again when a snapshot is no longer running.
+ */
+public abstract class AsyncRetryDuringSnapshotActionStep extends AsyncActionStep {
+    private final Logger logger = LogManager.getLogger(AsyncRetryDuringSnapshotActionStep.class);
+
+    public AsyncRetryDuringSnapshotActionStep(StepKey key, StepKey nextStepKey, Client client) {
+        super(key, nextStepKey, client);
+    }
+
+    @Override
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState,
+                              ClusterStateObserver observer, Listener listener) {
+        // Wrap the original listener to handle exceptions caused by ongoing snapshots
+        SnapshotExceptionListener snapshotExceptionListener = new SnapshotExceptionListener(indexMetaData.getIndex(), listener, observer);
+        performDuringNoSnapshot(indexMetaData, currentClusterState, snapshotExceptionListener);
+    }
+
+    /**
+     * Method to be performed during which no snapshots for the index are already underway.
+     */
+    abstract void performDuringNoSnapshot(IndexMetaData indexMetaData, ClusterState currentClusterState, Listener listener);
+
+    /**
+     * SnapshotExceptionListener is an injected listener wrapper that checks to see if a particular
+     * action failed due to a {@code SnapshotInProgressException}. If it did, then it registers a
+     * ClusterStateObserver listener waiting for the next time the snapshot is not running,
+     * re-running the step's {@link #performAction(IndexMetaData, ClusterState, ClusterStateObserver, Listener)}
+     * method when the snapshot is no longer running.
+     */
+    class SnapshotExceptionListener implements AsyncActionStep.Listener {
+        private final Index index;
+        private final Listener originalListener;
+        private final ClusterStateObserver observer;
+
+        SnapshotExceptionListener(Index index, Listener originalListener, ClusterStateObserver observer) {
+            this.index = index;
+            this.originalListener = originalListener;
+            this.observer = observer;
+        }
+
+        @Override
+        public void onResponse(boolean complete) {
+            originalListener.onResponse(complete);
+        }
+
+        @Override
+        public void onFailure(Exception e) {
+            if (e instanceof SnapshotInProgressException) {
+                try {
+                    logger.debug("[{}] attempted to run ILM step but a snapshot is in progress, step will retry at a later time",
+                        index.getName());
+                    observer.waitForNextChange(
+                        new NoSnapshotRunningListener(observer, index.getName(), state -> {
+                            IndexMetaData idxMeta = state.metaData().index(index);
+                            if (idxMeta == null) {
+                                // The index has since been deleted, mission accomplished!
+                                originalListener.onResponse(true);
+                            }
+                            // Re-invoke the performAction method with the new state
+                            performAction(idxMeta, state, observer, originalListener);
+                        }, originalListener::onFailure),
+                        // TODO: what is a good timeout value for no new state received during this time?
+                        TimeValue.timeValueHours(12));
+                } catch (Exception secondError) {
+                    // There was a second error trying to set up an observer,
+                    // fail the original listener
+                    secondError.addSuppressed(e);
+                    originalListener.onFailure(secondError);
+                }
+            } else {
+                originalListener.onFailure(e);
+            }
+        }
+    }
+
+    /**
+     * A {@link ClusterStateObserver.Listener} that invokes the given function with the new state,
+     * once no snapshots are running. If a snapshot is still running it registers a new listener
+     * and tries again. Passes any exceptions to the original exception listener if they occur.
+     */
+    class NoSnapshotRunningListener implements ClusterStateObserver.Listener {
+
+        private final Consumer<ClusterState> reRun;
+        private final Consumer<Exception> exceptionConsumer;
+        private final ClusterStateObserver observer;
+        private final String indexName;
+
+        NoSnapshotRunningListener(ClusterStateObserver observer, String indexName,
+                                  Consumer<ClusterState> reRun,
+                                  Consumer<Exception> exceptionConsumer) {
+            this.observer = observer;
+            this.reRun = reRun;
+            this.exceptionConsumer = exceptionConsumer;
+            this.indexName = indexName;
+        }
+
+        @Override
+        public void onNewClusterState(ClusterState state) {
+            try {
+                if (snapshotInProgress(state)) {
+                    observer.waitForNextChange(this);
+                } else {
+                    logger.debug("[{}] retrying ILM step after snapshot has completed", indexName);
+                    reRun.accept(state);
+                }
+            } catch (Exception e) {
+                exceptionConsumer.accept(e);
+            }
+        }
+
+        private boolean snapshotInProgress(ClusterState state) {
+            SnapshotsInProgress snapshotsInProgress = state.custom(SnapshotsInProgress.TYPE);
+            if (snapshotsInProgress == null || snapshotsInProgress.entries().isEmpty()) {
+                // No snapshots are running, new state is acceptable to proceed
+                return false;
+            }
+
+            for (SnapshotsInProgress.Entry snapshot : snapshotsInProgress.entries()) {
+                if (snapshot.indices().stream()
+                    .map(IndexId::getName)
+                    .anyMatch(name -> name.equals(indexName))) {
+                    // There is a snapshot running with this index name
+                    return true;
+                }
+            }
+            // There are snapshots, but none for this index, so it's okay to proceed with this state
+            return false;
+        }
+
+        @Override
+        public void onClusterServiceClose() {
+            // This means the cluster is being shut down, so nothing to do here
+        }
+
+        @Override
+        public void onTimeout(TimeValue timeout) {
+            exceptionConsumer.accept(new IllegalStateException("step timed out while waiting for snapshots to complete"));
+        }
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/CloseFollowerIndexStep.java

@@ -8,8 +8,14 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.close.CloseIndexRequest;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
 
-final class CloseFollowerIndexStep extends AbstractUnfollowIndexStep {
+import java.util.Map;
+
+import static org.elasticsearch.xpack.core.indexlifecycle.UnfollowAction.CCR_METADATA_KEY;
+
+final class CloseFollowerIndexStep extends AsyncRetryDuringSnapshotActionStep {
 
     static final String NAME = "close-follower-index";
 
@@ -18,7 +24,14 @@ final class CloseFollowerIndexStep extends AbstractUnfollowIndexStep {
     }
 
     @Override
-    void innerPerformAction(String followerIndex, Listener listener) {
+    void performDuringNoSnapshot(IndexMetaData indexMetaData, ClusterState currentClusterState, Listener listener) {
+        String followerIndex = indexMetaData.getIndex().getName();
+        Map<String, String> customIndexMetadata = indexMetaData.getCustomData(CCR_METADATA_KEY);
+        if (customIndexMetadata == null) {
+            listener.onResponse(true);
+            return;
+        }
+
         CloseIndexRequest closeIndexRequest = new CloseIndexRequest(followerIndex);
         getClient().admin().indices().close(closeIndexRequest, ActionListener.wrap(
             r -> {

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

@@ -14,18 +14,18 @@
 /**
  * Deletes a single index.
  */
-public class DeleteStep extends AsyncActionStep {
+public class DeleteStep extends AsyncRetryDuringSnapshotActionStep {
     public static final String NAME = "delete";
 
     public DeleteStep(StepKey key, StepKey nextStepKey, Client client) {
         super(key, nextStepKey, client);
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
+    public void performDuringNoSnapshot(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
         getClient().admin().indices()
             .delete(new DeleteIndexRequest(indexMetaData.getIndex().getName()),
-                ActionListener.wrap(response -> listener.onResponse(true) , listener::onFailure));
+                ActionListener.wrap(response -> listener.onResponse(true), listener::onFailure));
     }
 
     @Override

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

@@ -9,6 +9,7 @@
 import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeRequest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 
 import java.util.Objects;
@@ -30,7 +31,7 @@ public int getMaxNumSegments() {
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, ClusterStateObserver observer, Listener listener) {
         ForceMergeRequest request = new ForceMergeRequest(indexMetaData.getIndex().getName());
         request.maxNumSegments(maxNumSegments);
         getClient().admin().indices()

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/FreezeStep.java

@@ -14,15 +14,15 @@
 /**
  * Freezes an index.
  */
-public class FreezeStep extends AsyncActionStep {
+public class FreezeStep extends AsyncRetryDuringSnapshotActionStep {
     public static final String NAME = "freeze";
 
     public FreezeStep(StepKey key, StepKey nextStepKey, Client client) {
         super(key, nextStepKey, client);
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
+    public void performDuringNoSnapshot(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
         getClient().admin().indices().execute(TransportFreezeIndexAction.FreezeIndexAction.INSTANCE,
             new TransportFreezeIndexAction.FreezeRequest(indexMetaData.getIndex().getName()),
             ActionListener.wrap(response -> listener.onResponse(true), listener::onFailure));

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/OpenFollowerIndexStep.java

@@ -9,6 +9,7 @@
 import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 
 final class OpenFollowerIndexStep extends AsyncActionStep {
@@ -20,7 +21,8 @@ final class OpenFollowerIndexStep extends AsyncActionStep {
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState, Listener listener) {
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState,
+                              ClusterStateObserver observer, Listener listener) {
         OpenIndexRequest request = new OpenIndexRequest(indexMetaData.getIndex().getName());
         getClient().admin().indices().open(request, ActionListener.wrap(
             r -> {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/RolloverStep.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.Strings;
 
@@ -30,7 +31,8 @@ public RolloverStep(StepKey key, StepKey nextStepKey, Client client) {
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState, Listener listener) {
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentClusterState,
+                              ClusterStateObserver observer, Listener listener) {
         boolean indexingComplete = LifecycleSettings.LIFECYCLE_INDEXING_COMPLETE_SETTING.get(indexMetaData.getSettings());
         if (indexingComplete) {
             logger.trace(indexMetaData.getIndex() + " has lifecycle complete set, skipping " + RolloverStep.NAME);

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/SetSingleNodeAllocateStep.java

@@ -9,6 +9,7 @@
 import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNode;
@@ -42,7 +43,7 @@ public SetSingleNodeAllocateStep(StepKey key, StepKey nextStepKey, Client client
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState clusterState, Listener listener) {
+    public void performAction(IndexMetaData indexMetaData, ClusterState clusterState, ClusterStateObserver observer, Listener listener) {
         RoutingAllocation allocation = new RoutingAllocation(ALLOCATION_DECIDERS, clusterState.getRoutingNodes(), clusterState, null,
                 System.nanoTime());
         List<String> validNodeIds = new ArrayList<>();

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/ShrinkSetAliasStep.java

@@ -18,7 +18,7 @@
  * Following shrinking an index and deleting the original index, this step creates an alias with the same name as the original index which
  * points to the new shrunken index to allow clients to continue to use the original index name without being aware that it has shrunk.
  */
-public class ShrinkSetAliasStep extends AsyncActionStep {
+public class ShrinkSetAliasStep extends AsyncRetryDuringSnapshotActionStep {
     public static final String NAME = "aliases";
     private String shrunkIndexPrefix;
 
@@ -32,7 +32,7 @@ String getShrunkIndexPrefix() {
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
+    public void performDuringNoSnapshot(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
         // get source index
         String index = indexMetaData.getIndex().getName();
         // get target shrink index

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/ShrinkStep.java

@@ -9,6 +9,7 @@
 import org.elasticsearch.action.admin.indices.shrink.ResizeRequest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.settings.Settings;
 
@@ -38,7 +39,7 @@ String getShrunkIndexPrefix() {
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, ClusterStateObserver observer, Listener listener) {
         LifecycleExecutionState lifecycleState = LifecycleExecutionState.fromIndexMetadata(indexMetaData);
         if (lifecycleState.getLifecycleDate() == null) {
             throw new IllegalStateException("source index [" + indexMetaData.getIndex().getName() +

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/indexlifecycle/UpdateSettingsStep.java

@@ -9,6 +9,7 @@
 import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateObserver;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.common.settings.Settings;
 
@@ -28,7 +29,7 @@ public UpdateSettingsStep(StepKey key, StepKey nextStepKey, Client client, Setti
     }
 
     @Override
-    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, Listener listener) {
+    public void performAction(IndexMetaData indexMetaData, ClusterState currentState, ClusterStateObserver observer, Listener listener) {
         UpdateSettingsRequest updateSettingsRequest = new UpdateSettingsRequest(indexMetaData.getIndex().getName()).settings(settings);
         getClient().admin().indices().updateSettings(updateSettingsRequest,
                 ActionListener.wrap(response -> listener.onResponse(true), listener::onFailure));