DataStream creation validation allows for prefixed indices (#57750)

We want to validate the DataStreams on creation to make sure the future backing
indices would not clash with existing indices in the system (so we can
always rollover the data stream).
This changes the validation logic to allow for a DataStream to be created
with a backing index that has a prefix (eg. `shrink-foo-000001`) even if the
former backing index (`foo-000001`) exists in the system.
The new validation logic will look for potential index conflicts with indices
in the system that have the counter in the name greater than the data stream's
generation.

This ensures that the `DataStream`'s future rollovers are safe because for a
`DataStream` `foo` of generation 4, we will look for standalone indices in the
form of `foo-%06d` with the counter greater than 4 (ie. validation will fail if
`foo-000006` exists in the system), but will also allow replacing a
backing index with an index named by prefixing the backing index it replaces.

server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

@@ -1619,4 +1619,25 @@ public class IndexMetadata implements Diffable<IndexMetadata>, ToXContentFragmen
         }
         return factor;
     }
+
+    /**
+     * Parses the number from the rolled over index name. It also supports the date-math format (ie. index name is wrapped in &lt; and &gt;)
+     * Eg.
+     * - For "logs-000002" it'll return 2
+     * - For "&lt;logs-{now/d}-3&gt;" it'll return 3
+     * @throws IllegalArgumentException if the index doesn't contain a "-" separator or if the last token after the separator is not a
+     * number
+     */
+    public static int parseIndexNameCounter(String indexName) {
+        int numberIndex = indexName.lastIndexOf("-");
+        if (numberIndex == -1) {
+            throw new IllegalArgumentException("no - separator found in index name [" + indexName + "]");
+        }
+        try {
+            return Integer.parseInt(indexName.substring(numberIndex + 1, indexName.endsWith(">") ? indexName.length() - 1 :
+                indexName.length()));
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException("unable to parse the index name [" + indexName + "] to extract the counter", e);
+        }
+    }
 }

server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java

@@ -1376,7 +1376,7 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
 
             SortedMap<String, IndexAbstraction> indicesLookup = Collections.unmodifiableSortedMap(buildIndicesLookup());
 
-            validateDataStreams(indicesLookup);
+            validateDataStreams(indicesLookup, (DataStreamMetadata) customs.get(DataStreamMetadata.TYPE));
 
             // build all concrete indices arrays:
             // TODO: I think we can remove these arrays. it isn't worth the effort, for operations on all indices.
@@ -1452,27 +1452,41 @@ public class Metadata implements Iterable<IndexMetadata>, Diffable<Metadata>, To
             return indicesLookup;
         }
 
-        private void validateDataStreams(SortedMap<String, IndexAbstraction> indicesLookup) {
-            DataStreamMetadata dsMetadata = (DataStreamMetadata) customs.get(DataStreamMetadata.TYPE);
+        /**
+         * Validates there isn't any index with a name that would clash with the future backing indices of the existing data streams.
+         *
+         * For eg. if data stream `foo` has backing indices [`foo-000001`, `foo-000002`] and the indices lookup contains indices
+         * `foo-000001`, `foo-000002` and `foo-000006` this will throw an IllegalStateException (as attempting to rollover the `foo` data
+         * stream from generation 5 to 6 will not be possible)
+         *
+         * @param indicesLookup the indices in the system (this includes the data streams backing indices)
+         * @param dsMetadata    the data streams in the system
+         */
+        static void validateDataStreams(SortedMap<String, IndexAbstraction> indicesLookup, @Nullable DataStreamMetadata dsMetadata) {
             if (dsMetadata != null) {
                 for (DataStream ds : dsMetadata.dataStreams().values()) {
-                    SortedMap<String, IndexAbstraction> potentialConflicts =
-                        indicesLookup.subMap(ds.getName() + "-", ds.getName() + "."); // '.' is the char after '-'
-                    if (potentialConflicts.size() != 0) {
-                        List<String> indexNames = ds.getIndices().stream().map(Index::getName).collect(Collectors.toList());
-                        List<String> conflicts = new ArrayList<>();
-                        for (Map.Entry<String, IndexAbstraction> entry : potentialConflicts.entrySet()) {
-                            if (entry.getValue().getType() != IndexAbstraction.Type.CONCRETE_INDEX ||
-                                indexNames.contains(entry.getKey()) == false) {
-                                conflicts.add(entry.getKey());
-                            }
-                        }
-
-                        if (conflicts.size() > 0) {
-                            throw new IllegalStateException("data stream [" + ds.getName() +
-                                "] could create backing indices that conflict with " + conflicts.size() + " existing index(s) or alias(s)" +
-                                " including '" + conflicts.get(0) + "'");
-                        }
+                    Map<String, IndexAbstraction> conflicts =
+                        indicesLookup.subMap(ds.getName() + "-", ds.getName() + ".") // '.' is the char after '-'
+                            .entrySet().stream()
+                            .filter(entry -> {
+                                if (entry.getValue().getType() != IndexAbstraction.Type.CONCRETE_INDEX) {
+                                    return true;
+                                } else {
+                                    int indexNameCounter;
+                                    try {
+                                        indexNameCounter = IndexMetadata.parseIndexNameCounter(entry.getKey());
+                                    } catch (IllegalArgumentException e) {
+                                        // index name is not in the %s-%d+ format so it will not crash with backing indices
+                                        return false;
+                                    }
+                                    return indexNameCounter > ds.getGeneration();
+                                }
+                            }).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+                    if (conflicts.size() > 0) {
+                        throw new IllegalStateException("data stream [" + ds.getName() +
+                            "] could create backing indices that conflict with " + conflicts.size() + " existing index(s) or alias(s)" +
+                            " including '" + conflicts.keySet().iterator().next() + "'");
                     }
                 }
             }

server/src/test/java/org/elasticsearch/cluster/metadata/IndexMetadataTests.java

@@ -50,6 +50,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 
+import static org.elasticsearch.cluster.metadata.IndexMetadata.parseIndexNameCounter;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
@@ -331,4 +332,31 @@ public class IndexMetadataTests extends ESTestCase {
                 "Failed to parse value [" + numberOfReplicas + "] for setting [index.number_of_replicas] must be >= 0"));
     }
 
+    public void testParseIndexNameReturnsCounter() {
+        assertThat(parseIndexNameCounter("logs-000003"), is(3));
+        assertThat(parseIndexNameCounter("shrink-logs-000003"), is(3));
+    }
+
+    public void testParseIndexNameSupportsDateMathPattern() {
+        assertThat(parseIndexNameCounter("<logs-{now/d}-1>"), is(1));
+    }
+
+    public void testParseIndexNameThrowExceptionWhenNoSeparatorIsPresent() {
+        try {
+            parseIndexNameCounter("testIndexNameWithoutDash");
+            fail("expected to fail as the index name contains no - separator");
+        } catch (IllegalArgumentException e) {
+            assertThat(e.getMessage(), is("no - separator found in index name [testIndexNameWithoutDash]"));
+        }
+    }
+
+    public void testParseIndexNameCannotFormatNumber() {
+        try {
+            parseIndexNameCounter("testIndexName-000a2");
+            fail("expected to fail as the index name doesn't end with digits");
+        } catch (IllegalArgumentException e) {
+            assertThat(e.getMessage(), is("unable to parse the index name [testIndexName-000a2] to extract the counter"));
+        }
+    }
+
 }

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataTests.java

@@ -52,10 +52,12 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.SortedMap;
+import java.util.TreeMap;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.cluster.DataStreamTestHelper.createBackingIndex;
 import static org.elasticsearch.cluster.DataStreamTestHelper.createFirstBackingIndex;
+import static org.elasticsearch.cluster.metadata.Metadata.Builder.validateDataStreams;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
@@ -1078,6 +1080,138 @@ public class MetadataTests extends ESTestCase {
         assertTrue(Metadata.isGlobalStateEquals(orig, fromStreamMeta));
     }
 
+    public void testValidateDataStreamsNoConflicts() {
+        Metadata metadata = createIndices(5, 10, "foo-datastream").metadata;
+        // don't expect any exception when validating a system without indices that would conflict with future backing indices
+        validateDataStreams(metadata.getIndicesLookup(), (DataStreamMetadata) metadata.customs().get(DataStreamMetadata.TYPE));
+    }
+
+    public void testValidateDataStreamsThrowsExceptionOnConflict() {
+        String dataStreamName = "foo-datastream";
+        int generations = 10;
+        List<IndexMetadata> backingIndices = new ArrayList<>(generations);
+        for (int i = 1; i <= generations; i++) {
+            IndexMetadata idx = createBackingIndex(dataStreamName, i).build();
+            backingIndices.add(idx);
+        }
+        DataStream dataStream = new DataStream(
+            dataStreamName,
+            "ts",
+            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
+            backingIndices.size()
+        );
+
+        IndexAbstraction.DataStream dataStreamAbstraction = new IndexAbstraction.DataStream(dataStream, backingIndices);
+        // manually building the indices lookup as going through Metadata.Builder#build would trigger the validate method and would fail
+        SortedMap<String, IndexAbstraction> indicesLookup = new TreeMap<>();
+        for (IndexMetadata indexMeta : backingIndices) {
+            indicesLookup.put(indexMeta.getIndex().getName(), new IndexAbstraction.Index(indexMeta, dataStreamAbstraction));
+        }
+
+        // add the offending index to the indices lookup
+        IndexMetadata standaloneIndexConflictingWithBackingIndices = createBackingIndex(dataStreamName, 2 * generations).build();
+        Index index = standaloneIndexConflictingWithBackingIndices.getIndex();
+        indicesLookup.put(index.getName(), new IndexAbstraction.Index(standaloneIndexConflictingWithBackingIndices, null));
+
+        DataStreamMetadata dataStreamMetadata = new DataStreamMetadata(Map.of(dataStreamName, dataStream));
+
+        IllegalStateException illegalStateException =
+            expectThrows(IllegalStateException.class, () -> validateDataStreams(indicesLookup, dataStreamMetadata));
+        assertThat(illegalStateException.getMessage(),
+            is("data stream [foo-datastream] could create backing indices that conflict with 1 existing index(s) or alias(s) " +
+                "including 'foo-datastream-000020'"));
+    }
+
+    public void testValidateDataStreamsIgnoresIndicesWithoutCounter() {
+        String dataStreamName = "foo-datastream";
+        Metadata metadata = Metadata.builder(createIndices(10, 10, dataStreamName).metadata)
+            .put(
+                new IndexMetadata.Builder(dataStreamName + "-index-without-counter")
+                    .settings(settings(Version.CURRENT))
+                    .numberOfShards(1)
+                    .numberOfReplicas(1)
+            )
+            .put(
+                new IndexMetadata.Builder(dataStreamName + randomAlphaOfLength(10))
+                    .settings(settings(Version.CURRENT))
+                    .numberOfShards(1)
+                    .numberOfReplicas(1)
+
+            )
+            .put(
+                new IndexMetadata.Builder(randomAlphaOfLength(10))
+                    .settings(settings(Version.CURRENT))
+                    .numberOfShards(1)
+                    .numberOfReplicas(1)
+
+            )
+            .build();
+        // don't expect any exception when validating against non-backing indinces that don't conform to the backing indices naming
+        // convention
+        validateDataStreams(metadata.getIndicesLookup(), (DataStreamMetadata) metadata.customs().get(DataStreamMetadata.TYPE));
+    }
+
+    public void testValidateDataStreamsAllowsPrefixedBackingIndices() {
+        String dataStreamName = "foo-datastream";
+        int generations = 10;
+        List<IndexMetadata> backingIndices = new ArrayList<>(generations);
+        for (int i = 1; i <= generations; i++) {
+            IndexMetadata idx;
+            if (i % 2 == 0 && i < generations) {
+                idx = IndexMetadata.builder("shrink-" + DataStream.getBackingIndexName(dataStreamName, i))
+                    .settings(ESTestCase.settings(Version.CURRENT).put("index.hidden", true))
+                    .numberOfShards(1)
+                    .numberOfReplicas(1)
+                    .build();
+            } else {
+                idx = createBackingIndex(dataStreamName, i).build();
+            }
+            backingIndices.add(idx);
+        }
+        DataStream dataStream = new DataStream(
+            dataStreamName,
+            "ts",
+            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()),
+            backingIndices.size()
+        );
+
+        IndexAbstraction.DataStream dataStreamAbstraction = new IndexAbstraction.DataStream(dataStream, backingIndices);
+        // manually building the indices lookup as going through Metadata.Builder#build would trigger the validate method already
+        SortedMap<String, IndexAbstraction> indicesLookup = new TreeMap<>();
+        for (IndexMetadata indexMeta : backingIndices) {
+            indicesLookup.put(indexMeta.getIndex().getName(), new IndexAbstraction.Index(indexMeta, dataStreamAbstraction));
+        }
+
+        for (int i = 1; i <= generations; i++) {
+            // for the indices that we added in the data stream with a "shrink-" prefix, add the non-prefixed indices to the lookup
+            if (i % 2 == 0 && i < generations) {
+                IndexMetadata indexMeta = createBackingIndex(dataStreamName, i).build();
+                indicesLookup.put(indexMeta.getIndex().getName(), new IndexAbstraction.Index(indexMeta, dataStreamAbstraction));
+            }
+        }
+        DataStreamMetadata dataStreamMetadata = new DataStreamMetadata(Map.of(dataStreamName, dataStream));
+
+        // prefixed indices with a lower generation than the data stream's generation are allowed even if the non-prefixed, matching the
+        // data stream backing indices naming pattern, indices are already in the system
+        validateDataStreams(indicesLookup, dataStreamMetadata);
+    }
+
+
+    public void testValidateDataStreamsForNullDataStreamMetadata() {
+        Metadata metadata = Metadata.builder().put(
+            IndexMetadata.builder("foo-index")
+                .settings(settings(Version.CURRENT))
+                .numberOfShards(1)
+                .numberOfReplicas(1)
+        ).build();
+
+        try {
+            validateDataStreams(metadata.getIndicesLookup(), null);
+        } catch (Exception e) {
+            fail("did not expect exception when validating a system without any data streams but got " + e.getMessage());
+        }
+    }
+
     public static Metadata randomMetadata() {
         Metadata.Builder md = Metadata.builder()
             .put(buildIndexMetadata("index", "alias", randomBoolean() ? null : randomBoolean()).build(), randomBoolean())

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

@@ -27,6 +27,8 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Objects;
 
+import static org.elasticsearch.cluster.metadata.IndexMetadata.parseIndexNameCounter;
+
 /**
  * After we performed the index rollover we wait for the the configured number of shards for the rolled over index (ie. newly created
  * index) to become available.
@@ -134,27 +136,6 @@ public class WaitForActiveShardsStep extends ClusterStateWaitStep {
         return new Result(false, new Info(errorMessage));
     }
 
-    /**
-     * Parses the number from the rolled over index name. It also supports the date-math format (ie. index name is wrapped in < and >)
-     * <p>
-     * Eg.
-     * <p>
-     * - For "logs-000002" it'll return 2
-     * - For "&lt;logs-{now/d}-3&gt;" it'll return 3
-     */
-    static int parseIndexNameCounter(String indexName) {
-        int numberIndex = indexName.lastIndexOf("-");
-        if (numberIndex == -1) {
-            throw new IllegalArgumentException("no - separator found in index name [" + indexName + "]");
-        }
-        try {
-            return Integer.parseInt(indexName.substring(numberIndex + 1, indexName.endsWith(">") ? indexName.length() - 1 :
-                indexName.length()));
-        } catch (NumberFormatException e) {
-            throw new IllegalArgumentException("unable to parse the index name [" + indexName + "] to extract the counter", e);
-        }
-    }
-
     static final class ActiveShardsInfo implements ToXContentObject {
 
         private final long currentActiveShardsCount;

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

@@ -27,7 +27,6 @@ import java.io.IOException;
 import java.util.List;
 import java.util.UUID;
 
-import static org.elasticsearch.xpack.core.ilm.WaitForActiveShardsStep.parseIndexNameCounter;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.is;
 
@@ -251,30 +250,4 @@ public class WaitForActiveShardsTests extends AbstractStepTestCase<WaitForActive
             containsString("[" + step.getKey().getAction() + "] lifecycle action for index [index-000000] executed but " +
                 "index no longer exists"));
     }
-
-    public void testParseIndexNameReturnsCounter() {
-        assertThat(parseIndexNameCounter("logs-000003"), is(3));
-    }
-
-    public void testParseIndexNameSupportsDateMathPattern() {
-        assertThat(parseIndexNameCounter("<logs-{now/d}-1>"), is(1));
-    }
-
-    public void testParseIndexNameThrowExceptionWhenNoSeparatorIsPresent() {
-        try {
-            parseIndexNameCounter("testIndexNameWithoutDash");
-            fail("expected to fail as the index name contains no - separator");
-        } catch (IllegalArgumentException e) {
-            assertThat(e.getMessage(), is("no - separator found in index name [testIndexNameWithoutDash]"));
-        }
-    }
-
-    public void testParseIndexNameCannotFormatNumber() {
-        try {
-            parseIndexNameCounter("testIndexName-000a2");
-            fail("expected to fail as the index name doesn't end with digits");
-        } catch (IllegalArgumentException e) {
-            assertThat(e.getMessage(), is("unable to parse the index name [testIndexName-000a2] to extract the counter"));
-        }
-    }
 }