Move SLM history to data stream (#63038)

This change converts SLM history index to data stream.
Consequence of this is name change as name of data stream can't start with .. New name for SLM history data stream will be slm-history-4 (retaining version in the name).

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/slm/history/SnapshotHistoryStore.java

@@ -9,14 +9,11 @@ package org.elasticsearch.xpack.core.slm.history;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.elasticsearch.ResourceAlreadyExistsException;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.admin.indices.alias.Alias;
-import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.client.Client;
-import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.metadata.IndexAbstraction;
+import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.ToXContent;
@@ -27,6 +24,7 @@ import java.io.IOException;
 
 import static org.elasticsearch.xpack.core.ilm.LifecycleSettings.SLM_HISTORY_INDEX_ENABLED_SETTING;
 import static org.elasticsearch.xpack.core.slm.history.SnapshotLifecycleTemplateRegistry.INDEX_TEMPLATE_VERSION;
+import static org.elasticsearch.xpack.core.slm.history.SnapshotLifecycleTemplateRegistry.SLM_TEMPLATE_NAME;
 
 /**
  * Records Snapshot Lifecycle Management actions as represented by {@link SnapshotHistoryItem} into an index
@@ -35,8 +33,7 @@ import static org.elasticsearch.xpack.core.slm.history.SnapshotLifecycleTemplate
 public class SnapshotHistoryStore {
     private static final Logger logger = LogManager.getLogger(SnapshotHistoryStore.class);
 
-    public static final String SLM_HISTORY_INDEX_PREFIX = ".slm-history-" + INDEX_TEMPLATE_VERSION + "-";
-    public static final String SLM_HISTORY_ALIAS = ".slm-history-" + INDEX_TEMPLATE_VERSION;
+    public static final String SLM_HISTORY_DATA_STREAM = "slm-history-" + INDEX_TEMPLATE_VERSION;
 
     private final Client client;
     private final ClusterService clusterService;
@@ -59,85 +56,29 @@ public class SnapshotHistoryStore {
                 SLM_HISTORY_INDEX_ENABLED_SETTING.getKey(), item);
             return;
         }
-        logger.trace("about to index snapshot history item in index [{}]: [{}]", SLM_HISTORY_ALIAS, item);
-        ensureHistoryIndex(client, clusterService.state(), ActionListener.wrap(createdIndex -> {
-            try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
-                item.toXContent(builder, ToXContent.EMPTY_PARAMS);
-                IndexRequest request = new IndexRequest(SLM_HISTORY_ALIAS)
-                    .source(builder);
-                client.index(request, ActionListener.wrap(indexResponse -> {
-                    logger.debug("successfully indexed snapshot history item with id [{}] in index [{}]: [{}]",
-                        indexResponse.getId(), SLM_HISTORY_ALIAS, item);
-                }, exception -> {
-                    logger.error(new ParameterizedMessage("failed to index snapshot history item in index [{}]: [{}]",
-                        SLM_HISTORY_ALIAS, item), exception);
-                }));
-            } catch (IOException exception) {
-                logger.error(new ParameterizedMessage("failed to index snapshot history item in index [{}]: [{}]",
-                    SLM_HISTORY_ALIAS, item), exception);
-            }
-        }, ex -> logger.error(new ParameterizedMessage("failed to ensure SLM history index exists, not indexing history item [{}]",
-            item), ex)));
-    }
-
-    /**
-     * Checks if the SLM history index exists, and if not, creates it.
-     *
-     * @param client  The client to use to create the index if needed
-     * @param state   The current cluster state, to determine if the alias exists
-     * @param andThen Called after the index has been created. `onResponse` called with `true` if the index was created,
-     *                `false` if it already existed.
-     */
-    static void ensureHistoryIndex(Client client, ClusterState state, ActionListener<Boolean> andThen) {
-        final String initialHistoryIndexName = SLM_HISTORY_INDEX_PREFIX + "000001";
-        final IndexAbstraction slmHistory = state.metadata().getIndicesLookup().get(SLM_HISTORY_ALIAS);
-        final IndexAbstraction initialHistoryIndex = state.metadata().getIndicesLookup().get(initialHistoryIndexName);
-
-        if (slmHistory == null && initialHistoryIndex == null) {
-            // No alias or index exists with the expected names, so create the index with appropriate alias
-            client.admin().indices().prepareCreate(initialHistoryIndexName)
-                .setWaitForActiveShards(1)
-                .addAlias(new Alias(SLM_HISTORY_ALIAS)
-                    .writeIndex(true)
-                    .isHidden(true))
-                .execute(new ActionListener<>() {
-                    @Override
-                    public void onResponse(CreateIndexResponse response) {
-                        andThen.onResponse(true);
-                    }
-
-                    @Override
-                    public void onFailure(Exception e) {
-                        if (e instanceof ResourceAlreadyExistsException) {
-                            // The index didn't exist before we made the call, there was probably a race - just ignore this
-                            logger.debug("index [{}] was created after checking for its existence, likely due to a concurrent call",
-                                initialHistoryIndexName);
-                            andThen.onResponse(false);
-                        } else {
-                            andThen.onFailure(e);
-                        }
-                    }
-                });
-        } else if (slmHistory == null) {
-            // alias does not exist but initial index does, something is broken
-            andThen.onFailure(new IllegalStateException("SLM history index [" + initialHistoryIndexName +
-                "] already exists but does not have alias [" + SLM_HISTORY_ALIAS + "]"));
-        } else if (slmHistory.getType() == IndexAbstraction.Type.ALIAS) {
-            if (slmHistory.getWriteIndex() != null) {
-                // The alias exists and has a write index, so we're good
-                andThen.onResponse(false);
-            } else {
-                // The alias does not have a write index, so we can't index into it
-                andThen.onFailure(new IllegalStateException("SLM history alias [" + SLM_HISTORY_ALIAS + "does not have a write index"));
-            }
-        } else if (slmHistory.getType() != IndexAbstraction.Type.ALIAS) {
-            // This is not an alias, error out
-            andThen.onFailure(new IllegalStateException("SLM history alias [" + SLM_HISTORY_ALIAS +
-                "] already exists as " + slmHistory.getType().getDisplayName()));
-        } else {
-            logger.error("unexpected IndexOrAlias for [{}]: [{}]", SLM_HISTORY_ALIAS, slmHistory);
-            // (slmHistory.isAlias() == true) but (slmHistory instanceof Alias == false)?
-            assert false : SLM_HISTORY_ALIAS + " cannot be both an alias and not an alias simultaneously";
+        logger.trace("about to index snapshot history item in data stream [{}]: [{}]", SLM_HISTORY_DATA_STREAM, item);
+        Metadata metadata = clusterService.state().getMetadata();
+        if (metadata.dataStreams().containsKey(SLM_HISTORY_DATA_STREAM) == false &&
+            metadata.templatesV2().containsKey(SLM_TEMPLATE_NAME) == false) {
+            logger.error(new ParameterizedMessage("failed to index snapshot history item, data stream [{}] and template [{}] don't exist",
+                SLM_HISTORY_DATA_STREAM, SLM_TEMPLATE_NAME));
+            return;
+        }
+        try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
+            item.toXContent(builder, ToXContent.EMPTY_PARAMS);
+            IndexRequest request = new IndexRequest(SLM_HISTORY_DATA_STREAM)
+                .opType(DocWriteRequest.OpType.CREATE)
+                .source(builder);
+            client.index(request, ActionListener.wrap(indexResponse -> {
+                logger.debug("successfully indexed snapshot history item with id [{}] in data stream [{}]: [{}]",
+                    indexResponse.getId(), SLM_HISTORY_DATA_STREAM, item);
+            }, exception -> {
+                logger.error(new ParameterizedMessage("failed to index snapshot history item in data stream [{}]: [{}]",
+                    SLM_HISTORY_DATA_STREAM, item), exception);
+            }));
+        } catch (IOException exception) {
+            logger.error(new ParameterizedMessage("failed to index snapshot history item in data stream [{}]: [{}]",
+                SLM_HISTORY_DATA_STREAM, item), exception);
         }
     }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/slm/history/SnapshotLifecycleTemplateRegistry.java

@@ -37,7 +37,8 @@ public class SnapshotLifecycleTemplateRegistry extends IndexTemplateRegistry {
     // version 1: initial
     // version 2: converted to hidden index
     // version 3: templates moved to composable templates
-    public static final int INDEX_TEMPLATE_VERSION = 3;
+    // version 4:converted data stream
+    public static final int INDEX_TEMPLATE_VERSION = 4;
 
     public static final String SLM_TEMPLATE_VERSION_VARIABLE = "xpack.slm.template.version";
     public static final String SLM_TEMPLATE_NAME = ".slm-history";

x-pack/plugin/core/src/main/resources/slm-history.json

@@ -1,17 +1,15 @@
 {
   "index_patterns": [
-    ".slm-history-${xpack.slm.template.version}*"
+    "slm-history-${xpack.slm.template.version}*"
   ],
   "priority": 2147483647,
+  "data_stream": {},
   "template": {
     "settings": {
       "index.number_of_shards": 1,
       "index.number_of_replicas": 0,
       "index.auto_expand_replicas": "0-1",
-      "index.lifecycle.name": "slm-history-ilm-policy",
-      "index.lifecycle.rollover_alias": ".slm-history-${xpack.slm.template.version}",
-      "index.hidden": true,
-      "index.format": 1
+      "index.lifecycle.name": "slm-history-ilm-policy"
     },
     "mappings": {
       "dynamic": false,

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/slm/history/SnapshotHistoryStoreTests.java

@@ -6,23 +6,20 @@
 
 package org.elasticsearch.xpack.core.slm.history;
 
-import org.elasticsearch.ResourceAlreadyExistsException;
-import org.elasticsearch.Version;
-import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.LatchedActionListener;
 import org.elasticsearch.action.admin.indices.create.CreateIndexAction;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.index.IndexAction;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.metadata.AliasMetadata;
-import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.DeprecationHandler;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESTestCase;
@@ -30,22 +27,18 @@ import org.elasticsearch.threadpool.TestThreadPool;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.slm.SnapshotLifecyclePolicy;
 import org.junit.After;
-import org.junit.Assert;
 import org.junit.Before;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.awaitLatch;
 import static org.elasticsearch.xpack.core.ilm.GenerateSnapshotNameStep.generateSnapshotName;
 import static org.elasticsearch.xpack.core.ilm.LifecycleSettings.SLM_HISTORY_INDEX_ENABLED_SETTING;
-import static org.elasticsearch.xpack.core.slm.history.SnapshotHistoryStore.SLM_HISTORY_ALIAS;
-import static org.elasticsearch.xpack.core.slm.history.SnapshotHistoryStore.SLM_HISTORY_INDEX_PREFIX;
+import static org.elasticsearch.xpack.core.slm.history.SnapshotHistoryStore.SLM_HISTORY_DATA_STREAM;
+import static org.elasticsearch.xpack.core.slm.history.SnapshotLifecycleTemplateRegistry.TEMPLATE_SLM_HISTORY;
 import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.core.IsEqual.equalTo;
 
@@ -54,19 +47,29 @@ public class SnapshotHistoryStoreTests extends ESTestCase {
     private ThreadPool threadPool;
     private SnapshotLifecycleTemplateRegistryTests.VerifyingClient client;
     private SnapshotHistoryStore historyStore;
+    private ClusterService clusterService;
 
     @Before
-    public void setup() {
+    public void setup() throws IOException {
         threadPool = new TestThreadPool(this.getClass().getName());
         client = new SnapshotLifecycleTemplateRegistryTests.VerifyingClient(threadPool);
-        ClusterService clusterService = ClusterServiceUtils.createClusterService(threadPool);
+        clusterService = ClusterServiceUtils.createClusterService(threadPool);
+        ComposableIndexTemplate template =
+            ComposableIndexTemplate.parse(JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
+                DeprecationHandler.THROW_UNSUPPORTED_OPERATION, TEMPLATE_SLM_HISTORY.loadBytes()));
+        ClusterState state = clusterService.state();
+        Metadata.Builder metadataBuilder =
+            Metadata.builder(state.getMetadata()).indexTemplates(Map.of(TEMPLATE_SLM_HISTORY.getTemplateName(), template));
+        ClusterServiceUtils.setState(clusterService, ClusterState.builder(state).metadata(metadataBuilder).build());
         historyStore = new SnapshotHistoryStore(Settings.EMPTY, client, clusterService);
+        clusterService.stop();
     }
 
     @After
     @Override
     public void tearDown() throws Exception {
         super.tearDown();
+        clusterService.stop();
         threadPool.shutdownNow();
     }
 
@@ -97,14 +100,11 @@ public class SnapshotHistoryStoreTests extends ESTestCase {
 
             AtomicInteger calledTimes = new AtomicInteger(0);
             client.setVerifier((action, request, listener) -> {
-                if (action instanceof CreateIndexAction && request instanceof CreateIndexRequest) {
-                    return new CreateIndexResponse(true, true, ((CreateIndexRequest) request).index());
-                }
                 calledTimes.incrementAndGet();
                 assertThat(action, instanceOf(IndexAction.class));
                 assertThat(request, instanceOf(IndexRequest.class));
                 IndexRequest indexRequest = (IndexRequest) request;
-                assertEquals(SLM_HISTORY_ALIAS, indexRequest.index());
+                assertEquals(SLM_HISTORY_DATA_STREAM, indexRequest.index());
                 final String indexedDocument = indexRequest.source().utf8ToString();
                 assertThat(indexedDocument, containsString(policy.getId()));
                 assertThat(indexedDocument, containsString(policy.getRepository()));
@@ -141,7 +141,7 @@ public class SnapshotHistoryStoreTests extends ESTestCase {
                 assertThat(action, instanceOf(IndexAction.class));
                 assertThat(request, instanceOf(IndexRequest.class));
                 IndexRequest indexRequest = (IndexRequest) request;
-                assertEquals(SLM_HISTORY_ALIAS, indexRequest.index());
+                assertEquals(SLM_HISTORY_DATA_STREAM, indexRequest.index());
                 final String indexedDocument = indexRequest.source().utf8ToString();
                 assertThat(indexedDocument, containsString(policy.getId()));
                 assertThat(indexedDocument, containsString(policy.getRepository()));
@@ -167,180 +167,6 @@ public class SnapshotHistoryStoreTests extends ESTestCase {
         }
     }
 
-    public void testHistoryIndexNeedsCreation() throws InterruptedException {
-        ClusterState state = ClusterState.builder(new ClusterName(randomAlphaOfLength(5)))
-            .metadata(Metadata.builder())
-            .build();
-
-        client.setVerifier((a, r, l) -> {
-            assertThat(a, instanceOf(CreateIndexAction.class));
-            assertThat(r, instanceOf(CreateIndexRequest.class));
-            CreateIndexRequest request = (CreateIndexRequest) r;
-            assertThat(request.aliases(), hasSize(1));
-            request.aliases().forEach(alias -> {
-                assertThat(alias.name(), equalTo(SLM_HISTORY_ALIAS));
-                assertTrue(alias.writeIndex());
-            });
-            return new CreateIndexResponse(true, true, request.index());
-        });
-
-        CountDownLatch latch = new CountDownLatch(1);
-        SnapshotHistoryStore.ensureHistoryIndex(client, state, new LatchedActionListener<>(ActionListener.wrap(
-            Assert::assertTrue,
-            ex -> {
-                logger.error(ex);
-                fail("should have called onResponse, not onFailure");
-            }), latch));
-
-        awaitLatch(latch, 10, TimeUnit.SECONDS);
-    }
-
-    public void testHistoryIndexProperlyExistsAlready() throws InterruptedException {
-        ClusterState state = ClusterState.builder(new ClusterName(randomAlphaOfLength(5)))
-            .metadata(Metadata.builder()
-                .put(IndexMetadata.builder(SLM_HISTORY_INDEX_PREFIX + "000001")
-                    .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
-                    .numberOfShards(randomIntBetween(1,10))
-                    .numberOfReplicas(randomIntBetween(1,10))
-                    .putAlias(AliasMetadata.builder(SLM_HISTORY_ALIAS)
-                        .writeIndex(true)
-                        .build())))
-            .build();
-
-        client.setVerifier((a, r, l) -> {
-            fail("no client calls should have been made");
-            return null;
-        });
-
-        CountDownLatch latch = new CountDownLatch(1);
-        SnapshotHistoryStore.ensureHistoryIndex(client, state, new LatchedActionListener<>(ActionListener.wrap(
-            Assert::assertFalse,
-            ex -> {
-                logger.error(ex);
-                fail("should have called onResponse, not onFailure");
-            }), latch));
-
-        awaitLatch(latch, 10, TimeUnit.SECONDS);
-    }
-
-    public void testHistoryIndexHasNoWriteIndex() throws InterruptedException {
-        ClusterState state = ClusterState.builder(new ClusterName(randomAlphaOfLength(5)))
-            .metadata(Metadata.builder()
-                .put(IndexMetadata.builder(SLM_HISTORY_INDEX_PREFIX + "000001")
-                    .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
-                    .numberOfShards(randomIntBetween(1,10))
-                    .numberOfReplicas(randomIntBetween(1,10))
-                    .putAlias(AliasMetadata.builder(SLM_HISTORY_ALIAS)
-                        .build()))
-            .put(IndexMetadata.builder(randomAlphaOfLength(5))
-                .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
-                .numberOfShards(randomIntBetween(1,10))
-                .numberOfReplicas(randomIntBetween(1,10))
-                .putAlias(AliasMetadata.builder(SLM_HISTORY_ALIAS)
-                    .build())))
-            .build();
-
-        client.setVerifier((a, r, l) -> {
-            fail("no client calls should have been made");
-            return null;
-        });
-
-        CountDownLatch latch = new CountDownLatch(1);
-        SnapshotHistoryStore.ensureHistoryIndex(client, state, new LatchedActionListener<>(ActionListener.wrap(
-            indexCreated -> fail("should have called onFailure, not onResponse"),
-            ex -> {
-                assertThat(ex, instanceOf(IllegalStateException.class));
-                assertThat(ex.getMessage(), containsString("SLM history alias [" + SLM_HISTORY_ALIAS +
-                    "does not have a write index"));
-            }), latch));
-
-        awaitLatch(latch, 10, TimeUnit.SECONDS);
-    }
-
-    public void testHistoryIndexNotAlias() throws InterruptedException {
-        ClusterState state = ClusterState.builder(new ClusterName(randomAlphaOfLength(5)))
-            .metadata(Metadata.builder()
-                .put(IndexMetadata.builder(SLM_HISTORY_ALIAS)
-                    .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
-                    .numberOfShards(randomIntBetween(1,10))
-                    .numberOfReplicas(randomIntBetween(1,10))))
-            .build();
-
-        client.setVerifier((a, r, l) -> {
-            fail("no client calls should have been made");
-            return null;
-        });
-
-        CountDownLatch latch = new CountDownLatch(1);
-        SnapshotHistoryStore.ensureHistoryIndex(client, state, new LatchedActionListener<>(ActionListener.wrap(
-            indexCreated -> fail("should have called onFailure, not onResponse"),
-            ex -> {
-                assertThat(ex, instanceOf(IllegalStateException.class));
-                assertThat(ex.getMessage(), containsString("SLM history alias [" + SLM_HISTORY_ALIAS +
-                    "] already exists as concrete index"));
-            }), latch));
-
-        awaitLatch(latch, 10, TimeUnit.SECONDS);
-    }
-
-    public void testHistoryIndexCreatedConcurrently() throws InterruptedException {
-        ClusterState state = ClusterState.builder(new ClusterName(randomAlphaOfLength(5)))
-            .metadata(Metadata.builder())
-            .build();
-
-        client.setVerifier((a, r, l) -> {
-            assertThat(a, instanceOf(CreateIndexAction.class));
-            assertThat(r, instanceOf(CreateIndexRequest.class));
-            CreateIndexRequest request = (CreateIndexRequest) r;
-            assertThat(request.aliases(), hasSize(1));
-            request.aliases().forEach(alias -> {
-                assertThat(alias.name(), equalTo(SLM_HISTORY_ALIAS));
-                assertTrue(alias.writeIndex());
-            });
-            throw new ResourceAlreadyExistsException("that index already exists");
-        });
-
-        CountDownLatch latch = new CountDownLatch(1);
-        SnapshotHistoryStore.ensureHistoryIndex(client, state, new LatchedActionListener<>(ActionListener.wrap(
-            Assert::assertFalse,
-            ex -> {
-                logger.error(ex);
-                fail("should have called onResponse, not onFailure");
-            }), latch));
-
-        awaitLatch(latch, 10, TimeUnit.SECONDS);
-    }
-
-    public void testHistoryAliasDoesntExistButIndexDoes() throws InterruptedException {
-        final String initialIndex = SLM_HISTORY_INDEX_PREFIX + "000001";
-        ClusterState state = ClusterState.builder(new ClusterName(randomAlphaOfLength(5)))
-            .metadata(Metadata.builder()
-                .put(IndexMetadata.builder(initialIndex)
-                    .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
-                    .numberOfShards(randomIntBetween(1,10))
-                    .numberOfReplicas(randomIntBetween(1,10))))
-            .build();
-
-        client.setVerifier((a, r, l) -> {
-            fail("no client calls should have been made");
-            return null;
-        });
-
-        CountDownLatch latch = new CountDownLatch(1);
-        SnapshotHistoryStore.ensureHistoryIndex(client, state, new LatchedActionListener<>(ActionListener.wrap(
-            response -> {
-                logger.error(response);
-                fail("should have called onFailure, not onResponse");
-            },
-            ex -> {
-                assertThat(ex, instanceOf(IllegalStateException.class));
-                assertThat(ex.getMessage(), containsString("SLM history index [" + initialIndex +
-                    "] already exists but does not have alias [" + SLM_HISTORY_ALIAS + "]"));
-            }), latch));
-
-        awaitLatch(latch, 10, TimeUnit.SECONDS);
-    }
-
     @SuppressWarnings("unchecked")
     private void assertContainsMap(String indexedDocument, Map<String, Object> map) {
         map.forEach((k, v) -> {

x-pack/plugin/ilm/build.gradle

@@ -14,6 +14,7 @@ archivesBaseName = 'x-pack-ilm'
 dependencies {
   compileOnly project(path: xpackModule('core'), configuration: 'default')
   testImplementation project(path: xpackModule('core'), configuration: 'testArtifacts')
+  testImplementation project(path: xpackModule('data-streams'), configuration: 'default')
 }
 
 // add all sub-projects of the qa sub-project

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

@@ -55,7 +55,7 @@ import static org.elasticsearch.xpack.TimeSeriesRestDriver.createComposableTempl
 import static org.elasticsearch.xpack.TimeSeriesRestDriver.getStepKeyForIndex;
 import static org.elasticsearch.xpack.core.slm.history.SnapshotHistoryItem.CREATE_OPERATION;
 import static org.elasticsearch.xpack.core.slm.history.SnapshotHistoryItem.DELETE_OPERATION;
-import static org.elasticsearch.xpack.core.slm.history.SnapshotHistoryStore.SLM_HISTORY_INDEX_PREFIX;
+import static org.elasticsearch.xpack.core.slm.history.SnapshotHistoryStore.SLM_HISTORY_DATA_STREAM;
 import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -625,7 +625,7 @@ public class SnapshotLifecycleRestIT extends ESRestTestCase {
     // This method should be called inside an assertBusy, it has no retry logic of its own
     @SuppressWarnings("unchecked")
     private void assertHistoryIsPresent(String policyName, boolean success, String repository, String operation) throws IOException {
-        final Request historySearchRequest = new Request("GET", ".slm-history*/_search");
+        final Request historySearchRequest = new Request("GET", "slm-history*/_search");
         historySearchRequest.setJsonEntity("{\n" +
             "  \"query\": {\n" +
             "    \"bool\": {\n" +
@@ -674,7 +674,7 @@ public class SnapshotLifecycleRestIT extends ESRestTestCase {
     }
 
     private void assertHistoryIndexWaitingForRollover() throws IOException {
-        Step.StepKey stepKey = getStepKeyForIndex(client(), SLM_HISTORY_INDEX_PREFIX + "000001");
+        Step.StepKey stepKey = getStepKeyForIndex(client(), DataStream.getDefaultBackingIndexName(SLM_HISTORY_DATA_STREAM, 1));
         assertEquals("hot", stepKey.getPhase());
         assertEquals(RolloverAction.NAME, stepKey.getAction());
         assertEquals(WaitForRolloverReadyStep.NAME, stepKey.getName());

x-pack/plugin/ilm/src/internalClusterTest/java/org/elasticsearch/xpack/slm/SLMSnapshotBlockingIntegTests.java

@@ -30,6 +30,7 @@ import org.elasticsearch.snapshots.SnapshotState;
 import org.elasticsearch.snapshots.mockstore.MockRepository;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
+import org.elasticsearch.xpack.core.action.DeleteDataStreamAction;
 import org.elasticsearch.xpack.core.ilm.LifecycleSettings;
 import org.elasticsearch.xpack.core.slm.SnapshotInvocationRecord;
 import org.elasticsearch.xpack.core.slm.SnapshotLifecyclePolicy;
@@ -39,6 +40,7 @@ import org.elasticsearch.xpack.core.slm.action.ExecuteSnapshotLifecycleAction;
 import org.elasticsearch.xpack.core.slm.action.ExecuteSnapshotRetentionAction;
 import org.elasticsearch.xpack.core.slm.action.GetSnapshotLifecycleAction;
 import org.elasticsearch.xpack.core.slm.action.PutSnapshotLifecycleAction;
+import org.elasticsearch.xpack.datastreams.DataStreamsPlugin;
 import org.elasticsearch.xpack.ilm.IndexLifecycle;
 import org.junit.After;
 import org.junit.Before;
@@ -53,6 +55,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.xpack.core.slm.history.SnapshotHistoryStore.SLM_HISTORY_DATA_STREAM;
 import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
@@ -84,13 +87,16 @@ public class SLMSnapshotBlockingIntegTests extends AbstractSnapshotIntegTestCase
     }
 
     @After
-    public void awaitNoMoreRunningOps() throws Exception {
+    public void cleanUp() throws Exception {
         awaitNoMoreRunningOperations(internalCluster().getMasterName());
+        DeleteDataStreamAction.Request req = new DeleteDataStreamAction.Request(new String[]{SLM_HISTORY_DATA_STREAM});
+        assertAcked(client().execute(DeleteDataStreamAction.INSTANCE, req).actionGet());
     }
 
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return Arrays.asList(MockRepository.Plugin.class, LocalStateCompositeXPackPlugin.class, IndexLifecycle.class);
+        return Arrays.asList(MockRepository.Plugin.class, LocalStateCompositeXPackPlugin.class, IndexLifecycle.class,
+            DataStreamsPlugin.class);
     }
 
     public void testSnapshotInProgress() throws Exception {
@@ -240,7 +246,7 @@ public class SLMSnapshotBlockingIntegTests extends AbstractSnapshotIntegTestCase
 
             // Assert that the history document has been written for taking the snapshot and deleting it
             assertBusy(() -> {
-                SearchResponse resp = client().prepareSearch(".slm-history*")
+                SearchResponse resp = client().prepareSearch("slm-history*")
                     .setQuery(QueryBuilders.matchQuery("snapshot_name", completedSnapshotName)).get();
                 logger.info("--> checking history written for {}, got: {}",
                     completedSnapshotName, Strings.arrayToCommaDelimitedString(resp.getHits().getHits()));