ILM: Add support for the searchable_snapshot action in the hot phase (#64883)

This adds support for the searchable_snapshot ILM action in the hot phase.

We define a series of actions that cannot be executed after the index has been
mounted as a searchable snapshot. Namely: freeze, forcemerge, shrink,
and searchable_snapshot (also available in the cold phase).

If by virtue of snapshot/restoring a managed index or updating an ILM policy while it
is executing for an index, these actions could get to be executed on an index that was
mounted as searchable snapshot in the hot phase. If this happens the actions will
skip entirely. ILM will not move into the ERROR step.

Author: andreidan

docs/reference/ilm/actions/ilm-searchable-snapshot.asciidoc

@@ -4,13 +4,21 @@
 
 beta::[]
 
-Phases allowed: cold.
+Phases allowed: hot, cold.
 
 Takes a snapshot of the managed index in the configured repository
 and mounts it as a searchable snapshot.
 If the managed index is part of a <<data-streams, data stream>>,
 the mounted index replaces the original index in the data stream.
 
+To use the `searchable_snapshot` action in the `hot` phase, the `rollover`
+action *must* be present. If no rollover action is configured, {ilm-init}
+will reject the policy.
+
+IMPORTANT: If the `searchable_snapshot` action is used in the `hot` phase the
+subsequent phases cannot define any of the `shrink`, `forcemerge`, `freeze` or
+`searchable_snapshot` (also available in the cold phase) actions.
+
 [NOTE]
 This action cannot be performed on a data stream's write index. Attempts to do
 so will fail. To convert the index to a searchable snapshot, first

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

@@ -26,7 +26,7 @@
  * 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);
+    private static final Logger logger = LogManager.getLogger(AsyncRetryDuringSnapshotActionStep.class);
 
     public AsyncRetryDuringSnapshotActionStep(StepKey key, StepKey nextStepKey, Client client) {
         super(key, nextStepKey, client);

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

@@ -5,6 +5,8 @@
  */
 package org.elasticsearch.xpack.core.ilm;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Version;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
@@ -31,9 +33,12 @@
  * A {@link LifecycleAction} which force-merges the index.
  */
 public class ForceMergeAction implements LifecycleAction {
+    private static final Logger logger = LogManager.getLogger(ForceMergeAction.class);
+
     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 CONDITIONAL_SKIP_FORCE_MERGE_STEP = BranchingStep.NAME + "-forcemerge-check-prerequisites";
 
     private static final ConstructingObjectParser<ForceMergeAction, Void> PARSER = new ConstructingObjectParser<>(NAME,
         false, a -> {
@@ -120,6 +125,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
 
         final boolean codecChange = codec != null && codec.equals(CodecService.BEST_COMPRESSION_CODEC);
 
+        StepKey preForceMergeBranchingKey = new StepKey(phase, NAME, CONDITIONAL_SKIP_FORCE_MERGE_STEP);
         StepKey checkNotWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
         StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyAction.NAME);
 
@@ -131,6 +137,18 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
         StepKey forceMergeKey = new StepKey(phase, NAME, ForceMergeStep.NAME);
         StepKey countKey = new StepKey(phase, NAME, SegmentCountStep.NAME);
 
+        BranchingStep conditionalSkipShrinkStep = new BranchingStep(preForceMergeBranchingKey, checkNotWriteIndex, nextStepKey,
+            (index, clusterState) -> {
+                IndexMetadata indexMetadata = clusterState.metadata().index(index);
+                assert indexMetadata != null : "index " + index.getName() + " must exist in the cluster state";
+                if (indexMetadata.getSettings().get(LifecycleSettings.SNAPSHOT_INDEX_NAME) != null) {
+                    String policyName = LifecycleSettings.LIFECYCLE_NAME_SETTING.get(indexMetadata.getSettings());
+                    logger.warn("[{}] action is configured for index [{}] in policy [{}] which is mounted as searchable snapshot. " +
+                        "Skipping this action", ForceMergeAction.NAME, index.getName(), policyName);
+                    return true;
+                }
+                return false;
+            });
         CheckNotDataStreamWriteIndexStep checkNotWriteIndexStep = new CheckNotDataStreamWriteIndexStep(checkNotWriteIndex,
             readOnlyKey);
         UpdateSettingsStep readOnlyStep =
@@ -147,6 +165,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
         SegmentCountStep segmentCountStep = new SegmentCountStep(countKey, nextStepKey, client, maxNumSegments);
 
         List<Step> mergeSteps = new ArrayList<>();
+        mergeSteps.add(conditionalSkipShrinkStep);
         mergeSteps.add(checkNotWriteIndexStep);
         mergeSteps.add(readOnlyStep);
 

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

@@ -5,7 +5,10 @@
  */
 package org.elasticsearch.xpack.core.ilm;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -22,7 +25,10 @@
  * A {@link LifecycleAction} which freezes the index.
  */
 public class FreezeAction implements LifecycleAction {
+    private static final Logger logger = LogManager.getLogger(FreezeAction.class);
+
     public static final String NAME = "freeze";
+    public static final String CONDITIONAL_SKIP_FREEZE_STEP = BranchingStep.NAME + "-freeze-check-prerequisites";
 
     private static final ObjectParser<FreezeAction, Void> PARSER = new ObjectParser<>(NAME, FreezeAction::new);
 
@@ -59,13 +65,31 @@ public boolean isSafeAction() {
 
     @Override
     public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
+        StepKey preFreezeMergeBranchingKey = new StepKey(phase, NAME, CONDITIONAL_SKIP_FREEZE_STEP);
         StepKey checkNotWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
         StepKey freezeStepKey = new StepKey(phase, NAME, FreezeStep.NAME);
 
+        BranchingStep conditionalSkipFreezeStep = new BranchingStep(preFreezeMergeBranchingKey, checkNotWriteIndex, nextStepKey,
+            (index, clusterState) -> {
+                IndexMetadata indexMetadata = clusterState.getMetadata().index(index);
+                assert indexMetadata != null : "index " + index.getName() + " must exist in the cluster state";
+                String policyName = LifecycleSettings.LIFECYCLE_NAME_SETTING.get(indexMetadata.getSettings());
+                if (indexMetadata.getSettings().get(LifecycleSettings.SNAPSHOT_INDEX_NAME) != null) {
+                    logger.warn("[{}] action is configured for index [{}] in policy [{}] which is mounted as searchable snapshot. " +
+                        "Skipping this action", FreezeAction.NAME, index.getName(), policyName);
+                    return true;
+                }
+                if (indexMetadata.getSettings().getAsBoolean("index.frozen", false)) {
+                    logger.debug("skipping [{}] action for index [{}] in policy [{}] as the index is already frozen", FreezeAction.NAME,
+                        index.getName(), policyName);
+                    return true;
+                }
+                return false;
+            });
         CheckNotDataStreamWriteIndexStep checkNoWriteIndexStep = new CheckNotDataStreamWriteIndexStep(checkNotWriteIndex,
             freezeStepKey);
         FreezeStep freezeStep = new FreezeStep(freezeStepKey, nextStepKey, client);
-        return Arrays.asList(checkNoWriteIndexStep, freezeStep);
+        return Arrays.asList(conditionalSkipFreezeStep, checkNoWriteIndexStep, freezeStep);
     }
 
     @Override

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

@@ -27,6 +27,10 @@ public class LifecycleSettings {
     public static final String SLM_RETENTION_DURATION = "slm.retention_duration";
     public static final String SLM_MINIMUM_INTERVAL = "slm.minimum_interval";
 
+    // This is not a setting configuring ILM per se, but certain ILM actions need to validate the managed index is not
+    // already mounted as a searchable snapshot. Those ILM actions will check if the index has this setting name configured.
+    public static final String SNAPSHOT_INDEX_NAME = "index.store.snapshot.index_name";
+
     public static final Setting<TimeValue> LIFECYCLE_POLL_INTERVAL_SETTING = Setting.timeSetting(LIFECYCLE_POLL_INTERVAL,
         TimeValue.timeValueMinutes(10), TimeValue.timeValueSeconds(1), Setting.Property.Dynamic, Setting.Property.NodeScope);
     public static final Setting<String> LIFECYCLE_NAME_SETTING = Setting.simpleString(LIFECYCLE_NAME,

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

@@ -5,10 +5,13 @@
  */
 package org.elasticsearch.xpack.core.ilm;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Version;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -29,11 +32,14 @@
  * newly created searchable snapshot backed index.
  */
 public class SearchableSnapshotAction implements LifecycleAction {
+    private static final Logger logger = LogManager.getLogger(SearchableSnapshotAction.class);
+
     public static final String NAME = "searchable_snapshot";
 
     public static final ParseField SNAPSHOT_REPOSITORY = new ParseField("snapshot_repository");
     public static final ParseField FORCE_MERGE_INDEX = new ParseField("force_merge_index");
     public static final String CONDITIONAL_DATASTREAM_CHECK_KEY = BranchingStep.NAME + "-on-datastream-check";
+    public static final String CONDITIONAL_SKIP_ACTION_STEP = BranchingStep.NAME + "-check-prerequisites";
 
     public static final String RESTORED_INDEX_PREFIX = "restored-";
 
@@ -74,6 +80,7 @@ boolean isForceMergeIndex() {
 
     @Override
     public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
+        StepKey preActionBranchingKey = new StepKey(phase, NAME, CONDITIONAL_SKIP_ACTION_STEP);
         StepKey checkNoWriteIndex = new StepKey(phase, NAME, CheckNotDataStreamWriteIndexStep.NAME);
         StepKey waitForNoFollowerStepKey = new StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
         StepKey forceMergeStepKey = new StepKey(phase, NAME, ForceMergeStep.NAME);
@@ -90,6 +97,18 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
         StepKey replaceDataStreamIndexKey = new StepKey(phase, NAME, ReplaceDataStreamBackingIndexStep.NAME);
         StepKey deleteIndexKey = new StepKey(phase, NAME, DeleteStep.NAME);
 
+        BranchingStep conditionalSkipActionStep = new BranchingStep(preActionBranchingKey, checkNoWriteIndex, nextStepKey,
+            (index, clusterState) -> {
+                IndexMetadata indexMetadata = clusterState.getMetadata().index(index);
+                assert indexMetadata != null : "index " + index.getName() + " must exist in the cluster state";
+                if (indexMetadata.getSettings().get(LifecycleSettings.SNAPSHOT_INDEX_NAME) != null) {
+                    logger.warn("[{}] action is configured for index [{}] in policy [{}] which is already mounted as searchable " +
+                            "snapshot. Skipping this action", SearchableSnapshotAction.NAME, index.getName(),
+                        LifecycleSettings.LIFECYCLE_NAME_SETTING.get(indexMetadata.getSettings()));
+                    return true;
+                }
+                return false;
+            });
         CheckNotDataStreamWriteIndexStep checkNoWriteIndexStep = new CheckNotDataStreamWriteIndexStep(checkNoWriteIndex,
             waitForNoFollowerStepKey);
         final WaitForNoFollowersStep waitForNoFollowersStep;
@@ -130,6 +149,7 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
             null, client, RESTORED_INDEX_PREFIX);
 
         List<Step> steps = new ArrayList<>();
+        steps.add(conditionalSkipActionStep);
         steps.add(checkNoWriteIndexStep);
         steps.add(waitForNoFollowersStep);
         if (forceMergeIndex) {

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

@@ -29,14 +29,14 @@
  * A {@link LifecycleAction} which shrinks the index.
  */
 public class ShrinkAction implements LifecycleAction {
+    private static final Logger logger = LogManager.getLogger(ShrinkAction.class);
+
     public static final String NAME = "shrink";
     public static final String SHRUNKEN_INDEX_PREFIX = "shrink-";
     public static final ParseField NUMBER_OF_SHARDS_FIELD = new ParseField("number_of_shards");
     public static final String CONDITIONAL_SKIP_SHRINK_STEP = BranchingStep.NAME + "-check-prerequisites";
     public static final String CONDITIONAL_DATASTREAM_CHECK_KEY = BranchingStep.NAME + "-on-datastream-check";
 
-    private static final Logger logger = LogManager.getLogger(ShrinkAction.class);
-
     private static final ConstructingObjectParser<ShrinkAction, Void> PARSER =
         new ConstructingObjectParser<>(NAME, a -> new ShrinkAction((Integer) a[0]));
 
@@ -108,7 +108,19 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)
         StepKey deleteIndexKey = new StepKey(phase, NAME, DeleteStep.NAME);
 
         BranchingStep conditionalSkipShrinkStep = new BranchingStep(preShrinkBranchingKey, checkNotWriteIndex, nextStepKey,
-            (index, clusterState) -> clusterState.getMetadata().index(index).getNumberOfShards() == numberOfShards);
+            (index, clusterState) -> {
+                IndexMetadata indexMetadata = clusterState.getMetadata().index(index);
+                if (indexMetadata.getNumberOfShards() == numberOfShards) {
+                    return true;
+                }
+                if (indexMetadata.getSettings().get(LifecycleSettings.SNAPSHOT_INDEX_NAME) != null) {
+                    logger.warn("[{}] action is configured for index [{}] in policy [{}] which is mounted as searchable snapshot. " +
+                            "Skipping this action", ShrinkAction.NAME, indexMetadata.getIndex().getName(),
+                        LifecycleSettings.LIFECYCLE_NAME_SETTING.get(indexMetadata.getSettings()));
+                    return true;
+                }
+                return false;
+            });
         CheckNotDataStreamWriteIndexStep checkNotWriteIndexStep = new CheckNotDataStreamWriteIndexStep(checkNotWriteIndex,
             waitForNoFollowerStepKey);
         WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(waitForNoFollowerStepKey, readOnlyKey, client);

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

@@ -13,6 +13,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -40,7 +41,7 @@ public class TimeseriesLifecycleType implements LifecycleType {
     static final String DELETE_PHASE = "delete";
     static final List<String> VALID_PHASES = Arrays.asList(HOT_PHASE, WARM_PHASE, COLD_PHASE, DELETE_PHASE);
     static final List<String> ORDERED_VALID_HOT_ACTIONS = Arrays.asList(SetPriorityAction.NAME, UnfollowAction.NAME, RolloverAction.NAME,
-        ReadOnlyAction.NAME, ShrinkAction.NAME, ForceMergeAction.NAME);
+        ReadOnlyAction.NAME, ShrinkAction.NAME, ForceMergeAction.NAME, SearchableSnapshotAction.NAME);
     static final List<String> ORDERED_VALID_WARM_ACTIONS = Arrays.asList(SetPriorityAction.NAME, UnfollowAction.NAME, ReadOnlyAction.NAME,
         AllocateAction.NAME, MigrateAction.NAME, ShrinkAction.NAME, ForceMergeAction.NAME);
     static final List<String> ORDERED_VALID_COLD_ACTIONS = Arrays.asList(SetPriorityAction.NAME, UnfollowAction.NAME, AllocateAction.NAME,
@@ -57,7 +58,10 @@ public class TimeseriesLifecycleType implements LifecycleType {
         DELETE_PHASE, VALID_DELETE_ACTIONS);
 
     static final Set<String> HOT_ACTIONS_THAT_REQUIRE_ROLLOVER = Sets.newHashSet(ReadOnlyAction.NAME, ShrinkAction.NAME,
-        ForceMergeAction.NAME);
+        ForceMergeAction.NAME, SearchableSnapshotAction.NAME);
+    // a set of actions that cannot be defined (executed) after the managed index has been mounted as searchable snapshot
+    static final Set<String> ACTIONS_CANNOT_FOLLOW_SEARCHABLE_SNAPSHOT = Sets.newHashSet(ShrinkAction.NAME, ForceMergeAction.NAME,
+        FreezeAction.NAME, SearchableSnapshotAction.NAME);
 
     private TimeseriesLifecycleType() {
     }
@@ -255,6 +259,29 @@ public void validate(Collection<Phase> phases) {
                 MigrateAction.NAME + " action and an " + AllocateAction.NAME + " action with allocation rules. specify only a single " +
                 "data migration in each phase");
         }
+
+        validateActionsFollowingSearchableSnapshot(phases);
+    }
+
+    static void validateActionsFollowingSearchableSnapshot(Collection<Phase> phases) {
+        boolean hotPhaseContainsSearchableSnapshot = phases.stream()
+            .filter(phase -> HOT_PHASE.equals(phase.getName()))
+            .anyMatch(phase -> phase.getActions().containsKey(SearchableSnapshotAction.NAME));
+        if (hotPhaseContainsSearchableSnapshot) {
+            String phasesDefiningIllegalActions = phases.stream()
+                // we're looking for prohibited actions in phases other than hot
+                .filter(phase -> HOT_PHASE.equals(phase.getName()) == false)
+                // filter the phases that define illegal actions
+                .filter(phase ->
+                    Collections.disjoint(ACTIONS_CANNOT_FOLLOW_SEARCHABLE_SNAPSHOT, phase.getActions().keySet()) == false)
+                .map(Phase::getName)
+                .collect(Collectors.joining(","));
+            if (Strings.hasText(phasesDefiningIllegalActions)) {
+                throw new IllegalArgumentException("phases [" + phasesDefiningIllegalActions + "] define one or more of " +
+                    ACTIONS_CANNOT_FOLLOW_SEARCHABLE_SNAPSHOT + " actions which are not allowed after a " +
+                    "managed index is mounted as a searchable snapshot");
+            }
+        }
     }
 
     private static boolean definesAllocationRules(AllocateAction action) {