Don't run async actions when ILM is stopped

docs/changelog/133683.yaml

@@ -0,0 +1,8 @@
+pr: 133683
+summary: Avoid running asynchronous ILM actions while ILM is stopped
+area: ILM+SLM
+type: bug
+issues:
+ - 99859
+ - 81234
+ - 85097

x-pack/plugin/ilm/qa/multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ilm/TimeseriesMoveToStepIT.java

@@ -10,6 +10,7 @@
 import org.apache.http.util.EntityUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.admin.indices.rollover.RolloverConditions;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
@@ -19,18 +20,22 @@
 import org.elasticsearch.xpack.core.ilm.DeleteAction;
 import org.elasticsearch.xpack.core.ilm.ForceMergeAction;
 import org.elasticsearch.xpack.core.ilm.LifecycleSettings;
+import org.elasticsearch.xpack.core.ilm.Phase;
 import org.elasticsearch.xpack.core.ilm.PhaseCompleteStep;
+import org.elasticsearch.xpack.core.ilm.ReadOnlyAction;
 import org.elasticsearch.xpack.core.ilm.RolloverAction;
 import org.elasticsearch.xpack.core.ilm.ShrinkAction;
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 import org.junit.Before;
 
 import java.util.Locale;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import static org.elasticsearch.xpack.TimeSeriesRestDriver.createFullPolicy;
 import static org.elasticsearch.xpack.TimeSeriesRestDriver.createIndexWithSettings;
 import static org.elasticsearch.xpack.TimeSeriesRestDriver.createNewSingletonPolicy;
+import static org.elasticsearch.xpack.TimeSeriesRestDriver.createPolicy;
 import static org.elasticsearch.xpack.TimeSeriesRestDriver.getStepKeyForIndex;
 import static org.elasticsearch.xpack.TimeSeriesRestDriver.index;
 import static org.elasticsearch.xpack.TimeSeriesRestDriver.indexDocument;
@@ -51,6 +56,7 @@ public void refreshIndex() {
         index = "index-" + randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
         policy = "policy-" + randomAlphaOfLength(5);
         alias = "alias-" + randomAlphaOfLength(5);
+        logger.info("--> running [{}] with index [{}], alias [{}] and policy [{}]", getTestName(), index, alias, policy);
     }
 
     public void testMoveToAllocateStep() throws Exception {
@@ -245,6 +251,66 @@ public void testMoveToStepRereadsPolicy() throws Exception {
         assertBusy(() -> { indexExists("test-000002"); });
     }
 
+    /**
+     * Test that an async action does not execute when the Move To Step API is used while ILM is stopped.
+     * Unfortunately, this test doesn't prove that the async action never executes, as it's hard to prove that an asynchronous process
+     * never happens - waiting for a certain period would only increase our confidence but not actually prove it, and it would increase the
+     * runtime of the test significantly. We also assert that the remainder of the policy executes after ILM is started again to ensure that
+     * the index is not stuck in the async action step.
+     */
+    public void testAsyncActionDoesNotExecuteAfterILMStop() throws Exception {
+        String originalIndex = index + "-000001";
+        // Create a simply policy with the most important aspect being the readonly action, which contains the ReadOnlyStep AsyncActionStep.
+        var actions = Map.of(
+            "rollover",
+            new RolloverAction(RolloverConditions.newBuilder().addMaxIndexAgeCondition(TimeValue.timeValueHours(1)).build()),
+            "readonly",
+            new ReadOnlyAction()
+        );
+        Phase phase = new Phase("hot", TimeValue.ZERO, actions);
+        createPolicy(client(), policy, phase, null, null, null, null);
+
+        createIndexWithSettings(
+            client(),
+            originalIndex,
+            alias,
+            Settings.builder().put(LifecycleSettings.LIFECYCLE_NAME, policy).put(RolloverAction.LIFECYCLE_ROLLOVER_ALIAS, alias)
+        );
+
+        // Wait for ILM to do everything it can for this index
+        assertBusy(() -> assertEquals(new StepKey("hot", "rollover", "check-rollover-ready"), getStepKeyForIndex(client(), originalIndex)));
+
+        // Stop ILM
+        client().performRequest(new Request("POST", "/_ilm/stop"));
+
+        // Move ILM to the readonly step, which is an async action step.
+        Request moveToStepRequest = new Request("POST", "_ilm/move/" + originalIndex);
+        moveToStepRequest.setJsonEntity("""
+            {
+              "current_step": {
+                "phase": "hot",
+                "action": "rollover",
+                "name": "check-rollover-ready"
+              },
+              "next_step": {
+                "phase": "hot",
+                "action": "readonly",
+                "name": "readonly"
+              }
+            }""");
+        client().performRequest(moveToStepRequest);
+
+        // Since ILM is stopped, the async action should not execute and the index should remain in the readonly step.
+        // This is the tricky part of the test, as we can't really verify that the async action will never happen.
+        assertEquals(new StepKey("hot", "readonly", "readonly"), getStepKeyForIndex(client(), originalIndex));
+
+        // Restart ILM
+        client().performRequest(new Request("POST", "/_ilm/start"));
+
+        // Make sure we actually complete the remainder of the policy after ILM is started again.
+        assertBusy(() -> assertEquals(new StepKey("hot", "complete", "complete"), getStepKeyForIndex(client(), originalIndex)));
+    }
+
     public void testMoveToStepWithInvalidNextStep() throws Exception {
         createNewSingletonPolicy(client(), policy, "delete", DeleteAction.WITH_SNAPSHOT_DELETE, TimeValue.timeValueDays(100));
         createIndexWithSettings(

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

@@ -33,6 +33,7 @@
 import org.elasticsearch.xpack.core.ilm.ClusterStateWaitStep;
 import org.elasticsearch.xpack.core.ilm.ErrorStep;
 import org.elasticsearch.xpack.core.ilm.LifecycleSettings;
+import org.elasticsearch.xpack.core.ilm.OperationMode;
 import org.elasticsearch.xpack.core.ilm.PhaseCompleteStep;
 import org.elasticsearch.xpack.core.ilm.Step;
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
@@ -48,6 +49,7 @@
 
 import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.index.IndexSettings.LIFECYCLE_ORIGINATION_DATE;
+import static org.elasticsearch.xpack.core.ilm.LifecycleOperationMetadata.currentILMMode;
 
 class IndexLifecycleRunner {
     private static final Logger logger = LogManager.getLogger(IndexLifecycleRunner.class);
@@ -308,6 +310,12 @@ void onErrorMaybeRetryFailedStep(ProjectId projectId, String policy, StepKey cur
     void maybeRunAsyncAction(ProjectState state, IndexMetadata indexMetadata, String policy, StepKey expectedStepKey) {
         final var projectId = state.projectId();
         String index = indexMetadata.getIndex().getName();
+        OperationMode currentMode = currentILMMode(state.metadata());
+        if (OperationMode.RUNNING.equals(currentMode) == false) {
+            logger.info("[{}] not running async action in policy [{}] because ILM is [{}]", index, policy, currentMode);
+            return;
+        }
+
         if (LifecycleSettings.LIFECYCLE_SKIP_SETTING.get(indexMetadata.getSettings())) {
             logger.info("[{}] skipping policy [{}] because [{}] is true", index, policy, LifecycleSettings.LIFECYCLE_SKIP);
             return;

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

@@ -184,88 +184,78 @@ void onMaster(ClusterState clusterState) {
         maybeScheduleJob();
 
         for (var projectId : clusterState.metadata().projects().keySet()) {
-            onMaster(clusterState.projectState(projectId));
+            maybeRunAsyncActions(clusterState.projectState(projectId));
         }
     }
 
-    void onMaster(ProjectState state) {
+    /**
+     * Kicks off any async actions that may not have been run due to either master failover or ILM being manually stopped.
+     */
+    private void maybeRunAsyncActions(ProjectState state) {
         final ProjectMetadata projectMetadata = state.metadata();
         final IndexLifecycleMetadata currentMetadata = projectMetadata.custom(IndexLifecycleMetadata.TYPE);
-        if (currentMetadata != null) {
-            OperationMode currentMode = currentILMMode(projectMetadata);
-            if (OperationMode.STOPPED.equals(currentMode)) {
-                return;
-            }
-
-            boolean safeToStop = true; // true until proven false by a run policy
-
-            // If we just became master, we need to kick off any async actions that
-            // may have not been run due to master rollover
-            for (IndexMetadata idxMeta : projectMetadata.indices().values()) {
-                if (projectMetadata.isIndexManagedByILM(idxMeta)) {
-                    String policyName = idxMeta.getLifecyclePolicyName();
-                    final LifecycleExecutionState lifecycleState = idxMeta.getLifecycleExecutionState();
-                    StepKey stepKey = Step.getCurrentStepKey(lifecycleState);
-
-                    try {
-                        if (OperationMode.STOPPING == currentMode) {
-                            if (stepKey != null && IGNORE_STEPS_MAINTENANCE_REQUESTED.contains(stepKey.name())) {
-                                logger.info(
-                                    "waiting to stop ILM because index [{}] with policy [{}] is currently in step [{}]",
-                                    idxMeta.getIndex().getName(),
-                                    policyName,
-                                    stepKey.name()
-                                );
-                                lifecycleRunner.maybeRunAsyncAction(state, idxMeta, policyName, stepKey);
-                                // ILM is trying to stop, but this index is in a Shrink step (or other dangerous step) so we can't stop
-                                safeToStop = false;
-                            } else {
-                                logger.info(
-                                    "skipping policy execution of step [{}] for index [{}] with policy [{}]" + " because ILM is stopping",
-                                    stepKey == null ? "n/a" : stepKey.name(),
-                                    idxMeta.getIndex().getName(),
-                                    policyName
-                                );
-                            }
-                        } else {
-                            lifecycleRunner.maybeRunAsyncAction(state, idxMeta, policyName, stepKey);
-                        }
-                    } catch (Exception e) {
-                        if (logger.isTraceEnabled()) {
-                            logger.warn(
-                                () -> format(
-                                    "async action execution failed during master election trigger"
-                                        + " for index [%s] with policy [%s] in step [%s], lifecycle state: [%s]",
-                                    idxMeta.getIndex().getName(),
-                                    policyName,
-                                    stepKey,
-                                    lifecycleState.asMap()
-                                ),
-                                e
-                            );
-                        } else {
-                            logger.warn(
-                                () -> format(
-                                    "async action execution failed during master election trigger"
-                                        + " for index [%s] with policy [%s] in step [%s]",
-                                    idxMeta.getIndex().getName(),
-                                    policyName,
-                                    stepKey
-                                ),
-                                e
-                            );
+        if (currentMetadata == null) {
+            return;
+        }
+        OperationMode currentMode = currentILMMode(projectMetadata);
+        if (OperationMode.STOPPED.equals(currentMode)) {
+            return;
+        }
 
-                        }
-                        // Don't rethrow the exception, we don't want a failure for one index to be
-                        // called to cause actions not to be triggered for further indices
-                    }
-                }
+        boolean safeToStop = true; // true until proven false by a run policy
+        for (IndexMetadata idxMeta : projectMetadata.indices().values()) {
+            if (projectMetadata.isIndexManagedByILM(idxMeta) == false) {
+                continue;
             }
+            String policyName = idxMeta.getLifecyclePolicyName();
+            final LifecycleExecutionState lifecycleState = idxMeta.getLifecycleExecutionState();
+            StepKey stepKey = Step.getCurrentStepKey(lifecycleState);
+
+            try {
+                if (currentMode == OperationMode.RUNNING) {
+                    lifecycleRunner.maybeRunAsyncAction(state, idxMeta, policyName, stepKey);
+                    continue;
+                }
+                // We only get here if ILM is in STOPPING mode. In that case, we need to check if there is any index that is in a step
+                // that we can't stop ILM in. If there is, we don't stop ILM yet.
+                if (stepKey != null && IGNORE_STEPS_MAINTENANCE_REQUESTED.contains(stepKey.name())) {
+                    logger.info(
+                        "waiting to stop ILM because index [{}] with policy [{}] is currently in step [{}]",
+                        idxMeta.getIndex().getName(),
+                        policyName,
+                        stepKey.name()
+                    );
+                    lifecycleRunner.maybeRunAsyncAction(state, idxMeta, policyName, stepKey);
+                    // ILM is trying to stop, but this index is in a Shrink step (or other dangerous step) so we can't stop
+                    safeToStop = false;
+                } else {
+                    logger.info(
+                        "skipping policy execution of step [{}] for index [{}] with policy [{}]" + " because ILM is stopping",
+                        stepKey == null ? "n/a" : stepKey.name(),
+                        idxMeta.getIndex().getName(),
+                        policyName
+                    );
+                }
+            } catch (Exception e) {
+                String logMessage = format(
+                    "async action execution failed during master election trigger for index [%s] with policy [%s] in step [%s]",
+                    idxMeta.getIndex().getName(),
+                    policyName,
+                    stepKey
+                );
+                if (logger.isTraceEnabled()) {
+                    logMessage += format(", lifecycle state: [%s]", lifecycleState.asMap());
+                }
+                logger.warn(logMessage, e);
 
-            if (safeToStop && OperationMode.STOPPING == currentMode) {
-                stopILM(state.projectId());
+                // Don't rethrow the exception, we don't want a failure for one index to be
+                // called to cause actions not to be triggered for further indices
             }
         }
+
+        if (safeToStop && OperationMode.STOPPING == currentMode) {
+            stopILM(state.projectId());
+        }
     }
 
     private void stopILM(ProjectId projectId) {
@@ -333,6 +323,20 @@ public void clusterChanged(ClusterChangedEvent event) {
                 cancelJob();
                 policyRegistry.clear();
             }
+        } else if (this.isMaster) {
+            // If we are the master and we were before, check if any projects changed their ILM mode from non-RUNNING to RUNNING.
+            // If so, kick off any async actions that may not have run while not in RUNNING mode.
+            for (ProjectMetadata project : event.state().metadata().projects().values()) {
+                final var previousProject = event.previousState().metadata().projects().get(project.id());
+                if (previousProject == null || project == previousProject) {
+                    continue;
+                }
+                final OperationMode currentMode = currentILMMode(project);
+                final OperationMode previousMode = currentILMMode(previousProject);
+                if (currentMode == OperationMode.RUNNING && previousMode != OperationMode.RUNNING) {
+                    maybeRunAsyncActions(event.state().projectState(project.id()));
+                }
+            }
         }
 
         // if we're the master, then process deleted indices and trigger policies

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

@@ -265,6 +265,34 @@ public void testSkip_asyncAction() {
         Mockito.verifyNoMoreInteractions(clusterService);
     }
 
+    /**
+     * Test that an async action step is not executed when ILM is stopped.
+     */
+    public void testNotRunningAsyncActionWhenILMIsStopped() {
+        String policyName = "stopped_policy";
+        Step.StepKey stepKey = new Step.StepKey("phase", "action", "async_action_step");
+
+        MockAsyncActionStep step = new MockAsyncActionStep(stepKey, null);
+
+        PolicyStepsRegistry stepRegistry = createOneStepPolicyStepRegistry(policyName, step);
+        ClusterService clusterService = mock(ClusterService.class);
+        newMockTaskQueue(clusterService); // ensure constructor call to createTaskQueue is satisfied
+        IndexLifecycleRunner runner = new IndexLifecycleRunner(stepRegistry, historyStore, clusterService, threadPool, () -> 0L);
+
+        IndexMetadata indexMetadata = IndexMetadata.builder("test")
+            .settings(randomIndexSettings().put(LifecycleSettings.LIFECYCLE_NAME, policyName))
+            .build();
+
+        IndexLifecycleMetadata ilm = new IndexLifecycleMetadata(Map.of(), OperationMode.STOPPED);
+        final var project = ProjectMetadata.builder(randomProjectIdOrDefault())
+            .put(indexMetadata, true)
+            .putCustom(IndexLifecycleMetadata.TYPE, ilm)
+            .build();
+        runner.maybeRunAsyncAction(projectStateFromProject(project), indexMetadata, policyName, stepKey);
+
+        assertThat(step.getExecuteCount(), equalTo(0L));
+    }
+
     public void testRunPolicyErrorStepOnRetryableFailedStep() {
         String policyName = "rollover_policy";
         String phaseName = "hot";
@@ -586,7 +614,6 @@ public void testRunStateChangePolicyWithAsyncActionNextStep() throws Exception {
             .putProjectMetadata(project)
             .nodes(DiscoveryNodes.builder().add(node).masterNodeId(node.getId()).localNodeId(node.getId()))
             .build();
-        logger.info("--> state: {}", state);
         ClusterServiceUtils.setState(clusterService, state);
         IndexLifecycleRunner runner = new IndexLifecycleRunner(stepRegistry, historyStore, clusterService, threadPool, () -> 0L);