Validation for data stream creation (#54083)

Author: danhermann

server/src/main/java/org/elasticsearch/action/admin/indices/datastream/CreateDataStreamAction.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.cluster.metadata.MetaDataCreateIndexService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
@@ -165,9 +166,11 @@ static ClusterState createDataStream(ClusterState currentState, Request request)
                 throw new IllegalArgumentException("data_stream [" + request.name + "] already exists");
             }
 
+            MetaDataCreateIndexService.validateIndexOrAliasName(request.name,
+                (s1, s2) -> new IllegalArgumentException("data_stream [" + s1 + "] " + s2));
+
             MetaData.Builder builder = MetaData.builder(currentState.metaData()).put(
                 new DataStream(request.name, request.timestampFieldName, Collections.emptyList()));
-
             logger.info("adding data stream [{}]", request.name);
             return ClusterState.builder(currentState).metaData(builder).build();
         }

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

@@ -1277,7 +1277,7 @@ public Builder generateClusterUuidIfNeeded() {
 
         public MetaData build() {
             // TODO: We should move these datastructures to IndexNameExpressionResolver, this will give the following benefits:
-            // 1) The datastructures will only be rebuilded when needed. Now during serializing we rebuild these datastructures
+            // 1) The datastructures will be rebuilt only when needed. Now during serializing we rebuild these datastructures
             //    while these datastructures aren't even used.
             // 2) The aliasAndIndexLookup can be updated instead of rebuilding it all the time.
 
@@ -1315,20 +1315,21 @@ public MetaData build() {
                 // iterate again and constructs a helpful message
                 ArrayList<String> duplicates = new ArrayList<>();
                 for (ObjectCursor<IndexMetaData> cursor : indices.values()) {
-                    for (String alias: duplicateAliasesIndices) {
+                    for (String alias : duplicateAliasesIndices) {
                         if (cursor.value.getAliases().containsKey(alias)) {
                             duplicates.add(alias + " (alias of " + cursor.value.getIndex() + ")");
                         }
                     }
                 }
                 assert duplicates.size() > 0;
                 throw new IllegalStateException("index and alias names need to be unique, but the following duplicates were found ["
-                    + Strings.collectionToCommaDelimitedString(duplicates)+ "]");
+                    + Strings.collectionToCommaDelimitedString(duplicates) + "]");
 
             }
 
             SortedMap<String, AliasOrIndex> aliasAndIndexLookup = Collections.unmodifiableSortedMap(buildAliasAndIndexLookup());
 
+            validateDataStreams(aliasAndIndexLookup);
 
             // build all concrete indices arrays:
             // TODO: I think we can remove these arrays. it isn't worth the effort, for operations on all indices.
@@ -1371,6 +1372,24 @@ private SortedMap<String, AliasOrIndex> buildAliasAndIndexLookup() {
             return aliasAndIndexLookup;
         }
 
+        private void validateDataStreams(SortedMap<String, AliasOrIndex> aliasAndIndexLookup) {
+            DataStreamMetadata dsMetadata = (DataStreamMetadata) customs.get(DataStreamMetadata.TYPE);
+            if (dsMetadata != null) {
+                for (DataStream ds : dsMetadata.dataStreams().values()) {
+                    if (aliasAndIndexLookup.containsKey(ds.getName())) {
+                        throw new IllegalStateException("data stream [" + ds.getName() + "] conflicts with existing index or alias");
+                    }
+
+                    SortedMap<?, ?> map = aliasAndIndexLookup.subMap(ds.getName() + "-", ds.getName() + "."); // '.' is the char after '-'
+                    if (map.size() != 0) {
+                        throw new IllegalStateException("data stream [" + ds.getName() +
+                            "] could create backing indices that conflict with " + map.size() + " existing index(s) or alias(s)" +
+                            " including '" + map.firstKey() + "'");
+                    }
+                }
+            }
+        }
+
         public static void toXContent(MetaData metaData, XContentBuilder builder, ToXContent.Params params) throws IOException {
             XContentContext context = XContentContext.valueOf(params.param(CONTEXT_MODE_PARAM, CONTEXT_MODE_API));
 

server/src/test/java/org/elasticsearch/action/admin/indices/datastream/CreateDataStreamRequestTests.java

@@ -82,4 +82,13 @@ public void testCreateDuplicateDataStream() {
             () -> CreateDataStreamAction.TransportAction.createDataStream(cs, req));
         assertThat(e.getMessage(), containsString("data_stream [" + dataStreamName + "] already exists"));
     }
+
+    public void testCreateDataStreamWithInvalidName() {
+        final String dataStreamName = "_My-da#ta- ,stream-";
+        ClusterState cs = ClusterState.builder(new ClusterName("_name")).build();
+        CreateDataStreamAction.Request req = new CreateDataStreamAction.Request(dataStreamName);
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class,
+            () -> CreateDataStreamAction.TransportAction.createDataStream(cs, req));
+        assertThat(e.getMessage(), containsString("must not contain the following characters"));
+    }
 }

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

@@ -44,6 +44,7 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -907,6 +908,51 @@ public void testBuilderRejectsNullInCustoms() {
         assertThat(expectThrows(NullPointerException.class, () -> builder.customs(map)).getMessage(), containsString(key));
     }
 
+    public void testBuilderRejectsDataStreamThatConflictsWithIndex() {
+        final String dataStreamName = "my-data-stream";
+        MetaData.Builder b = MetaData.builder()
+            .put(IndexMetaData.builder(dataStreamName)
+                .settings(settings(Version.CURRENT))
+                .numberOfShards(1)
+                .numberOfReplicas(1)
+                .build(), false)
+            .put(new DataStream(dataStreamName, "ts", Collections.emptyList()));
+
+        IllegalStateException e = expectThrows(IllegalStateException.class, b::build);
+        assertThat(e.getMessage(), containsString("data stream [" + dataStreamName + "] conflicts with existing index or alias"));
+    }
+
+    public void testBuilderRejectsDataStreamThatConflictsWithAlias() {
+        final String dataStreamName = "my-data-stream";
+        MetaData.Builder b = MetaData.builder()
+            .put(IndexMetaData.builder(dataStreamName + "z")
+                .settings(settings(Version.CURRENT))
+                .numberOfShards(1)
+                .numberOfReplicas(1)
+                .putAlias(AliasMetaData.builder(dataStreamName).build())
+                .build(), false)
+            .put(new DataStream(dataStreamName, "ts", Collections.emptyList()));
+
+        IllegalStateException e = expectThrows(IllegalStateException.class, b::build);
+        assertThat(e.getMessage(), containsString("data stream [" + dataStreamName + "] conflicts with existing index or alias"));
+    }
+
+    public void testBuilderRejectsDataStreamWithConflictingBackingIndices() {
+        final String dataStreamName = "my-data-stream";
+        final String conflictingIndex = dataStreamName + "-00001";
+        MetaData.Builder b = MetaData.builder()
+            .put(IndexMetaData.builder(conflictingIndex)
+                .settings(settings(Version.CURRENT))
+                .numberOfShards(1)
+                .numberOfReplicas(1)
+                .build(), false)
+            .put(new DataStream(dataStreamName, "ts", Collections.emptyList()));
+
+        IllegalStateException e = expectThrows(IllegalStateException.class, b::build);
+        assertThat(e.getMessage(), containsString("data stream [" + dataStreamName +
+            "] could create backing indices that conflict with 1 existing index(s) or alias(s) including '" + conflictingIndex + "'"));
+    }
+
     public void testSerialization() throws IOException {
         final MetaData orig = randomMetaData();
         final BytesStreamOutput out = new BytesStreamOutput();