Autoscale frozen tier into existence (#73435)

This commit adds two related changes:
* ILM WaitForDataTierStep
* Autoscaling frozen_existence decider

The first part ensures that we wait mounting an index until a node that
can hold the index is available, avoiding a failed restore and red
cluster state. This is in particular important for the frozen phase, but
is done generically in the searchable snapshot action.

The second part triggers on indices in the ILM frozen phase to scale the
tier into existence by requiring a minimal amount of memory and storage.

Closes #72771

docs/reference/autoscaling/autoscaling-deciders.asciidoc

@@ -20,6 +20,10 @@ Estimates required storage capacity as a percentage of the total data set of
 partially mounted indices.
 Available for policies governing frozen data nodes.
 
+<<autoscaling-frozen-existence-decider,Frozen existence decider>>::
+Estimates a minimum require frozen memory and storage capacity when any index is
+in the frozen <<index-lifecycle-management,ILM>> phase.
+
 <<autoscaling-machine-learning-decider,Machine learning decider>>::
 Estimates required memory capacity based on machine learning jobs.
 Available for policies governing machine learning nodes.
@@ -31,5 +35,6 @@ include::deciders/reactive-storage-decider.asciidoc[]
 include::deciders/proactive-storage-decider.asciidoc[]
 include::deciders/frozen-shards-decider.asciidoc[]
 include::deciders/frozen-storage-decider.asciidoc[]
+include::deciders/frozen-existence-decider.asciidoc[]
 include::deciders/machine-learning-decider.asciidoc[]
 include::deciders/fixed-decider.asciidoc[]

docs/reference/autoscaling/deciders/frozen-existence-decider.asciidoc

@@ -0,0 +1,9 @@
+[role="xpack"]
+[[autoscaling-frozen-existence-decider]]
+=== Frozen existence decider
+
+The frozen existence decider (`frozen_existence`) ensures that once the first
+index enters the frozen ILM phase, the frozen tier is scaled into existence.
+
+The frozen existence decider is enabled for all policies governing frozen data
+nodes and has no configuration options.

x-pack/plugin/autoscaling/build.gradle

@@ -16,6 +16,7 @@ dependencies {
   testImplementation(testArtifact(project(xpackModule('core'))))
   testImplementation project(path: xpackModule('data-streams'))
   testImplementation project(path: xpackModule('searchable-snapshots'))
+  testImplementation project(path: xpackModule('ilm'))
 }
 
 addQaCheckDependencies()

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/AbstractFrozenAutoscalingIntegTestCase.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.autoscaling;
 
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
@@ -25,7 +26,6 @@ import org.elasticsearch.xpack.core.DataTier;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotAction;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
 import org.elasticsearch.xpack.searchablesnapshots.cache.shared.FrozenCacheService;
-import org.junit.Before;
 
 import java.util.Collection;
 import java.util.List;
@@ -66,18 +66,19 @@ public abstract class AbstractFrozenAutoscalingIntegTestCase extends AbstractSna
         Settings.Builder builder = Settings.builder()
             .put(super.nodeSettings(nodeOrdinal, otherSettings))
             .put(SELF_GENERATED_LICENSE_TYPE.getKey(), "trial");
-        if (DiscoveryNode.canContainData(otherSettings)) {
+        if (DiscoveryNode.hasRole(otherSettings, DiscoveryNodeRole.DATA_FROZEN_NODE_ROLE)) {
             builder.put(FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(), new ByteSizeValue(10, ByteSizeUnit.MB));
         }
         return builder.build();
     }
 
-    @Before
-    public void setupPolicyAndMountedIndex() throws Exception {
+    protected void setupRepoAndPolicy() {
         createRepository(fsRepoName, "fs");
         putAutoscalingPolicy();
-        assertAcked(prepareCreate(indexName, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true)));
+    }
 
+    protected void createAndMountIndex() throws InterruptedException, java.util.concurrent.ExecutionException {
+        assertAcked(prepareCreate(indexName, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true)));
         indexRandom(
             randomBoolean(),
             IntStream.range(0, 10).mapToObj(i -> client().prepareIndex(indexName).setSource()).collect(Collectors.toList())

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/existence/FrozenExistenceDeciderIT.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.existence;
+
+import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.snapshots.SnapshotInfo;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.NodeRoles;
+import org.elasticsearch.xpack.autoscaling.AbstractFrozenAutoscalingIntegTestCase;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingCapacity;
+import org.elasticsearch.xpack.core.ilm.ExplainLifecycleRequest;
+import org.elasticsearch.xpack.core.ilm.ExplainLifecycleResponse;
+import org.elasticsearch.xpack.core.ilm.IndexLifecycleExplainResponse;
+import org.elasticsearch.xpack.core.ilm.LifecyclePolicy;
+import org.elasticsearch.xpack.core.ilm.LifecycleSettings;
+import org.elasticsearch.xpack.core.ilm.Phase;
+import org.elasticsearch.xpack.core.ilm.SearchableSnapshotAction;
+import org.elasticsearch.xpack.core.ilm.WaitForDataTierStep;
+import org.elasticsearch.xpack.core.ilm.action.ExplainLifecycleAction;
+import org.elasticsearch.xpack.core.ilm.action.PutLifecycleAction;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Collections.singletonMap;
+import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS;
+import static org.elasticsearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_SHARDS;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.anyOf;
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.not;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class FrozenExistenceDeciderIT extends AbstractFrozenAutoscalingIntegTestCase {
+
+    private static final String INDEX_NAME = "index";
+    private static final String PARTIAL_INDEX_NAME = "partial-index";
+
+    @Override
+    protected String deciderName() {
+        return FrozenExistenceDeciderService.NAME;
+    }
+
+    @Override
+    protected Settings.Builder addDeciderSettings(Settings.Builder builder) {
+        return builder;
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        Settings.Builder settings = Settings.builder().put(super.nodeSettings(nodeOrdinal, otherSettings));
+        settings.put(LifecycleSettings.LIFECYCLE_POLL_INTERVAL, "1s");
+        settings.put(LifecycleSettings.LIFECYCLE_HISTORY_INDEX_ENABLED, false);
+        settings.put(LifecycleSettings.SLM_HISTORY_INDEX_ENABLED_SETTING.getKey(), false);
+        return settings.build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return List.of(LocalStateAutoscalingAndSearchableSnapshotsAndIndexLifecycle.class);
+    }
+
+    public void testZeroToOne() throws Exception {
+        internalCluster().startMasterOnlyNode();
+        setupRepoAndPolicy();
+        logger.info("starting 2 content data nodes");
+        internalCluster().startNode(NodeRoles.onlyRole(DiscoveryNodeRole.DATA_CONTENT_NODE_ROLE));
+        internalCluster().startNode(NodeRoles.onlyRole(DiscoveryNodeRole.DATA_CONTENT_NODE_ROLE));
+        // create an ignored snapshot to initialize the latest-N file.
+        final SnapshotInfo snapshotInfo = createFullSnapshot(fsRepoName, snapshotName);
+
+        Phase hotPhase = new Phase("hot", TimeValue.ZERO, Collections.emptyMap());
+        Phase frozenPhase = new Phase(
+            "frozen",
+            TimeValue.ZERO,
+            singletonMap(SearchableSnapshotAction.NAME, new SearchableSnapshotAction(fsRepoName, randomBoolean()))
+        );
+        LifecyclePolicy lifecyclePolicy = new LifecyclePolicy("policy", Map.of("hot", hotPhase, "frozen", frozenPhase));
+        PutLifecycleAction.Request putLifecycleRequest = new PutLifecycleAction.Request(lifecyclePolicy);
+        assertAcked(client().execute(PutLifecycleAction.INSTANCE, putLifecycleRequest).get());
+
+        Settings settings = Settings.builder()
+            .put(indexSettings())
+            .put(SETTING_NUMBER_OF_SHARDS, 1)
+            .put(SETTING_NUMBER_OF_REPLICAS, 1)
+            .put(LifecycleSettings.LIFECYCLE_NAME, "policy")
+            .build();
+        CreateIndexResponse res = client().admin().indices().prepareCreate(INDEX_NAME).setSettings(settings).get();
+        assertTrue(res.isAcknowledged());
+        logger.info("created index");
+
+        assertBusy(() -> { assertMinimumCapacity(capacity().results().get("frozen").requiredCapacity().total()); });
+        assertMinimumCapacity(capacity().results().get("frozen").requiredCapacity().node());
+
+        assertThat(
+            client().admin().cluster().prepareHealth().get().getStatus(),
+            anyOf(equalTo(ClusterHealthStatus.YELLOW), equalTo(ClusterHealthStatus.GREEN))
+        );
+
+        assertBusy(() -> {
+            ExplainLifecycleResponse response = client().execute(
+                ExplainLifecycleAction.INSTANCE,
+                new ExplainLifecycleRequest().indices(INDEX_NAME)
+            ).actionGet();
+            IndexLifecycleExplainResponse indexResponse = response.getIndexResponses().get(INDEX_NAME);
+            assertNotNull(indexResponse);
+            assertThat(indexResponse.getStep(), equalTo(WaitForDataTierStep.NAME));
+        });
+
+        // verify that SearchableSnapshotAction uses WaitForDataTierStep and that it waits.
+        assertThat(indices(), not(arrayContaining(PARTIAL_INDEX_NAME)));
+
+        logger.info("starting dedicated frozen node");
+        internalCluster().startNode(NodeRoles.onlyRole(DiscoveryNodeRole.DATA_FROZEN_NODE_ROLE));
+
+        assertBusy(() -> {
+            String[] indices = indices();
+            assertThat(indices, arrayContaining(PARTIAL_INDEX_NAME));
+            assertThat(indices, not(arrayContaining(INDEX_NAME)));
+        });
+        ensureGreen();
+    }
+
+    private String[] indices() {
+        return client().admin().indices().prepareGetIndex().addIndices("index").get().indices();
+    }
+
+    private void assertMinimumCapacity(AutoscalingCapacity.AutoscalingResources resources) {
+        assertThat(resources.memory(), equalTo(FrozenExistenceDeciderService.MINIMUM_FROZEN_MEMORY));
+        assertThat(resources.storage(), equalTo(FrozenExistenceDeciderService.MINIMUM_FROZEN_STORAGE));
+    }
+}

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/existence/LocalStateAutoscalingAndSearchableSnapshotsAndIndexLifecycle.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.existence;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.xpack.autoscaling.shards.LocalStateAutoscalingAndSearchableSnapshots;
+import org.elasticsearch.xpack.ilm.IndexLifecycle;
+
+/**
+ * We need a local state plugin including both searchable snapshots and ilm in order to verify the frozen 0-1 case works through ilm.
+ * The local state plugin is necessary to avoid touching the "static SetOnce" licenseState field in XPackPlugin.
+ */
+public class LocalStateAutoscalingAndSearchableSnapshotsAndIndexLifecycle extends LocalStateAutoscalingAndSearchableSnapshots {
+
+    public LocalStateAutoscalingAndSearchableSnapshotsAndIndexLifecycle(final Settings settings) {
+        super(settings);
+        plugins.add(new IndexLifecycle(settings) {
+            @Override
+            protected XPackLicenseState getLicenseState() {
+                return LocalStateAutoscalingAndSearchableSnapshotsAndIndexLifecycle.this.getLicenseState();
+            }
+        });
+    }
+}

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/shards/FrozenShardsDeciderIT.java

@@ -18,7 +18,10 @@ public class FrozenShardsDeciderIT extends org.elasticsearch.xpack.autoscaling.A
         return 1;
     }
 
-    public void testScale() {
+    public void testScale() throws Exception {
+        setupRepoAndPolicy();
+        createAndMountIndex();
+
         assertThat(
             capacity().results().get("frozen").requiredCapacity().total().memory(),
             equalTo(FrozenShardsDeciderService.DEFAULT_MEMORY_PER_SHARD)

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/storage/FrozenStorageDeciderIT.java

@@ -20,7 +20,10 @@ import static org.hamcrest.Matchers.equalTo;
 
 public class FrozenStorageDeciderIT extends AbstractFrozenAutoscalingIntegTestCase {
 
-    public void testScale() {
+    public void testScale() throws Exception {
+        setupRepoAndPolicy();
+        createAndMountIndex();
+
         IndicesStatsResponse statsResponse = client().admin()
             .indices()
             .stats(new IndicesStatsRequest().indices(restoredIndexName))

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/Autoscaling.java

@@ -49,6 +49,7 @@ import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderResult;
 import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderService;
 import org.elasticsearch.xpack.autoscaling.capacity.FixedAutoscalingDeciderService;
 import org.elasticsearch.xpack.autoscaling.capacity.memory.AutoscalingMemoryInfoService;
+import org.elasticsearch.xpack.autoscaling.existence.FrozenExistenceDeciderService;
 import org.elasticsearch.xpack.autoscaling.rest.RestDeleteAutoscalingPolicyHandler;
 import org.elasticsearch.xpack.autoscaling.rest.RestGetAutoscalingCapacityHandler;
 import org.elasticsearch.xpack.autoscaling.rest.RestGetAutoscalingPolicyHandler;
@@ -176,6 +177,11 @@ public class Autoscaling extends Plugin implements ActionPlugin, ExtensiblePlugi
                 AutoscalingDeciderResult.Reason.class,
                 FrozenStorageDeciderService.NAME,
                 FrozenStorageDeciderService.FrozenReason::new
+            ),
+            new NamedWriteableRegistry.Entry(
+                AutoscalingDeciderResult.Reason.class,
+                FrozenExistenceDeciderService.NAME,
+                FrozenExistenceDeciderService.FrozenExistenceReason::new
             )
         );
     }
@@ -208,7 +214,8 @@ public class Autoscaling extends Plugin implements ActionPlugin, ExtensiblePlugi
                 allocationDeciders.get()
             ),
             new FrozenShardsDeciderService(),
-            new FrozenStorageDeciderService()
+            new FrozenStorageDeciderService(),
+            new FrozenExistenceDeciderService()
         );
     }
 

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/existence/FrozenExistenceDeciderService.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.existence;
+
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingCapacity;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderContext;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderResult;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderService;
+import org.elasticsearch.xpack.core.ilm.LifecycleExecutionState;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+/**
+ * This decider looks at all indices and ensures a minimum capacity is available if any indices are in the frozen ILM phase, since that
+ * is designated for partially mounted indices/frozen tier only. Effectively, this scales the tier into existence.
+ *
+ * This works in concert with the `WaitForDataTierStep` in ILM that ensures we wait for autoscaling to spin up the first frozen tier node.
+ */
+public class FrozenExistenceDeciderService implements AutoscalingDeciderService {
+    public static final String NAME = "frozen_existence";
+    static final ByteSizeValue MINIMUM_FROZEN_MEMORY = ByteSizeValue.ofGb(1);
+    static final ByteSizeValue MINIMUM_FROZEN_STORAGE = ByteSizeValue.ofGb(8);
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @Override
+    public AutoscalingDeciderResult scale(Settings configuration, AutoscalingDeciderContext context) {
+        List<String> indicesNeedingFrozen = StreamSupport.stream(context.state().metadata().spliterator(), false)
+            .filter(this::needsTier)
+            .map(imd -> imd.getIndex().getName())
+            .limit(10)
+            .collect(Collectors.toList());
+        AutoscalingCapacity.Builder builder = AutoscalingCapacity.builder();
+        if (indicesNeedingFrozen.size() > 0) {
+            builder.total(MINIMUM_FROZEN_STORAGE, MINIMUM_FROZEN_MEMORY);
+            builder.node(MINIMUM_FROZEN_STORAGE, MINIMUM_FROZEN_MEMORY);
+        } else {
+            builder.total(0L, 0L);
+        }
+
+        return new AutoscalingDeciderResult(builder.build(), new FrozenExistenceReason(indicesNeedingFrozen));
+    }
+
+    boolean needsTier(IndexMetadata idxMeta) {
+        return LifecycleExecutionState.isFrozenPhase(idxMeta);
+    }
+
+    @Override
+    public List<Setting<?>> deciderSettings() {
+        return Collections.emptyList();
+    }
+
+    @Override
+    public List<DiscoveryNodeRole> roles() {
+        return List.of(DiscoveryNodeRole.DATA_FROZEN_NODE_ROLE);
+    }
+
+    public static class FrozenExistenceReason implements AutoscalingDeciderResult.Reason {
+        private final List<String> indices;
+
+        public FrozenExistenceReason(List<String> indices) {
+            this.indices = indices;
+        }
+
+        public FrozenExistenceReason(StreamInput in) throws IOException {
+            this.indices = in.readStringList();
+        }
+
+        @Override
+        public String summary() {
+            return "indices " + indices;
+        }
+
+        public List<String> indices() {
+            return indices;
+        }
+
+        @Override
+        public String getWriteableName() {
+            return NAME;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeStringCollection(indices);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field("indices", indices);
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            FrozenExistenceReason that = (FrozenExistenceReason) o;
+            return indices.equals(that.indices);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(indices);
+        }
+    }
+
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/existence/FrozenExistenceDeciderServiceTests.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.existence;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingCapacity;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderContext;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderResult;
+import org.elasticsearch.xpack.core.ilm.LifecycleExecutionState;
+
+import java.util.function.Consumer;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class FrozenExistenceDeciderServiceTests extends AutoscalingTestCase {
+
+    public void testScale() {
+        verify(ClusterState.EMPTY_STATE, this::assertZeroCapacity);
+
+        final Settings versionSettings = Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT).build();
+        final int shards = between(1, 3);
+        final int replicas = between(0, 2);
+        final Metadata nonFrozenMetadata = Metadata.builder()
+            .put(IndexMetadata.builder("index").settings(versionSettings).numberOfShards(shards).numberOfReplicas(replicas))
+            .build();
+        verify(nonFrozenMetadata, this::assertZeroCapacity);
+
+        final Metadata frozenMetadata = (randomBoolean() ? Metadata.builder() : Metadata.builder(nonFrozenMetadata)).put(
+            IndexMetadata.builder("index")
+                .settings(versionSettings)
+                .putCustom(
+                    LifecycleExecutionState.ILM_CUSTOM_METADATA_KEY,
+                    LifecycleExecutionState.builder().setPhase("frozen").build().asMap()
+                )
+                .numberOfShards(shards)
+                .numberOfReplicas(replicas)
+        ).build();
+        verify(frozenMetadata, this::assertMinimumCapacity);
+    }
+
+    private void verify(Metadata metadata, Consumer<AutoscalingDeciderResult> resultConsumer) {
+        verify(ClusterState.builder(ClusterName.DEFAULT).metadata(metadata).build(), resultConsumer);
+    }
+
+    private void verify(ClusterState state, Consumer<AutoscalingDeciderResult> resultConsumer) {
+        FrozenExistenceDeciderService service = new FrozenExistenceDeciderService();
+        AutoscalingDeciderContext context = mock(AutoscalingDeciderContext.class);
+        when(context.state()).thenReturn(state);
+        resultConsumer.accept(service.scale(Settings.EMPTY, context));
+    }
+
+    private void assertMinimumCapacity(AutoscalingDeciderResult result) {
+        AutoscalingCapacity capacity = result.requiredCapacity();
+        assertThat(capacity.total().memory(), equalTo(FrozenExistenceDeciderService.MINIMUM_FROZEN_MEMORY));
+        assertThat(capacity.total().storage(), equalTo(FrozenExistenceDeciderService.MINIMUM_FROZEN_STORAGE));
+        assertThat(capacity.node().memory(), equalTo(FrozenExistenceDeciderService.MINIMUM_FROZEN_MEMORY));
+        assertThat(capacity.node().storage(), equalTo(FrozenExistenceDeciderService.MINIMUM_FROZEN_STORAGE));
+        assertThat(result.reason().summary(), equalTo("indices [index]"));
+    }
+
+    private void assertZeroCapacity(AutoscalingDeciderResult result) {
+        AutoscalingCapacity capacity = result.requiredCapacity();
+        assertThat(capacity.total().memory(), equalTo(ByteSizeValue.ZERO));
+        assertThat(capacity.total().storage(), equalTo(ByteSizeValue.ZERO));
+        assertThat(capacity.node(), is(nullValue()));
+        assertThat(result.reason().summary(), equalTo("indices []"));
+    }
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/existence/FrozenExistenceReasonWireSerializationTests.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.existence;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class FrozenExistenceReasonWireSerializationTests extends AbstractWireSerializingTestCase<
+    FrozenExistenceDeciderService.FrozenExistenceReason> {
+    @Override
+    protected Writeable.Reader<FrozenExistenceDeciderService.FrozenExistenceReason> instanceReader() {
+        return FrozenExistenceDeciderService.FrozenExistenceReason::new;
+    }
+
+    @Override
+    protected FrozenExistenceDeciderService.FrozenExistenceReason createTestInstance() {
+        return new FrozenExistenceDeciderService.FrozenExistenceReason(randomList(between(0, 10), () -> randomAlphaOfLength(5)));
+    }
+
+    @Override
+    protected FrozenExistenceDeciderService.FrozenExistenceReason mutateInstance(
+        FrozenExistenceDeciderService.FrozenExistenceReason instance
+    ) {
+        List<String> indices = new ArrayList<>(instance.indices());
+        if (indices.isEmpty() || randomBoolean()) {
+            indices.add(randomAlphaOfLength(5));
+        } else {
+            indices.remove(between(0, indices.size() - 1));
+        }
+        return new FrozenExistenceDeciderService.FrozenExistenceReason(indices);
+    }
+}

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

@@ -10,6 +10,7 @@ import org.elasticsearch.client.Client;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.common.io.stream.NamedWriteable;
 import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.license.XPackLicenseState;
 
 import java.util.List;
 
@@ -30,6 +31,21 @@ public interface LifecycleAction extends ToXContentObject, NamedWriteable {
      */
     List<Step> toSteps(Client client, String phase, @Nullable Step.StepKey nextStepKey);
 
+    /**
+     * converts the {@link LifecycleAction}'s execution plan into a series of
+     * {@link Step}s that reference each other to preserve order of operations. This overload allows access to license state.
+     * @param client      the client that will be used by {@link AsyncActionStep} and {@link AsyncWaitStep} steps
+     * @param phase       the name of the phase this action is being executed within
+     * @param nextStepKey the next step to execute after this action's steps. If null, then there are no further
+     *                    steps to run. It is the responsibility of each {@link LifecycleAction} to implement this
+     *                    correctly and not forget to link to this final step so that the policy can continue.
+     * @param licenseState the license state to use for actions that need license checks.
+     * @return an ordered list of steps that represent the execution plan of the action
+     */
+    default List<Step> toSteps(Client client, String phase, @Nullable Step.StepKey nextStepKey, XPackLicenseState licenseState) {
+        return toSteps(client, phase, nextStepKey);
+    }
+
     /**
      * @return true if this action is considered safe. An action is not safe if
      *         it will produce unwanted side effects or will get stuck when the

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

@@ -97,6 +97,17 @@ public class LifecycleExecutionState {
         return fromCustomMetadata(customData);
     }
 
+    /**
+     * Return true if this index is in the frozen phase, false if not controlled by ILM or not in frozen.
+     * @param indexMetadata the metadata of the index to retrieve phase from.
+     * @return true if frozen phase, false otherwise.
+     */
+    public static boolean isFrozenPhase(IndexMetadata indexMetadata) {
+        Map<String, String> customData = indexMetadata.getCustomData(ILM_CUSTOM_METADATA_KEY);
+        // deliberately do not parse out the entire `LifeCycleExecutionState` to avoid the extra work involved since this method is
+        // used heavily by autoscaling.
+        return customData != null && TimeseriesLifecycleType.FROZEN_PHASE.equals(customData.get(PHASE));
+    }
     /**
      * Retrieves the current {@link Step.StepKey} from the lifecycle state. Note that
      * it is illegal for the step to be set with the phase and/or action unset,

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

@@ -19,6 +19,7 @@ import org.elasticsearch.common.xcontent.ConstructingObjectParser;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 
 import java.io.IOException;
@@ -209,9 +210,10 @@ public class LifecyclePolicy extends AbstractDiffable<LifecyclePolicy>
      *
      * @param client The Elasticsearch Client to use during execution of {@link AsyncActionStep}
      *               and {@link AsyncWaitStep} steps.
+     * @param licenseState The license state to use in actions and steps
      * @return The list of {@link Step} objects in order of their execution.
      */
-    public List<Step> toSteps(Client client) {
+    public List<Step> toSteps(Client client, XPackLicenseState licenseState) {
         List<Step> steps = new ArrayList<>();
         List<Phase> orderedPhases = type.getOrderedPhases(phases);
         ListIterator<Phase> phaseIterator = orderedPhases.listIterator(orderedPhases.size());
@@ -240,7 +242,7 @@ public class LifecyclePolicy extends AbstractDiffable<LifecyclePolicy>
             // add steps for each action, in reverse
             while (actionIterator.hasPrevious()) {
                 LifecycleAction action = actionIterator.previous();
-                List<Step> actionSteps = action.toSteps(client, phase.getName(), lastStepKey);
+                List<Step> actionSteps = action.toSteps(client, phase.getName(), lastStepKey, licenseState);
                 ListIterator<Step> actionStepsIterator = actionSteps.listIterator(actionSteps.size());
                 while (actionStepsIterator.hasPrevious()) {
                     Step step = actionStepsIterator.previous();

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

@@ -23,6 +23,7 @@ import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotA
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
 
 import java.util.Objects;
+import java.util.Optional;
 
 import static org.elasticsearch.xpack.core.ilm.LifecycleExecutionState.fromIndexMetadata;
 
@@ -104,10 +105,10 @@ public class MountSnapshotStep extends AsyncRetryDuringSnapshotActionStep {
         }
 
         final Settings.Builder settingsBuilder = Settings.builder();
-        // if we are mounting a searchable snapshot in the hot phase, then the index should be pinned to the hot nodes
-        if (TimeseriesLifecycleType.HOT_PHASE.equals(this.getKey().getPhase())) {
-            settingsBuilder.put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, DataTier.DATA_HOT);
-        }
+
+        overrideTierPreference(this.getKey().getPhase())
+            .ifPresent(override -> settingsBuilder.put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, override));
+
         final MountSearchableSnapshotRequest mountSearchableSnapshotRequest = new MountSearchableSnapshotRequest(mountedIndexName,
             snapshotRepository, snapshotName, indexName, settingsBuilder.build(),
             // we captured the index metadata when we took the snapshot. the index likely had the ILM execution state in the metadata.
@@ -143,6 +144,19 @@ public class MountSnapshotStep extends AsyncRetryDuringSnapshotActionStep {
         return originalName;
     }
 
+    /**
+     * return the tier preference to use or empty to use default.
+     * @param phase the phase the step will run in.
+     * @return tier preference override or empty.
+     */
+    static Optional<String> overrideTierPreference(String phase) {
+        // if we are mounting a searchable snapshot in the hot phase, then the index should be pinned to the hot nodes
+        if (TimeseriesLifecycleType.HOT_PHASE.equals(phase)) {
+            return Optional.of(DataTier.DATA_HOT);
+        }
+        return Optional.empty();
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(super.hashCode(), restoredIndexPrefix, storageType);

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

@@ -20,6 +20,7 @@ import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.license.XPackLicenseState;
 
 import java.util.ArrayList;
 import java.util.LinkedHashSet;
@@ -107,9 +108,9 @@ public final class PhaseCacheManagement {
      */
     public static ClusterState updateIndicesForPolicy(final ClusterState state, final NamedXContentRegistry xContentRegistry,
                                                       final Client client, final LifecyclePolicy oldPolicy,
-                                                      final LifecyclePolicyMetadata newPolicy) {
+                                                      final LifecyclePolicyMetadata newPolicy, XPackLicenseState licenseState) {
         Metadata.Builder mb = Metadata.builder(state.metadata());
-        if (updateIndicesForPolicy(mb, state, xContentRegistry, client, oldPolicy, newPolicy)) {
+        if (updateIndicesForPolicy(mb, state, xContentRegistry, client, oldPolicy, newPolicy, licenseState)) {
             return ClusterState.builder(state).metadata(mb).build();
         }
         return state;
@@ -122,7 +123,8 @@ public final class PhaseCacheManagement {
      */
     public static boolean updateIndicesForPolicy(final Metadata.Builder mb, final ClusterState currentState,
                                                  final NamedXContentRegistry xContentRegistry, final Client client,
-                                                 final LifecyclePolicy oldPolicy, final LifecyclePolicyMetadata newPolicy) {
+                                                 final LifecyclePolicy oldPolicy, final LifecyclePolicyMetadata newPolicy,
+                                                 final XPackLicenseState licenseState) {
         assert oldPolicy.getName().equals(newPolicy.getName()) : "expected both policies to have the same id but they were: [" +
             oldPolicy.getName() + "] vs. [" + newPolicy.getName() + "]";
 
@@ -135,7 +137,7 @@ public final class PhaseCacheManagement {
         final List<IndexMetadata> indicesThatCanBeUpdated =
             StreamSupport.stream(Spliterators.spliteratorUnknownSize(currentState.metadata().indices().valuesIt(), 0), false)
                 .filter(meta -> newPolicy.getName().equals(LifecycleSettings.LIFECYCLE_NAME_SETTING.get(meta.getSettings())))
-                .filter(meta -> isIndexPhaseDefinitionUpdatable(xContentRegistry, client, meta, newPolicy.getPolicy()))
+                .filter(meta -> isIndexPhaseDefinitionUpdatable(xContentRegistry, client, meta, newPolicy.getPolicy(), licenseState))
                 .collect(Collectors.toList());
 
         final List<String> refreshedIndices = new ArrayList<>(indicesThatCanBeUpdated.size());
@@ -156,7 +158,8 @@ public final class PhaseCacheManagement {
      * Returns 'true' if the index's cached phase JSON can be safely reread, 'false' otherwise.
      */
     public static boolean isIndexPhaseDefinitionUpdatable(final NamedXContentRegistry xContentRegistry, final Client client,
-                                                          final IndexMetadata metadata, final LifecyclePolicy newPolicy) {
+                                                          final IndexMetadata metadata, final LifecyclePolicy newPolicy,
+                                                          final XPackLicenseState licenseState) {
         final String index = metadata.getIndex().getName();
         if (eligibleToCheckForRefresh(metadata) == false) {
             logger.debug("[{}] does not contain enough information to check for eligibility of refreshing phase", index);
@@ -168,7 +171,7 @@ public final class PhaseCacheManagement {
         final Step.StepKey currentStepKey = LifecycleExecutionState.getCurrentStepKey(executionState);
         final String currentPhase = currentStepKey.getPhase();
 
-        final Set<Step.StepKey> newStepKeys = newPolicy.toSteps(client).stream()
+        final Set<Step.StepKey> newStepKeys = newPolicy.toSteps(client, licenseState).stream()
             .map(Step::getKey)
             .collect(Collectors.toCollection(LinkedHashSet::new));
 
@@ -181,7 +184,7 @@ public final class PhaseCacheManagement {
         }
 
         final String phaseDef = executionState.getPhaseDefinition();
-        final Set<Step.StepKey> oldStepKeys = readStepKeys(xContentRegistry, client, phaseDef, currentPhase);
+        final Set<Step.StepKey> oldStepKeys = readStepKeys(xContentRegistry, client, phaseDef, currentPhase, licenseState);
         if (oldStepKeys == null) {
             logger.debug("[{}] unable to parse phase definition for cached policy [{}], policy phase will not be refreshed",
                 index, policyId);
@@ -195,7 +198,7 @@ public final class PhaseCacheManagement {
         final PhaseExecutionInfo phaseExecutionInfo = new PhaseExecutionInfo(policyId, newPolicy.getPhases().get(currentPhase), 1L, 1L);
         final String peiJson = Strings.toString(phaseExecutionInfo);
 
-        final Set<Step.StepKey> newPhaseStepKeys = readStepKeys(xContentRegistry, client, peiJson, currentPhase);
+        final Set<Step.StepKey> newPhaseStepKeys = readStepKeys(xContentRegistry, client, peiJson, currentPhase, licenseState);
         if (newPhaseStepKeys == null) {
             logger.debug(new ParameterizedMessage("[{}] unable to parse phase definition for policy [{}] " +
                 "to determine if it could be refreshed", index, policyId));
@@ -222,7 +225,7 @@ public final class PhaseCacheManagement {
      */
     @Nullable
     static Set<Step.StepKey> readStepKeys(final NamedXContentRegistry xContentRegistry, final Client client,
-                                          final String phaseDef, final String currentPhase) {
+                                          final String phaseDef, final String currentPhase, final XPackLicenseState licenseState) {
         final PhaseExecutionInfo phaseExecutionInfo;
         try (XContentParser parser = JsonXContent.jsonXContent.createParser(xContentRegistry,
             DeprecationHandler.THROW_UNSUPPORTED_OPERATION, phaseDef)) {
@@ -238,7 +241,7 @@ public final class PhaseCacheManagement {
         }
 
         return phaseExecutionInfo.getPhase().getActions().values().stream()
-            .flatMap(a -> a.toSteps(client, phaseExecutionInfo.getPhase().getName(), null).stream())
+            .flatMap(a -> a.toSteps(client, phaseExecutionInfo.getPhase().getName(), null, licenseState).stream())
             .map(Step::getKey)
             .collect(Collectors.toCollection(LinkedHashSet::new));
     }

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

@@ -22,7 +22,6 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.repositories.RepositoriesService;
-import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
 
@@ -93,6 +92,12 @@ public class SearchableSnapshotAction implements LifecycleAction {
 
     @Override
     public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
+        assert false;
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public List<Step> toSteps(Client client, String phase, StepKey nextStepKey, XPackLicenseState licenseState) {
         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);
@@ -102,6 +107,7 @@ public class SearchableSnapshotAction implements LifecycleAction {
         StepKey generateSnapshotNameKey = new StepKey(phase, NAME, GenerateSnapshotNameStep.NAME);
         StepKey cleanSnapshotKey = new StepKey(phase, NAME, CleanupSnapshotStep.NAME);
         StepKey createSnapshotKey = new StepKey(phase, NAME, CreateSnapshotStep.NAME);
+        StepKey waitForDataTierKey = new StepKey(phase, NAME, WaitForDataTierStep.NAME);
         StepKey mountSnapshotKey = new StepKey(phase, NAME, MountSnapshotStep.NAME);
         StepKey waitForGreenRestoredIndexKey = new StepKey(phase, NAME, WaitForIndexColorStep.NAME);
         StepKey copyMetadataKey = new StepKey(phase, NAME, CopyExecutionStateStep.NAME);
@@ -117,7 +123,6 @@ public class SearchableSnapshotAction implements LifecycleAction {
         // here before generating snapshots that can't be used if the user doesn't have the right license level.
         BranchingStep conditionalSkipActionStep = new BranchingStep(preActionBranchingKey, checkNoWriteIndex, nextStepKey,
             (index, clusterState) -> {
-                XPackLicenseState licenseState = XPackPlugin.getSharedLicenseState();
                 if (licenseState.isAllowed(XPackLicenseState.Feature.SEARCHABLE_SNAPSHOTS) == false) {
                     logger.error("[{}] action is not available in the current license", SearchableSnapshotAction.NAME);
                     throw LicenseUtils.newComplianceException("searchable-snapshots");
@@ -171,7 +176,7 @@ public class SearchableSnapshotAction implements LifecycleAction {
         // Branch, deciding whether there is an existing searchable snapshot snapshot that can be used for mounting the index
         // (in which case, skip generating a new name and the snapshot cleanup), or if we need to generate a new snapshot
         BranchingStep skipGeneratingSnapshotStep =
-            new BranchingStep(skipGeneratingSnapshotKey, keyForSnapshotGeneration, mountSnapshotKey, (index, clusterState) -> {
+            new BranchingStep(skipGeneratingSnapshotKey, keyForSnapshotGeneration, waitForDataTierKey, (index, clusterState) -> {
                 IndexMetadata indexMetadata = clusterState.getMetadata().index(index);
                 String policyName = LifecycleSettings.LIFECYCLE_NAME_SETTING.get(indexMetadata.getSettings());
                 LifecycleExecutionState lifecycleExecutionState = LifecycleExecutionState.fromIndexMetadata(indexMetadata);
@@ -204,12 +209,17 @@ public class SearchableSnapshotAction implements LifecycleAction {
             snapshotRepository);
         CleanupSnapshotStep cleanupSnapshotStep = new CleanupSnapshotStep(cleanSnapshotKey, createSnapshotKey, client);
         AsyncActionBranchingStep createSnapshotBranchingStep = new AsyncActionBranchingStep(
-            new CreateSnapshotStep(createSnapshotKey, mountSnapshotKey, client), cleanSnapshotKey, client);
+            new CreateSnapshotStep(createSnapshotKey, waitForDataTierKey, client), cleanSnapshotKey, client);
+
+        MountSearchableSnapshotRequest.Storage storageType = getConcreteStorageType(mountSnapshotKey);
 
-        // Now mount the snapshot to create the new index, if the skipGeneratingSnapshotStep determined a snapshot already existed that
+        // If the skipGeneratingSnapshotStep determined a snapshot already existed that
         // can be used, it jumps directly here, skipping the snapshot generation steps above.
+        WaitForDataTierStep waitForDataTierStep =
+            new WaitForDataTierStep(waitForDataTierKey, mountSnapshotKey,
+                MountSnapshotStep.overrideTierPreference(phase).orElse(storageType.defaultDataTiersPreference()));
         MountSnapshotStep mountSnapshotStep = new MountSnapshotStep(mountSnapshotKey, waitForGreenRestoredIndexKey,
-            client, getRestoredIndexPrefix(mountSnapshotKey), getConcreteStorageType(mountSnapshotKey));
+            client, getRestoredIndexPrefix(mountSnapshotKey), storageType);
         WaitForIndexColorStep waitForGreenIndexHealthStep = new WaitForIndexColorStep(waitForGreenRestoredIndexKey,
             copyMetadataKey, ClusterHealthStatus.GREEN, getRestoredIndexPrefix(waitForGreenRestoredIndexKey));
         CopyExecutionStateStep copyMetadataStep = new CopyExecutionStateStep(copyMetadataKey, copyLifecyclePolicySettingKey,
@@ -242,6 +252,7 @@ public class SearchableSnapshotAction implements LifecycleAction {
         steps.add(generateSnapshotNameStep);
         steps.add(cleanupSnapshotStep);
         steps.add(createSnapshotBranchingStep);
+        steps.add(waitForDataTierStep);
         steps.add(mountSnapshotStep);
         steps.add(waitForGreenIndexHealthStep);
         steps.add(copyMetadataStep);

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.ilm;
+
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDecider;
+import org.elasticsearch.xpack.core.ilm.step.info.SingleMessageFieldInfo;
+
+import java.util.Objects;
+
+/**
+ * This step waits for one of the data tiers to be available in the cluster. This has two purposes:
+ * <ul>
+ *     <li>Avoid a mounted index going RED, it is better to pause ILM on this condition</li>
+ *     <li>Leave a signal to autoscaling to scale up the first node for the tier</li>
+ * </ul>
+ */
+public class WaitForDataTierStep extends ClusterStateWaitStep {
+    public static final String NAME = "wait-for-data-tier";
+    private final String tierPreference;
+
+    public WaitForDataTierStep(StepKey key, StepKey nextStepKey, String tierPreference) {
+        super(key, nextStepKey);
+        this.tierPreference = Objects.requireNonNull(tierPreference);
+    }
+
+    @Override
+    public Result isConditionMet(Index index, ClusterState clusterState) {
+        boolean present = DataTierAllocationDecider.preferredAvailableTier(tierPreference, clusterState.nodes()).isPresent();
+        SingleMessageFieldInfo info = present ? null : new SingleMessageFieldInfo("no nodes for tiers [" + tierPreference + "] available");
+        return new Result(present, info);
+    }
+
+    @Override
+    public boolean isRetryable() {
+        return true;
+    }
+
+    String tierPreference() {
+        return tierPreference;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        if (super.equals(o) == false) return false;
+        WaitForDataTierStep that = (WaitForDataTierStep) o;
+        return tierPreference.equals(that.tierPreference);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), tierPreference);
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/searchablesnapshots/MountSearchableSnapshotRequest.java

@@ -20,6 +20,7 @@ import org.elasticsearch.common.xcontent.ConstructingObjectParser;
 import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.xpack.core.DataTier;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -230,8 +231,23 @@ public class MountSearchableSnapshotRequest extends MasterNodeRequest<MountSearc
      * Enumerates the different ways that nodes can use their local storage to accelerate searches of a snapshot.
      */
     public enum Storage implements Writeable {
-        FULL_COPY,
-        SHARED_CACHE;
+        FULL_COPY(String.join(",", DataTier.DATA_COLD, DataTier.DATA_WARM, DataTier.DATA_HOT)),
+        SHARED_CACHE(DataTier.DATA_FROZEN);
+
+        private final String defaultDataTiersPreference;
+
+        Storage(String defaultDataTiersPreference) {
+            this.defaultDataTiersPreference = defaultDataTiersPreference;
+        }
+
+        /**
+         * Returns the default preference for new searchable snapshot indices. When
+         * performing a full mount the preference is cold - warm - hot. When
+         * performing a partial mount the preference is only frozen
+         */
+        public String defaultDataTiersPreference() {
+            return defaultDataTiersPreference;
+        }
 
         public static Storage fromString(String type) {
             if ("full_copy".equals(type)) {

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

@@ -312,7 +312,7 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
         lifecycleName = randomAlphaOfLengthBetween(1, 20);
         Map<String, Phase> phases = new LinkedHashMap<>();
         LifecyclePolicy policy = new LifecyclePolicy(TestLifecycleType.INSTANCE, lifecycleName, phases, randomMeta());
-        List<Step> steps = policy.toSteps(client);
+        List<Step> steps = policy.toSteps(client, null);
         assertThat(steps.size(), equalTo(2));
         assertThat(steps.get(0), instanceOf(InitializePolicyContextStep.class));
         assertThat(steps.get(0).getKey(), equalTo(new StepKey("new", "init", "init")));
@@ -334,7 +334,7 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
         LifecyclePolicy policy = new LifecyclePolicy(TestLifecycleType.INSTANCE, lifecycleName, phases, randomMeta());
         StepKey firstStepKey = InitializePolicyContextStep.KEY;
         StepKey secondStepKey = PhaseCompleteStep.finalStep("new").getKey();
-        List<Step> steps = policy.toSteps(client);
+        List<Step> steps = policy.toSteps(client, null);
         assertThat(steps.size(), equalTo(4));
         assertSame(steps.get(0).getKey(), firstStepKey);
         assertThat(steps.get(0).getNextStepKey(), equalTo(secondStepKey));
@@ -368,7 +368,7 @@ public class LifecyclePolicyTests extends AbstractSerializingTestCase<LifecycleP
         phases.put(secondPhase.getName(), secondPhase);
         LifecyclePolicy policy = new LifecyclePolicy(TestLifecycleType.INSTANCE, lifecycleName, phases, randomMeta());
 
-        List<Step> steps = policy.toSteps(client);
+        List<Step> steps = policy.toSteps(client, null);
         assertThat(steps.size(), equalTo(7));
         assertThat(steps.get(0).getClass(), equalTo(InitializePolicyContextStep.class));
         assertThat(steps.get(0).getKey(), equalTo(init.getKey()));

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

@@ -188,9 +188,9 @@ public class PhaseCacheManagementTests extends ESTestCase {
     }
 
     public void testReadStepKeys() {
-        assertNull(readStepKeys(REGISTRY, client, "{}", "phase"));
-        assertNull(readStepKeys(REGISTRY, client, "aoeu", "phase"));
-        assertNull(readStepKeys(REGISTRY, client, "", "phase"));
+        assertNull(readStepKeys(REGISTRY, client, "{}", "phase", null));
+        assertNull(readStepKeys(REGISTRY, client, "aoeu", "phase", null));
+        assertNull(readStepKeys(REGISTRY, client, "", "phase", null));
 
         assertThat(readStepKeys(REGISTRY, client, "{\n" +
                 "        \"policy\": \"my_lifecycle3\",\n" +
@@ -204,7 +204,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
                 "        },\n" +
                 "        \"version\": 3, \n" +
                 "        \"modified_date_in_millis\": 1539609701576 \n" +
-                "      }", "phase"),
+                "      }", "phase", null),
             contains(new Step.StepKey("phase", "rollover", WaitForRolloverReadyStep.NAME),
                 new Step.StepKey("phase", "rollover", RolloverStep.NAME),
                 new Step.StepKey("phase", "rollover", WaitForActiveShardsStep.NAME),
@@ -226,7 +226,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
                 "        },\n" +
                 "        \"version\" : 1,\n" +
                 "        \"modified_date_in_millis\" : 1578521007076\n" +
-                "      }", "phase"),
+                "      }", "phase", null),
             contains(new Step.StepKey("phase", "rollover", WaitForRolloverReadyStep.NAME),
                 new Step.StepKey("phase", "rollover", RolloverStep.NAME),
                 new Step.StepKey("phase", "rollover", WaitForActiveShardsStep.NAME),
@@ -241,7 +241,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
         String phaseDef = Strings.toString(pei);
         logger.info("--> phaseDef: {}", phaseDef);
 
-        assertThat(readStepKeys(REGISTRY, client, phaseDef, "phase"),
+        assertThat(readStepKeys(REGISTRY, client, phaseDef, "phase", null),
             contains(
                 new Step.StepKey("phase", "allocate", AllocateAction.NAME),
                 new Step.StepKey("phase", "allocate", AllocationRoutedStep.NAME),
@@ -290,7 +290,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
             Map<String, Phase> phases = Collections.singletonMap("hot", hotPhase);
             LifecyclePolicy newPolicy = new LifecyclePolicy("my-policy", phases);
 
-            assertTrue(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy));
+            assertTrue(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy, null));
         }
 
         // Failure case, can't update because the step we're currently on has been removed in the new policy
@@ -327,7 +327,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
             Map<String, Phase> phases = Collections.singletonMap("hot", hotPhase);
             LifecyclePolicy newPolicy = new LifecyclePolicy("my-policy", phases);
 
-            assertFalse(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy));
+            assertFalse(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy, null));
         }
 
         // Failure case, can't update because the future step has been deleted
@@ -364,7 +364,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
             Map<String, Phase> phases = Collections.singletonMap("hot", hotPhase);
             LifecyclePolicy newPolicy = new LifecyclePolicy("my-policy", phases);
 
-            assertFalse(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy));
+            assertFalse(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy, null));
         }
 
         // Failure case, index doesn't have enough info to check
@@ -399,7 +399,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
             Map<String, Phase> phases = Collections.singletonMap("hot", hotPhase);
             LifecyclePolicy newPolicy = new LifecyclePolicy("my-policy", phases);
 
-            assertFalse(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy));
+            assertFalse(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy, null));
         }
 
         // Failure case, the phase JSON is unparseable
@@ -422,7 +422,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
             Map<String, Phase> phases = Collections.singletonMap("hot", hotPhase);
             LifecyclePolicy newPolicy = new LifecyclePolicy("my-policy", phases);
 
-            assertFalse(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy));
+            assertFalse(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy, null));
         }
     }
 
@@ -456,7 +456,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
         LifecyclePolicy newPolicy = new LifecyclePolicy("my-policy", phases);
         LifecyclePolicyMetadata policyMetadata = new LifecyclePolicyMetadata(newPolicy, Collections.emptyMap(), 2L, 2L);
 
-        assertTrue(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy));
+        assertTrue(isIndexPhaseDefinitionUpdatable(REGISTRY, client, meta, newPolicy, null));
 
         ClusterState existingState = ClusterState.builder(ClusterState.EMPTY_STATE)
             .metadata(Metadata.builder(Metadata.EMPTY_METADATA)
@@ -465,7 +465,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
             .build();
 
         logger.info("--> update for unchanged policy");
-        ClusterState updatedState = updateIndicesForPolicy(existingState, REGISTRY, client, oldPolicy, policyMetadata);
+        ClusterState updatedState = updateIndicesForPolicy(existingState, REGISTRY, client, oldPolicy, policyMetadata, null);
 
         // No change, because the policies were identical
         assertThat(updatedState, equalTo(existingState));
@@ -479,7 +479,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
         policyMetadata = new LifecyclePolicyMetadata(newPolicy, Collections.emptyMap(), 2L, 2L);
 
         logger.info("--> update with changed policy, but not configured in settings");
-        updatedState = updateIndicesForPolicy(existingState, REGISTRY, client, oldPolicy, policyMetadata);
+        updatedState = updateIndicesForPolicy(existingState, REGISTRY, client, oldPolicy, policyMetadata, null);
 
         // No change, because the index doesn't have a lifecycle.name setting for this policy
         assertThat(updatedState, equalTo(existingState));
@@ -500,7 +500,7 @@ public class PhaseCacheManagementTests extends ESTestCase {
             .build();
 
         logger.info("--> update with changed policy and this index has the policy");
-        updatedState = updateIndicesForPolicy(existingState, REGISTRY, client, oldPolicy, policyMetadata);
+        updatedState = updateIndicesForPolicy(existingState, REGISTRY, client, oldPolicy, policyMetadata, null);
 
         IndexMetadata newIdxMeta = updatedState.metadata().index(index);
         LifecycleExecutionState afterExState = LifecycleExecutionState.fromIndexMetadata(newIdxMeta);

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

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.core.ilm;
 
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.xpack.core.DataTier;
 import org.elasticsearch.xpack.core.ilm.Step.StepKey;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
 
@@ -22,12 +23,12 @@ public class SearchableSnapshotActionTests extends AbstractActionTestCase<Search
 
     @Override
     public void testToSteps() {
-        String phase = randomAlphaOfLengthBetween(1, 10);
+        String phase = randomBoolean() ? randomFrom(TimeseriesLifecycleType.ORDERED_VALID_PHASES) : randomAlphaOfLengthBetween(1, 10);
         SearchableSnapshotAction action = createTestInstance();
         StepKey nextStepKey = new StepKey(phase, randomAlphaOfLengthBetween(1, 5), randomAlphaOfLengthBetween(1, 5));
 
-        List<Step> steps = action.toSteps(null, phase, nextStepKey);
-        assertThat(steps.size(), is(action.isForceMergeIndex() ? 17 : 15));
+        List<Step> steps = action.toSteps(null, phase, nextStepKey, null);
+        assertThat(steps.size(), is(action.isForceMergeIndex() ? 18 : 16));
 
         List<StepKey> expectedSteps = action.isForceMergeIndex() ? expectedStepKeysWithForceMerge(phase) :
             expectedStepKeysNoForceMerge(phase);
@@ -47,15 +48,28 @@ public class SearchableSnapshotActionTests extends AbstractActionTestCase<Search
         assertThat(steps.get(12).getKey(), is(expectedSteps.get(12)));
         assertThat(steps.get(13).getKey(), is(expectedSteps.get(13)));
         assertThat(steps.get(14).getKey(), is(expectedSteps.get(14)));
+        assertThat(steps.get(15).getKey(), is(expectedSteps.get(15)));
 
         if (action.isForceMergeIndex()) {
-            assertThat(steps.get(15).getKey(), is(expectedSteps.get(15)));
             assertThat(steps.get(16).getKey(), is(expectedSteps.get(16)));
+            assertThat(steps.get(17).getKey(), is(expectedSteps.get(17)));
             AsyncActionBranchingStep branchStep = (AsyncActionBranchingStep) steps.get(8);
             assertThat(branchStep.getNextKeyOnIncompleteResponse(), is(expectedSteps.get(7)));
+            validateWaitForDataTierStep(phase, steps, 9, 10);
         } else {
             AsyncActionBranchingStep branchStep = (AsyncActionBranchingStep) steps.get(6);
             assertThat(branchStep.getNextKeyOnIncompleteResponse(), is(expectedSteps.get(5)));
+            validateWaitForDataTierStep(phase, steps, 7, 8);
+        }
+    }
+
+    private void validateWaitForDataTierStep(String phase, List<Step> steps, int waitForDataTierStepIndex, int mountStepIndex) {
+        WaitForDataTierStep waitForDataTierStep = (WaitForDataTierStep) steps.get(waitForDataTierStepIndex);
+        if (phase.equals(TimeseriesLifecycleType.HOT_PHASE)) {
+            assertThat(waitForDataTierStep.tierPreference(), equalTo(DataTier.DATA_HOT));
+        } else {
+            MountSnapshotStep mountStep = (MountSnapshotStep) steps.get(mountStepIndex);
+            assertThat(waitForDataTierStep.tierPreference(), equalTo(mountStep.getStorage().defaultDataTiersPreference()));
         }
     }
 
@@ -82,6 +96,7 @@ public class SearchableSnapshotActionTests extends AbstractActionTestCase<Search
             new StepKey(phase, NAME, GenerateSnapshotNameStep.NAME),
             new StepKey(phase, NAME, CleanupSnapshotStep.NAME),
             new StepKey(phase, NAME, CreateSnapshotStep.NAME),
+            new StepKey(phase, NAME, WaitForDataTierStep.NAME),
             new StepKey(phase, NAME, MountSnapshotStep.NAME),
             new StepKey(phase, NAME, WaitForIndexColorStep.NAME),
             new StepKey(phase, NAME, CopyExecutionStateStep.NAME),
@@ -101,6 +116,7 @@ public class SearchableSnapshotActionTests extends AbstractActionTestCase<Search
             new StepKey(phase, NAME, GenerateSnapshotNameStep.NAME),
             new StepKey(phase, NAME, CleanupSnapshotStep.NAME),
             new StepKey(phase, NAME, CreateSnapshotStep.NAME),
+            new StepKey(phase, NAME, WaitForDataTierStep.NAME),
             new StepKey(phase, NAME, MountSnapshotStep.NAME),
             new StepKey(phase, NAME, WaitForIndexColorStep.NAME),
             new StepKey(phase, NAME, CopyExecutionStateStep.NAME),

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.ilm;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.xpack.core.DataTier;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+
+public class WaitForDataTierStepTests extends AbstractStepTestCase<WaitForDataTierStep> {
+
+    @Override
+    protected WaitForDataTierStep createRandomInstance() {
+        return new WaitForDataTierStep(randomStepKey(), randomStepKey(), randomAlphaOfLength(5));
+    }
+
+    @Override
+    protected WaitForDataTierStep mutateInstance(WaitForDataTierStep instance) {
+        switch (between(0, 2)) {
+            case 0:
+                return new WaitForDataTierStep(randomValueOtherThan(instance.getKey(), AbstractStepTestCase::randomStepKey),
+                    instance.getNextStepKey(),
+                    instance.tierPreference());
+            case 1:
+                return new WaitForDataTierStep(instance.getKey(),
+                    randomValueOtherThan(instance.getNextStepKey(), AbstractStepTestCase::randomStepKey),
+                    instance.tierPreference());
+            case 2:
+                return new WaitForDataTierStep(instance.getKey(), instance.getNextStepKey(),
+                    randomValueOtherThan(instance.tierPreference(), () -> randomAlphaOfLength(5)));
+        }
+        throw new AssertionError();
+    }
+
+    @Override
+    protected WaitForDataTierStep copyInstance(WaitForDataTierStep instance) {
+        return new WaitForDataTierStep(instance.getKey(), instance.getNextStepKey(), instance.tierPreference());
+    }
+
+    public void testConditionMet() {
+        String notIncludedTier = randomFrom(DataTier.ALL_DATA_TIERS);
+        List<String> otherTiers = DataTier.ALL_DATA_TIERS.stream()
+            .filter(tier -> notIncludedTier.equals(tier) == false).collect(Collectors.toList());
+        List<String> includedTiers = randomSubsetOf(between(1, otherTiers.size()), otherTiers);
+        String tierPreference = String.join(",", includedTiers);
+        WaitForDataTierStep step = new WaitForDataTierStep(randomStepKey(), randomStepKey(), tierPreference);
+
+        verify (step, ClusterState.EMPTY_STATE, false, "no nodes for tiers [" + tierPreference + "] available");
+        verify(step, state(List.of(notIncludedTier)), false, "no nodes for tiers [" + tierPreference + "] available");
+        verify(step, state(includedTiers), true, null);
+        verify(step, state(List.of(DiscoveryNodeRole.DATA_ROLE.roleName())), true, null);
+    }
+
+    private void verify(WaitForDataTierStep step, ClusterState state, boolean complete, String message) {
+        ClusterStateWaitStep.Result result = step.isConditionMet(null, state);
+        assertThat(result.isComplete(), is(complete));
+        if (message != null) {
+            assertThat(Strings.toString(result.getInfomationContext()), containsString(message));
+        } else {
+            assertThat(result.getInfomationContext(), is(nullValue()));
+        }
+    }
+
+    private ClusterState state(Collection<String> roles) {
+        DiscoveryNodes.Builder builder = DiscoveryNodes.builder();
+        IntStream.range(0, between(1, 5))
+            .mapToObj(i ->
+                new DiscoveryNode(
+                    "node_" + i,
+                    UUIDs.randomBase64UUID(),
+                    buildNewFakeTransportAddress(),
+                    Map.of(),
+                    randomSubsetOf(between(1, roles.size()), roles).stream()
+                        .map(DiscoveryNodeRole::getRoleFromRoleName).collect(Collectors.toSet()),
+                    Version.CURRENT
+                )
+            ).forEach(builder::add);
+        return ClusterState.builder(ClusterName.DEFAULT).nodes(builder).build();
+    }
+}

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/cluster/metadata/MetadataMigrateToDataTiersRoutingService.java

@@ -20,6 +20,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Tuple;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.xpack.core.DataTier;
 import org.elasticsearch.xpack.core.ilm.AllocateAction;
 import org.elasticsearch.xpack.core.ilm.IndexLifecycleMetadata;
@@ -119,7 +120,8 @@ public final class MetadataMigrateToDataTiersRoutingService {
     public static Tuple<ClusterState, MigratedEntities> migrateToDataTiersRouting(ClusterState currentState,
                                                                                   @Nullable String nodeAttrName,
                                                                                   @Nullable String indexTemplateToDelete,
-                                                                                  NamedXContentRegistry xContentRegistry, Client client) {
+                                                                                  NamedXContentRegistry xContentRegistry, Client client,
+                                                                                  XPackLicenseState licenseState) {
         IndexLifecycleMetadata currentMetadata = currentState.metadata().custom(IndexLifecycleMetadata.TYPE);
         if (currentMetadata != null && currentMetadata.getOperationMode() != STOPPED) {
             throw new IllegalStateException("stop ILM before migrating to data tiers, current state is [" +
@@ -142,7 +144,7 @@ public final class MetadataMigrateToDataTiersRoutingService {
         if (Strings.isNullOrEmpty(nodeAttrName)) {
             attribute = DEFAULT_NODE_ATTRIBUTE_NAME;
         }
-        List<String> migratedPolicies = migrateIlmPolicies(mb, currentState, attribute, xContentRegistry, client);
+        List<String> migratedPolicies = migrateIlmPolicies(mb, currentState, attribute, xContentRegistry, client, licenseState);
         // Creating an intermediary cluster state view as when migrating policy we also update the cachesd phase definition stored in the
         // index metadata so the metadata.builder will probably contain an already updated view over the indices metadata which we don't
         // want to lose when migrating the indices settings
@@ -161,7 +163,7 @@ public final class MetadataMigrateToDataTiersRoutingService {
      * for each of these managed indices.
      */
     static List<String> migrateIlmPolicies(Metadata.Builder mb, ClusterState currentState, String nodeAttrName,
-                                           NamedXContentRegistry xContentRegistry, Client client) {
+                                           NamedXContentRegistry xContentRegistry, Client client, XPackLicenseState licenseState) {
         IndexLifecycleMetadata currentLifecycleMetadata = currentState.metadata().custom(IndexLifecycleMetadata.TYPE);
         if (currentLifecycleMetadata == null) {
             return Collections.emptyList();
@@ -181,7 +183,7 @@ public final class MetadataMigrateToDataTiersRoutingService {
                 assert oldPolicyMetadata != null :
                     "we must only update policies, not create new ones, but " + policyMetadataEntry.getKey() + " didn't exist";
 
-                refreshCachedPhases(mb, currentState, oldPolicyMetadata, newPolicyMetadata, xContentRegistry, client);
+                refreshCachedPhases(mb, currentState, oldPolicyMetadata, newPolicyMetadata, xContentRegistry, client, licenseState);
                 migratedPolicies.add(policyMetadataEntry.getKey());
             }
         }
@@ -198,10 +200,10 @@ public final class MetadataMigrateToDataTiersRoutingService {
      */
     static void refreshCachedPhases(Metadata.Builder mb, ClusterState currentState, LifecyclePolicyMetadata oldPolicyMetadata,
                                     LifecyclePolicyMetadata newPolicyMetadata, NamedXContentRegistry xContentRegistry,
-                                    Client client) {
+                                    Client client, XPackLicenseState licenseState) {
         // this performs a walk through the managed indices and safely updates the cached phase (ie. for the phases we did not
         // remove the allocate action)
-        updateIndicesForPolicy(mb, currentState, xContentRegistry, client, oldPolicyMetadata.getPolicy(), newPolicyMetadata);
+        updateIndicesForPolicy(mb, currentState, xContentRegistry, client, oldPolicyMetadata.getPolicy(), newPolicyMetadata, licenseState);
 
         LifecyclePolicy newLifecyclePolicy = newPolicyMetadata.getPolicy();
         List<String> migratedPhasesWithoutAllocateAction =
@@ -215,7 +217,7 @@ public final class MetadataMigrateToDataTiersRoutingService {
             // not the same as in the cached phase) so let's forcefully (and still safely :) ) refresh the cached phase for the managed
             // indices in these phases.
             refreshCachedPhaseForPhasesWithoutAllocateAction(mb, currentState, oldPolicyMetadata.getPolicy(), newPolicyMetadata,
-                migratedPhasesWithoutAllocateAction, client);
+                migratedPhasesWithoutAllocateAction, client, licenseState);
         }
     }
 
@@ -230,7 +232,8 @@ public final class MetadataMigrateToDataTiersRoutingService {
     private static void refreshCachedPhaseForPhasesWithoutAllocateAction(Metadata.Builder mb, ClusterState currentState,
                                                                          LifecyclePolicy oldPolicy,
                                                                          LifecyclePolicyMetadata newPolicyMetadata,
-                                                                         List<String> phasesWithoutAllocateAction, Client client) {
+                                                                         List<String> phasesWithoutAllocateAction, Client client,
+                                                                         XPackLicenseState licenseState) {
         String policyName = oldPolicy.getName();
         final List<IndexMetadata> managedIndices =
             StreamSupport.stream(Spliterators.spliteratorUnknownSize(currentState.metadata().indices().valuesIt(), 0), false)
@@ -249,7 +252,7 @@ public final class MetadataMigrateToDataTiersRoutingService {
                         // anymore so let's try to move the index to the next action
 
                         LifecycleExecutionState newLifecycleState = moveStateToNextActionAndUpdateCachedPhase(indexMetadata,
-                            currentExState, System::currentTimeMillis, oldPolicy, newPolicyMetadata, client);
+                            currentExState, System::currentTimeMillis, oldPolicy, newPolicyMetadata, client, licenseState);
                         if (currentExState.equals(newLifecycleState) == false) {
                             mb.put(IndexMetadata.builder(indexMetadata).putCustom(ILM_CUSTOM_METADATA_KEY, newLifecycleState.asMap()));
                         }

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

@@ -29,6 +29,7 @@ import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.index.IndexModule;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.repositories.RepositoriesService;
@@ -38,6 +39,7 @@ import org.elasticsearch.rollup.RollupV2;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.watcher.ResourceWatcherService;
+import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.action.XPackInfoFeatureAction;
 import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
 import org.elasticsearch.xpack.core.ilm.AllocateAction;
@@ -173,6 +175,10 @@ public class IndexLifecycle extends Plugin implements ActionPlugin {
             LifecycleSettings.SLM_MINIMUM_INTERVAL_SETTING);
     }
 
+    protected XPackLicenseState getLicenseState() {
+        return XPackPlugin.getSharedLicenseState();
+    }
+
     @Override
     public Collection<Object> createComponents(Client client, ClusterService clusterService, ThreadPool threadPool,
                                                ResourceWatcherService resourceWatcherService, ScriptService scriptService,
@@ -187,7 +193,7 @@ public class IndexLifecycle extends Plugin implements ActionPlugin {
         ilmHistoryStore.set(new ILMHistoryStore(settings, new OriginSettingClient(client, INDEX_LIFECYCLE_ORIGIN),
             clusterService, threadPool));
         indexLifecycleInitialisationService.set(new IndexLifecycleService(settings, client, clusterService, threadPool,
-            getClock(), System::currentTimeMillis, xContentRegistry, ilmHistoryStore.get()));
+            getClock(), System::currentTimeMillis, xContentRegistry, ilmHistoryStore.get(), getLicenseState()));
         components.add(indexLifecycleInitialisationService.get());
 
         SnapshotLifecycleTemplateRegistry templateRegistry = new SnapshotLifecycleTemplateRegistry(settings, clusterService, threadPool,

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

@@ -30,6 +30,7 @@ import org.elasticsearch.index.Index;
 import org.elasticsearch.index.shard.IndexEventListener;
 import org.elasticsearch.plugins.ShutdownAwarePlugin;
 import org.elasticsearch.shutdown.PluginShutdownService;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.XPackField;
 import org.elasticsearch.xpack.core.ilm.CheckShrinkReadyStep;
@@ -80,14 +81,14 @@ public class IndexLifecycleService
 
     public IndexLifecycleService(Settings settings, Client client, ClusterService clusterService, ThreadPool threadPool, Clock clock,
                                  LongSupplier nowSupplier, NamedXContentRegistry xContentRegistry,
-                                 ILMHistoryStore ilmHistoryStore) {
+                                 ILMHistoryStore ilmHistoryStore, XPackLicenseState licenseState) {
         super();
         this.settings = settings;
         this.clusterService = clusterService;
         this.clock = clock;
         this.nowSupplier = nowSupplier;
         this.scheduledJob = null;
-        this.policyRegistry = new PolicyStepsRegistry(xContentRegistry, client);
+        this.policyRegistry = new PolicyStepsRegistry(xContentRegistry, client, licenseState);
         this.lifecycleRunner = new IndexLifecycleRunner(policyRegistry, ilmHistoryStore, clusterService, threadPool, nowSupplier);
         this.pollInterval = LifecycleSettings.LIFECYCLE_POLL_INTERVAL_SETTING.get(settings);
         clusterService.addStateApplier(this);

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

@@ -23,6 +23,7 @@ import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.xpack.core.ilm.ErrorStep;
 import org.elasticsearch.xpack.core.ilm.IndexLifecycleMetadata;
 import org.elasticsearch.xpack.core.ilm.InitializePolicyContextStep;
@@ -277,7 +278,7 @@ public final class IndexLifecycleTransition {
                                                                                     LifecycleExecutionState existingState,
                                                                                     LongSupplier nowSupplier, LifecyclePolicy oldPolicy,
                                                                                     LifecyclePolicyMetadata newPolicyMetadata,
-                                                                                    Client client) {
+                                                                                    Client client, XPackLicenseState licenseState) {
         String policyName = LifecycleSettings.LIFECYCLE_NAME_SETTING.get(indexMetadata.getSettings());
         Step.StepKey currentStepKey = LifecycleExecutionState.getCurrentStepKey(existingState);
         if (currentStepKey == null) {
@@ -286,7 +287,7 @@ public final class IndexLifecycleTransition {
             return existingState;
         }
 
-        List<Step> policySteps = oldPolicy.toSteps(client);
+        List<Step> policySteps = oldPolicy.toSteps(client, licenseState);
         Optional<Step> currentStep = policySteps.stream()
             .filter(step -> step.getKey().equals(currentStepKey))
             .findFirst();

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

@@ -24,6 +24,7 @@ import org.elasticsearch.common.xcontent.XContentParseException;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.Index;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.ilm.ErrorStep;
 import org.elasticsearch.xpack.core.ilm.IndexLifecycleMetadata;
@@ -52,6 +53,7 @@ public class PolicyStepsRegistry {
     private static final Logger logger = LogManager.getLogger(PolicyStepsRegistry.class);
 
     private final Client client;
+    private final XPackLicenseState licenseState;
     // keeps track of existing policies in the cluster state
     private final SortedMap<String, LifecyclePolicyMetadata> lifecyclePolicyMap;
     // keeps track of what the first step in a policy is, the key is policy name
@@ -60,18 +62,19 @@ public class PolicyStepsRegistry {
     private final Map<String, Map<Step.StepKey, Step>> stepMap;
     private final NamedXContentRegistry xContentRegistry;
 
-    public PolicyStepsRegistry(NamedXContentRegistry xContentRegistry, Client client) {
-        this(new TreeMap<>(), new HashMap<>(), new HashMap<>(), xContentRegistry, client);
+    public PolicyStepsRegistry(NamedXContentRegistry xContentRegistry, Client client, XPackLicenseState licenseState) {
+        this(new TreeMap<>(), new HashMap<>(), new HashMap<>(), xContentRegistry, client, licenseState);
     }
 
     PolicyStepsRegistry(SortedMap<String, LifecyclePolicyMetadata> lifecyclePolicyMap,
                         Map<String, Step> firstStepMap, Map<String, Map<Step.StepKey, Step>> stepMap,
-                        NamedXContentRegistry xContentRegistry, Client client) {
+                        NamedXContentRegistry xContentRegistry, Client client, XPackLicenseState licenseState) {
         this.lifecyclePolicyMap = lifecyclePolicyMap;
         this.firstStepMap = firstStepMap;
         this.stepMap = stepMap;
         this.xContentRegistry = xContentRegistry;
         this.client = client;
+        this.licenseState = licenseState;
     }
 
     SortedMap<String, LifecyclePolicyMetadata> getLifecyclePolicyMap() {
@@ -124,7 +127,7 @@ public class PolicyStepsRegistry {
                 LifecyclePolicySecurityClient policyClient = new LifecyclePolicySecurityClient(client, ClientHelper.INDEX_LIFECYCLE_ORIGIN,
                         policyMetadata.getHeaders());
                 lifecyclePolicyMap.put(policyMetadata.getName(), policyMetadata);
-                List<Step> policyAsSteps = policyMetadata.getPolicy().toSteps(policyClient);
+                List<Step> policyAsSteps = policyMetadata.getPolicy().toSteps(policyClient, licenseState);
                 if (policyAsSteps.isEmpty() == false) {
                     firstStepMap.put(policyMetadata.getName(), policyAsSteps.get(0));
                     final Map<Step.StepKey, Step> stepMapForPolicy = new LinkedHashMap<>();
@@ -166,7 +169,7 @@ public class PolicyStepsRegistry {
         }
         LifecyclePolicySecurityClient policyClient = new LifecyclePolicySecurityClient(client,
             ClientHelper.INDEX_LIFECYCLE_ORIGIN, lifecyclePolicyMap.get(policy).getHeaders());
-        final List<Step> steps = policyToExecute.toSteps(policyClient);
+        final List<Step> steps = policyToExecute.toSteps(policyClient, licenseState);
         // Build a list of steps that correspond with the phase the index is currently in
         final List<Step> phaseSteps;
         if (steps == null) {

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

@@ -115,7 +115,7 @@ public class TransportPutLifecycleAction extends TransportMasterNodeAction<Reque
                         } else {
                             try {
                                 return updateIndicesForPolicy(nonRefreshedState, xContentRegistry, client,
-                                    oldPolicy.getPolicy(), lifecyclePolicyMetadata);
+                                    oldPolicy.getPolicy(), lifecyclePolicyMetadata, licenseState);
                             } catch (Exception e) {
                                 logger.warn(new ParameterizedMessage("unable to refresh indices phase JSON for updated policy [{}]",
                                     oldPolicy.getName()), e);

x-pack/plugin/ilm/src/test/java/org/elasticsearch/xpack/cluster/metadata/MetadataMigrateToDataTiersRoutingServiceTests.java

@@ -102,7 +102,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
             .build();
 
         Metadata.Builder newMetadata = Metadata.builder(state.metadata());
-        List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client);
+        List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client, null);
         assertThat(migratedPolicies.size(), is(1));
         assertThat(migratedPolicies.get(0), is(lifecycleName));
 
@@ -150,7 +150,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
                 .build();
 
             Metadata.Builder newMetadata = Metadata.builder(state.metadata());
-            List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client);
+            List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client, null);
 
             assertThat(migratedPolicies.get(0), is(lifecycleName));
             ClusterState newState = ClusterState.builder(state).metadata(newMetadata).build();
@@ -188,7 +188,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
                 .build();
 
             Metadata.Builder newMetadata = Metadata.builder(state.metadata());
-            List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client);
+            List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client, null);
 
             assertThat(migratedPolicies.get(0), is(lifecycleName));
             ClusterState newState = ClusterState.builder(state).metadata(newMetadata).build();
@@ -230,7 +230,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
                 .build();
 
             Metadata.Builder newMetadata = Metadata.builder(state.metadata());
-            List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client);
+            List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client, null);
 
             assertThat(migratedPolicies.get(0), is(lifecycleName));
             ClusterState newState = ClusterState.builder(state).metadata(newMetadata).build();
@@ -270,7 +270,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
                 .build();
 
             Metadata.Builder newMetadata = Metadata.builder(state.metadata());
-            List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client);
+            List<String> migratedPolicies = migrateIlmPolicies(newMetadata, state, "data", REGISTRY, client, null);
 
             assertThat(migratedPolicies.get(0), is(lifecycleName));
             ClusterState newState = ClusterState.builder(state).metadata(newMetadata).build();
@@ -551,7 +551,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
 
         {
             Tuple<ClusterState, MigratedEntities> migratedEntitiesTuple =
-                migrateToDataTiersRouting(state, "data", "catch-all", REGISTRY, client);
+                migrateToDataTiersRouting(state, "data", "catch-all", REGISTRY, client, null);
 
             MigratedEntities migratedEntities = migratedEntitiesTuple.v2();
             assertThat(migratedEntities.removedIndexTemplateName, is("catch-all"));
@@ -569,7 +569,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
         {
             // let's test a null template name to make sure nothing is removed
             Tuple<ClusterState, MigratedEntities> migratedEntitiesTuple =
-                migrateToDataTiersRouting(state, "data", null, REGISTRY, client);
+                migrateToDataTiersRouting(state, "data", null, REGISTRY, client, null);
 
             MigratedEntities migratedEntities = migratedEntitiesTuple.v2();
             assertThat(migratedEntities.removedIndexTemplateName, nullValue());
@@ -587,7 +587,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
         {
             // let's test a null node attribute parameter defaults to "data"
             Tuple<ClusterState, MigratedEntities> migratedEntitiesTuple =
-                migrateToDataTiersRouting(state, null, null, REGISTRY, client);
+                migrateToDataTiersRouting(state, null, null, REGISTRY, client, null);
 
             MigratedEntities migratedEntities = migratedEntitiesTuple.v2();
             assertThat(migratedEntities.migratedPolicies.size(), is(1));
@@ -607,7 +607,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
                     Map.of(), OperationMode.RUNNING)))
                 .build();
             IllegalStateException illegalStateException = expectThrows(IllegalStateException.class,
-                () -> migrateToDataTiersRouting(ilmRunningState, "data", "catch-all", REGISTRY, client));
+                () -> migrateToDataTiersRouting(ilmRunningState, "data", "catch-all", REGISTRY, client, null));
             assertThat(illegalStateException.getMessage(), is("stop ILM before migrating to data tiers, current state is [RUNNING]"));
         }
 
@@ -617,7 +617,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
                     Map.of(), OperationMode.STOPPING)))
                 .build();
             IllegalStateException illegalStateException = expectThrows(IllegalStateException.class,
-                () -> migrateToDataTiersRouting(ilmStoppingState, "data", "catch-all", REGISTRY, client));
+                () -> migrateToDataTiersRouting(ilmStoppingState, "data", "catch-all", REGISTRY, client, null));
             assertThat(illegalStateException.getMessage(), is("stop ILM before migrating to data tiers, current state is [STOPPING]"));
         }
 
@@ -627,7 +627,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
                     Map.of(), OperationMode.STOPPED)))
                 .build();
             Tuple<ClusterState, MigratedEntities> migratedState = migrateToDataTiersRouting(ilmStoppedState, "data", "catch-all",
-                REGISTRY, client);
+                REGISTRY, client, null);
             assertThat(migratedState.v2().migratedIndices, empty());
             assertThat(migratedState.v2().migratedPolicies, empty());
             assertThat(migratedState.v2().removedIndexTemplateName, nullValue());
@@ -645,7 +645,7 @@ public class MetadataMigrateToDataTiersRoutingServiceTests extends ESTestCase {
             .put(composableTemplateName, composableIndexTemplate).build())
             .build();
         Tuple<ClusterState, MigratedEntities> migratedEntitiesTuple =
-            migrateToDataTiersRouting(clusterState, "data", composableTemplateName, REGISTRY, client);
+            migrateToDataTiersRouting(clusterState, "data", composableTemplateName, REGISTRY, client, null);
         assertThat(migratedEntitiesTuple.v2().removedIndexTemplateName, nullValue());
         assertThat(migratedEntitiesTuple.v1().metadata().templatesV2().get(composableTemplateName), is(composableIndexTemplate));
     }

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

@@ -107,7 +107,7 @@ public class ExecuteStepsUpdateTaskTests extends ESTestCase {
             randomNonNegativeLong(), randomNonNegativeLong()));
         policyMap.put(invalidPolicyName, new LifecyclePolicyMetadata(invalidPolicy, Collections.emptyMap(),
             randomNonNegativeLong(), randomNonNegativeLong()));
-        policyStepsRegistry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY, client);
+        policyStepsRegistry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY, client, null);
 
         indexName = randomAlphaOfLength(5);
         lifecycleMetadata = new IndexLifecycleMetadata(policyMap, OperationMode.RUNNING);

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

@@ -175,7 +175,7 @@ public class IndexLifecycleRunnerTests extends ESTestCase {
         PhaseExecutionInfo phaseExecutionInfo = new PhaseExecutionInfo(policy.getName(), phase, 1, randomNonNegativeLong());
         String phaseJson = Strings.toString(phaseExecutionInfo);
         LifecycleAction action = randomValueOtherThan(new MigrateAction(false), () -> randomFrom(phase.getActions().values()));
-        Step step = randomFrom(action.toSteps(new NoOpClient(threadPool), phaseName, null));
+        Step step = randomFrom(action.toSteps(new NoOpClient(threadPool), phaseName, null, null));
         StepKey stepKey = step.getKey();
 
         PolicyStepsRegistry stepRegistry = createOneStepPolicyStepRegistry(policyName, step);
@@ -765,7 +765,7 @@ public class IndexLifecycleRunnerTests extends ESTestCase {
     public void testRunPolicyThatDoesntExist() {
         String policyName = "cluster_state_action_policy";
         ClusterService clusterService = mock(ClusterService.class);
-        IndexLifecycleRunner runner = new IndexLifecycleRunner(new PolicyStepsRegistry(NamedXContentRegistry.EMPTY, null),
+        IndexLifecycleRunner runner = new IndexLifecycleRunner(new PolicyStepsRegistry(NamedXContentRegistry.EMPTY, null, null),
             historyStore, clusterService, threadPool, () -> 0L);
         IndexMetadata indexMetadata = IndexMetadata.builder("my_index").settings(settings(Version.CURRENT))
             .numberOfShards(randomIntBetween(1, 5)).numberOfReplicas(randomIntBetween(0, 5)).build();
@@ -805,7 +805,7 @@ public class IndexLifecycleRunnerTests extends ESTestCase {
         PhaseExecutionInfo pei = new PhaseExecutionInfo(policy.getName(), phase, 1, randomNonNegativeLong());
         String phaseJson = Strings.toString(pei);
         LifecycleAction action = randomValueOtherThan(new MigrateAction(false), () -> randomFrom(phase.getActions().values()));
-        Step step = randomFrom(action.toSteps(client, phaseName, MOCK_STEP_KEY));
+        Step step = randomFrom(action.toSteps(client, phaseName, MOCK_STEP_KEY, null));
         Settings indexSettings = Settings.builder()
             .put("index.number_of_shards", 1)
             .put("index.number_of_replicas", 0)
@@ -823,7 +823,7 @@ public class IndexLifecycleRunnerTests extends ESTestCase {
             .build();
         SortedMap<String, LifecyclePolicyMetadata> metas = new TreeMap<>();
         metas.put(policyName, policyMetadata);
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(metas, firstStepMap, stepMap, REGISTRY, client);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(metas, firstStepMap, stepMap, REGISTRY, client, null);
 
         // First step is retrieved because there are no settings for the index
         IndexMetadata indexMetadataWithNoKey = IndexMetadata.builder(index.getName())
@@ -850,7 +850,7 @@ public class IndexLifecycleRunnerTests extends ESTestCase {
         Map<StepKey, Step> policySteps = Collections.singletonMap(step.getKey(), step);
         Map<String, Map<StepKey, Step>> stepMap = Collections.singletonMap(policyName, policySteps);
         PolicyStepsRegistry policyStepsRegistry = new PolicyStepsRegistry(lifecyclePolicyMap, firstStepMap,
-            stepMap, NamedXContentRegistry.EMPTY, null);
+            stepMap, NamedXContentRegistry.EMPTY, null, null);
         ClusterService clusterService = mock(ClusterService.class);
         final AtomicLong now = new AtomicLong(5);
         IndexLifecycleRunner runner = new IndexLifecycleRunner(policyStepsRegistry, historyStore,
@@ -1211,7 +1211,7 @@ public class IndexLifecycleRunnerTests extends ESTestCase {
 
         MockPolicyStepsRegistry(SortedMap<String, LifecyclePolicyMetadata> lifecyclePolicyMap, Map<String, Step> firstStepMap,
                                 Map<String, Map<StepKey, Step>> stepMap, NamedXContentRegistry xContentRegistry, Client client) {
-            super(lifecyclePolicyMap, firstStepMap, stepMap, xContentRegistry, client);
+            super(lifecyclePolicyMap, firstStepMap, stepMap, xContentRegistry, client, null);
         }
 
         public void setResolver(BiFunction<IndexMetadata, StepKey, Step> fn) {

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

@@ -124,7 +124,7 @@ public class IndexLifecycleServiceTests extends ESTestCase {
 
         threadPool = new TestThreadPool("test");
         indexLifecycleService = new IndexLifecycleService(Settings.EMPTY, client, clusterService, threadPool,
-            clock, () -> now, null, null);
+            clock, () -> now, null, null, null);
         Mockito.verify(clusterService).addListener(indexLifecycleService);
         Mockito.verify(clusterService).addStateApplier(indexLifecycleService);
     }
@@ -461,7 +461,7 @@ public class IndexLifecycleServiceTests extends ESTestCase {
 
     public void testClusterChangedWaitsForTheStateToBeRecovered() {
         IndexLifecycleService ilmService = new IndexLifecycleService(Settings.EMPTY, mock(Client.class), clusterService, threadPool,
-            systemUTC(), () -> now, null, null) {
+            systemUTC(), () -> now, null, null, null) {
 
             @Override
             void onMaster(ClusterState clusterState) {

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

@@ -831,7 +831,7 @@ public class IndexLifecycleTransitionTests extends ESTestCase {
             try (Client client = new NoOpClient(getTestName())) {
                 LifecycleExecutionState newState = moveStateToNextActionAndUpdateCachedPhase(meta,
                     LifecycleExecutionState.fromIndexMetadata(meta), System::currentTimeMillis, currentPolicy, updatedPolicyMetadata,
-                    client);
+                    client, null);
 
                 Step.StepKey hotPhaseCompleteStepKey = PhaseCompleteStep.finalStep("hot").getKey();
                 assertThat(newState.getAction(), is(hotPhaseCompleteStepKey.getAction()));
@@ -855,7 +855,7 @@ public class IndexLifecycleTransitionTests extends ESTestCase {
             try (Client client = new NoOpClient(getTestName())) {
                 LifecycleExecutionState newState = moveStateToNextActionAndUpdateCachedPhase(meta,
                     LifecycleExecutionState.fromIndexMetadata(meta), System::currentTimeMillis, currentPolicy, updatedPolicyMetadata,
-                    client);
+                    client, null);
 
                 Step.StepKey hotPhaseCompleteStepKey = PhaseCompleteStep.finalStep("hot").getKey();
                 // the state was still moved into the next action, even if the updated policy still contained the action the index was

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

@@ -65,7 +65,7 @@ public class MoveToNextStepUpdateTaskTests extends ESTestCase {
 
     public void testExecuteSuccessfullyMoved() {
         long now = randomNonNegativeLong();
-        List<Step> steps = lifecyclePolicy.toSteps(null);
+        List<Step> steps = lifecyclePolicy.toSteps(null, null);
         StepKey currentStepKey = steps.get(0).getKey();
         StepKey nextStepKey = steps.get(0).getNextStepKey();
 
@@ -109,7 +109,7 @@ public class MoveToNextStepUpdateTaskTests extends ESTestCase {
 
     public void testExecuteSuccessfulMoveWithInvalidNextStep() {
         long now = randomNonNegativeLong();
-        List<Step> steps = lifecyclePolicy.toSteps(null);
+        List<Step> steps = lifecyclePolicy.toSteps(null, null);
         StepKey currentStepKey = steps.get(0).getKey();
         StepKey invalidNextStep = new StepKey("next-invalid", "next-invalid", "next-invalid");
 
@@ -152,7 +152,7 @@ public class MoveToNextStepUpdateTaskTests extends ESTestCase {
     private static class AlwaysExistingStepRegistry extends PolicyStepsRegistry {
 
         AlwaysExistingStepRegistry() {
-            super(new NamedXContentRegistry(Collections.emptyList()), null);
+            super(new NamedXContentRegistry(Collections.emptyList()), null, null);
         }
 
         @Override

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

@@ -70,7 +70,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
         String policyName = randomAlphaOfLengthBetween(2, 10);
         Step expectedFirstStep = new MockStep(MOCK_STEP_KEY, null);
         Map<String, Step> firstStepMap = Collections.singletonMap(policyName, expectedFirstStep);
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(null, firstStepMap, null, NamedXContentRegistry.EMPTY, null);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(null, firstStepMap, null, NamedXContentRegistry.EMPTY, null, null);
         Step actualFirstStep = registry.getFirstStep(policyName);
         assertThat(actualFirstStep, sameInstance(expectedFirstStep));
     }
@@ -79,7 +79,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
         String policyName = randomAlphaOfLengthBetween(2, 10);
         Step expectedFirstStep = new MockStep(MOCK_STEP_KEY, null);
         Map<String, Step> firstStepMap = Collections.singletonMap(policyName, expectedFirstStep);
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(null, firstStepMap, null, NamedXContentRegistry.EMPTY, null);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(null, firstStepMap, null, NamedXContentRegistry.EMPTY, null, null);
         Step actualFirstStep = registry.getFirstStep(policyName + "unknown");
         assertNull(actualFirstStep);
     }
@@ -94,7 +94,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
         PhaseExecutionInfo pei = new PhaseExecutionInfo(policy.getName(), phase, 1, randomNonNegativeLong());
         String phaseJson = Strings.toString(pei);
         LifecycleAction action = randomValueOtherThan(new MigrateAction(false), () -> randomFrom(phase.getActions().values()));
-        Step step = randomFrom(action.toSteps(client, phaseName, MOCK_STEP_KEY));
+        Step step = randomFrom(action.toSteps(client, phaseName, MOCK_STEP_KEY, null));
         LifecycleExecutionState.Builder lifecycleState = LifecycleExecutionState.builder();
         lifecycleState.setPhaseDefinition(phaseJson);
         IndexMetadata indexMetadata = IndexMetadata.builder("test")
@@ -108,7 +108,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
             .build();
         SortedMap<String, LifecyclePolicyMetadata> metas = new TreeMap<>();
         metas.put("policy", policyMetadata);
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(metas, null, null, REGISTRY, client);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(metas, null, null, REGISTRY, client, null);
         Step actualStep = registry.getStep(indexMetadata, step.getKey());
         assertThat(actualStep.getKey(), equalTo(step.getKey()));
     }
@@ -118,13 +118,13 @@ public class PolicyStepsRegistryTests extends ESTestCase {
         Step expectedStep = new ErrorStep(errorStepKey);
         Index index = new Index("test", "uuid");
         Map<Index, List<Step>> indexSteps = Collections.singletonMap(index, Collections.singletonList(expectedStep));
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(null, null, null, NamedXContentRegistry.EMPTY, null);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(null, null, null, NamedXContentRegistry.EMPTY, null, null);
         Step actualStep = registry.getStep(emptyMetadata(index), errorStepKey);
         assertThat(actualStep, equalTo(expectedStep));
     }
 
     public void testGetStepUnknownPolicy() {
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(null, null, null, NamedXContentRegistry.EMPTY, null);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(null, null, null, NamedXContentRegistry.EMPTY, null, null);
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
             () -> registry.getStep(emptyMetadata(new Index("test", "uuid")), MOCK_STEP_KEY));
         assertThat(e.getMessage(),
@@ -147,7 +147,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
             .build();
         SortedMap<String, LifecyclePolicyMetadata> metas = new TreeMap<>();
         metas.put("policy", policyMetadata);
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(metas, null, null, REGISTRY, client);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(metas, null, null, REGISTRY, client, null);
         Step step = registry.getStep(indexMetadata, InitializePolicyContextStep.KEY);
         assertNotNull(step);
     }
@@ -162,7 +162,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
         PhaseExecutionInfo pei = new PhaseExecutionInfo(policy.getName(), phase, 1, randomNonNegativeLong());
         String phaseJson = Strings.toString(pei);
         LifecycleAction action = randomValueOtherThan(new MigrateAction(false), () -> randomFrom(phase.getActions().values()));
-        Step step = randomFrom(action.toSteps(client, phaseName, MOCK_STEP_KEY));
+        Step step = randomFrom(action.toSteps(client, phaseName, MOCK_STEP_KEY, null));
         LifecycleExecutionState.Builder lifecycleState = LifecycleExecutionState.builder();
         lifecycleState.setPhaseDefinition(phaseJson);
         IndexMetadata indexMetadata = IndexMetadata.builder("test")
@@ -176,7 +176,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
             .build();
         SortedMap<String, LifecyclePolicyMetadata> metas = new TreeMap<>();
         metas.put("policy", policyMetadata);
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(metas, null, null, REGISTRY, client);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(metas, null, null, REGISTRY, client, null);
         Step actualStep = registry.getStep(indexMetadata,
             new Step.StepKey(step.getKey().getPhase(), step.getKey().getAction(), step.getKey().getName() + "-bad"));
         assertNull(actualStep);
@@ -189,7 +189,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
         String policyName = randomAlphaOfLength(5);
         LifecyclePolicy newPolicy = LifecyclePolicyTests.randomTestLifecyclePolicy(policyName);
         logger.info("--> policy: {}", newPolicy);
-        List<Step> policySteps = newPolicy.toSteps(client);
+        List<Step> policySteps = newPolicy.toSteps(client, null);
         Map<String, String> headers = new HashMap<>();
         if (randomBoolean()) {
             headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
@@ -229,7 +229,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
             .build();
 
         // start with empty registry
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY, client);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY, client, null);
 
         // add new policy
         registry.update(currentState);
@@ -303,7 +303,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
             .metadata(metadata)
             .nodes(DiscoveryNodes.builder().localNodeId(nodeId).masterNodeId(nodeId).add(masterNode).build())
             .build();
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY, client);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(NamedXContentRegistry.EMPTY, client, null);
         // add new policy
         registry.update(currentState);
 
@@ -339,7 +339,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
         LifecyclePolicy updatedPolicy = new LifecyclePolicy(policyName, phases);
         logger.info("--> policy: {}", newPolicy);
         logger.info("--> updated policy: {}", updatedPolicy);
-        List<Step> policySteps = newPolicy.toSteps(client);
+        List<Step> policySteps = newPolicy.toSteps(client, null);
         Map<String, String> headers = new HashMap<>();
         if (randomBoolean()) {
             headers.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
@@ -380,7 +380,7 @@ public class PolicyStepsRegistryTests extends ESTestCase {
             .build();
 
         // start with empty registry
-        PolicyStepsRegistry registry = new PolicyStepsRegistry(REGISTRY, client);
+        PolicyStepsRegistry registry = new PolicyStepsRegistry(REGISTRY, client, null);
 
         // add new policy
         registry.update(currentState);

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/FrozenSearchableSnapshotsIntegTests.java

@@ -60,7 +60,6 @@ import static org.elasticsearch.index.query.QueryBuilders.matchQuery;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotsConstants.SNAPSHOT_DIRECTORY_FACTORY_KEY;
 import static org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotsConstants.SNAPSHOT_RECOVERY_STATE_FACTORY_KEY;
-import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.getDataTiersPreference;
 import static org.hamcrest.Matchers.arrayWithSize;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
@@ -174,7 +173,7 @@ public class FrozenSearchableSnapshotsIntegTests extends BaseFrozenSearchableSna
             indexCheckOnStartup = "false";
         }
         final String expectedDataTiersPreference;
-        expectedDataTiersPreference = getDataTiersPreference(MountSearchableSnapshotRequest.Storage.SHARED_CACHE);
+        expectedDataTiersPreference = MountSearchableSnapshotRequest.Storage.SHARED_CACHE.defaultDataTiersPreference();
 
         indexSettingsBuilder.put(Store.INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING.getKey(), TimeValue.ZERO);
         final AtomicBoolean statsWatcherRunning = new AtomicBoolean(true);

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsIntegTests.java

@@ -81,7 +81,6 @@ import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcke
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotsConstants.SNAPSHOT_DIRECTORY_FACTORY_KEY;
 import static org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotsConstants.SNAPSHOT_RECOVERY_STATE_FACTORY_KEY;
-import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.getDataTiersPreference;
 import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -206,7 +205,7 @@ public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegT
             );
             indexSettingsBuilder.put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, expectedDataTiersPreference);
         } else {
-            expectedDataTiersPreference = getDataTiersPreference(MountSearchableSnapshotRequest.Storage.FULL_COPY);
+            expectedDataTiersPreference = MountSearchableSnapshotRequest.Storage.FULL_COPY.defaultDataTiersPreference();
         }
 
         final MountSearchableSnapshotRequest req = new MountSearchableSnapshotRequest(

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshots.java

@@ -78,7 +78,6 @@ import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.action.XPackInfoFeatureAction;
 import org.elasticsearch.xpack.core.action.XPackUsageFeatureAction;
 import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotAction;
-import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
 import org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotsConstants;
 import org.elasticsearch.xpack.searchablesnapshots.action.ClearSearchableSnapshotsCacheAction;
 import org.elasticsearch.xpack.searchablesnapshots.action.SearchableSnapshotsInfoTransportAction;
@@ -241,22 +240,6 @@ public class SearchableSnapshots extends Plugin implements IndexStorePlugin, Eng
      */
     public static final String DATA_TIERS_CACHE_INDEX_PREFERENCE = String.join(",", DataTier.DATA_CONTENT, DataTier.DATA_HOT);
 
-    /**
-     * Returns the preference for new searchable snapshot indices. When
-     * performing a full mount the preference is cold - warm - hot. When
-     * performing a partial mount the preference is only frozen
-     */
-    public static String getDataTiersPreference(MountSearchableSnapshotRequest.Storage type) {
-        switch (type) {
-            case FULL_COPY:
-                return String.join(",", DataTier.DATA_COLD, DataTier.DATA_WARM, DataTier.DATA_HOT);
-            case SHARED_CACHE:
-                return DataTier.DATA_FROZEN;
-            default:
-                throw new IllegalArgumentException("unknown searchable snapshot type [" + type + "]");
-        }
-    }
-
     private volatile Supplier<RepositoriesService> repositoriesServiceSupplier;
     private final SetOnce<BlobStoreCacheService> blobStoreCacheService = new SetOnce<>();
     private final SetOnce<CacheService> cacheService = new SetOnce<>();

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/action/TransportMountSearchableSnapshotAction.java

@@ -59,7 +59,6 @@ import java.util.Set;
 import static org.elasticsearch.index.IndexModule.INDEX_RECOVERY_TYPE_SETTING;
 import static org.elasticsearch.index.IndexModule.INDEX_STORE_TYPE_SETTING;
 import static org.elasticsearch.xpack.core.searchablesnapshots.SearchableSnapshotsConstants.isSearchableSnapshotStore;
-import static org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshots.getDataTiersPreference;
 
 /**
  * Action that mounts a snapshot as a searchable snapshot, by converting the mount request into a restore request with specific settings
@@ -233,7 +232,7 @@ public class TransportMountSearchableSnapshotAction extends TransportMasterNodeA
                 .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0) // can be overridden
                 .put(IndexMetadata.SETTING_AUTO_EXPAND_REPLICAS, false) // can be overridden
                 .put(IndexSettings.INDEX_CHECK_ON_STARTUP.getKey(), false) // can be overridden
-                .put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, getDataTiersPreference(request.storage()))
+                .put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, request.storage().defaultDataTiersPreference())
                 .put(request.indexSettings())
                 .put(
                     buildIndexSettings(