Validate Data Streams reference a template on composable template update (#59106)

This commit adds validation that when a composable index template is updated, that the number
of unreferenced data streams does not increase. While it is still possible to have data streams
without a backing template (through snapshot restoration), this reduces the chance of getting
in to that scenario.

Relates to #53100

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

@@ -69,6 +69,7 @@ import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
@@ -493,6 +494,7 @@ public class MetadataIndexTemplateService {
         }
 
         validate(name, finalIndexTemplate);
+        validateDataStreamsStillReferenced(currentState, name, finalIndexTemplate);
 
         // Finally, right before adding the template, we need to ensure that the composite settings,
         // mappings, and aliases are valid after it's been composed with the component templates
@@ -511,6 +513,54 @@ public class MetadataIndexTemplateService {
             .build();
     }
 
+    /**
+     * Validate that by changing or adding {@code newTemplate}, there are
+     * no unreferenced data streams. Note that this scenario is still possible
+     * due to snapshot restores, but this validation is best-effort at template
+     * addition/update time
+     */
+    private static void validateDataStreamsStillReferenced(ClusterState state, String templateName,
+                                                           ComposableIndexTemplate newTemplate) {
+        final Set<String> dataStreams = state.metadata().dataStreams().keySet();
+
+        Function<Metadata, Set<String>> findUnreferencedDataStreams = meta -> {
+            final Set<String> unreferenced = new HashSet<>();
+            // For each data stream that we have, see whether it's covered by a different
+            // template (which is great), or whether it's now uncovered by any template
+            for (String dataStream : dataStreams) {
+                final String matchingTemplate = findV2Template(meta, dataStream, false);
+                if (matchingTemplate == null) {
+                    unreferenced.add(dataStream);
+                } else {
+                    // We found a template that still matches, great! Buuuuttt... check whether it
+                    // is a data stream template, as it's only useful if it has a data stream definition
+                    if (meta.templatesV2().get(matchingTemplate).getDataStreamTemplate() == null) {
+                        unreferenced.add(dataStream);
+                    }
+                }
+            }
+            return unreferenced;
+        };
+
+        // Find data streams that are currently unreferenced
+        final Set<String> currentlyUnreferenced = findUnreferencedDataStreams.apply(state.metadata());
+
+        // Generate a metadata as if the new template were actually in the cluster state
+        final Metadata updatedMetadata = Metadata.builder(state.metadata()).put(templateName, newTemplate).build();
+        // Find the data streams that would be unreferenced now that the template is updated/added
+        final Set<String> newlyUnreferenced = findUnreferencedDataStreams.apply(updatedMetadata);
+
+        // If we found any data streams that used to be covered, but will no longer be covered by
+        // changing this template, then blow up with as much helpful information as we can muster
+        if (newlyUnreferenced.size() > currentlyUnreferenced.size()) {
+            throw new IllegalArgumentException("composable template [" + templateName + "] with index patterns " +
+                newTemplate.indexPatterns() + ", priority [" + newTemplate.priority() + "] " +
+                (newTemplate.getDataStreamTemplate() == null ? "and no data stream configuration " : "") +
+                "would cause data streams " +
+                newlyUnreferenced + " to no longer match a data stream template");
+        }
+    }
+
     /**
      * Return a map of v1 template names to their index patterns for v1 templates that would overlap
      * with the given v2 template's index patterns.

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

@@ -20,6 +20,7 @@
 package org.elasticsearch.cluster.metadata;
 
 import com.fasterxml.jackson.core.JsonParseException;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.alias.Alias;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
@@ -37,6 +38,7 @@ import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.index.Index;
 import org.elasticsearch.index.mapper.MapperParsingException;
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.indices.IndexTemplateMissingException;
@@ -1062,6 +1064,88 @@ public class MetadataIndexTemplateServiceTests extends ESSingleNodeTestCase {
         assertThat(metadataIndexTemplateService.addIndexTemplateV2(state, false, "foo", template), equalTo(state));
     }
 
+    public void testUnreferencedDataStreamsWhenAddingTemplate() throws Exception {
+        ClusterState state = ClusterState.EMPTY_STATE;
+        final MetadataIndexTemplateService service = getMetadataIndexTemplateService();
+        state = ClusterState.builder(state)
+            .metadata(Metadata.builder(state.metadata())
+                .put(new DataStream("unreferenced",
+                    new DataStream.TimestampField("@timestamp", Collections.singletonMap("type", "date")),
+                    Collections.singletonList(new Index(".ds-unreferenced-000001", "uuid2"))))
+                .put(IndexMetadata.builder(".ds-unreferenced-000001")
+                    .settings(Settings.builder()
+                        .put(IndexMetadata.SETTING_INDEX_UUID, "uuid2")
+                        .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                        .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                        .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT)
+                        .build()))
+                .build())
+            .build();
+
+        ComposableIndexTemplate template = new ComposableIndexTemplate(Collections.singletonList("logs-*-*"), null, null,
+            100L, null, null, new ComposableIndexTemplate.DataStreamTemplate("@timestamp"));
+
+        state = service.addIndexTemplateV2(state, false, "logs", template);
+
+        ClusterState stateWithDS = ClusterState.builder(state)
+            .metadata(Metadata.builder(state.metadata())
+                .put(new DataStream("logs-mysql-default",
+                    new DataStream.TimestampField("@timestamp", Collections.singletonMap("type", "date")),
+                    Collections.singletonList(new Index(".ds-logs-mysql-default-000001", "uuid"))))
+                .put(IndexMetadata.builder(".ds-logs-mysql-default-000001")
+                    .settings(Settings.builder()
+                        .put(IndexMetadata.SETTING_INDEX_UUID, "uuid")
+                        .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                        .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                        .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT)
+                        .build()))
+                .build())
+            .build();
+
+        // Test replacing it with a version without the data stream config
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> {
+            ComposableIndexTemplate nonDSTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-*-*"), null, null,
+                100L, null, null, null);
+            service.addIndexTemplateV2(stateWithDS, false, "logs", nonDSTemplate);
+        });
+
+        assertThat(e.getMessage(),
+            containsString("composable template [logs] with index patterns [logs-*-*], priority [100] and no data stream " +
+                "configuration would cause data streams [unreferenced, logs-mysql-default] to no longer match a data stream template"));
+
+        // Test adding a higher priority version that would cause problems
+        e = expectThrows(IllegalArgumentException.class, () -> {
+            ComposableIndexTemplate nonDSTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-my*-*"), null, null,
+                105L, null, null, null);
+            service.addIndexTemplateV2(stateWithDS, false, "logs2", nonDSTemplate);
+        });
+
+        assertThat(e.getMessage(),
+            containsString("composable template [logs2] with index patterns [logs-my*-*], priority [105] and no data stream " +
+                "configuration would cause data streams [unreferenced, logs-mysql-default] to no longer match a data stream template"));
+
+        // Change the pattern to one that doesn't match the data stream
+        e = expectThrows(IllegalArgumentException.class, () -> {
+            ComposableIndexTemplate newTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-postgres-*"), null, null,
+                100L, null, null, new ComposableIndexTemplate.DataStreamTemplate("@timestamp"));
+            service.addIndexTemplateV2(stateWithDS, false, "logs", newTemplate);
+        });
+
+        assertThat(e.getMessage(),
+            containsString("composable template [logs] with index patterns [logs-postgres-*], priority [100] would " +
+                "cause data streams [unreferenced, logs-mysql-default] to no longer match a data stream template"));
+
+        // Add an additional template that matches our data stream at a lower priority
+        ComposableIndexTemplate mysqlTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-mysql-*"), null, null,
+            50L, null, null, new ComposableIndexTemplate.DataStreamTemplate("@timestamp"));
+        ClusterState stateWithDSAndTemplate = service.addIndexTemplateV2(stateWithDS, false, "logs-mysql", mysqlTemplate);
+
+        // We should be able to replace the "logs" template, because we have the "logs-mysql" template that can handle the data stream
+        ComposableIndexTemplate nonDSTemplate = new ComposableIndexTemplate(Collections.singletonList("logs-postgres-*"), null, null,
+            100L, null, null, null);
+        service.addIndexTemplateV2(stateWithDSAndTemplate, false, "logs", nonDSTemplate);
+    }
+
     private static List<Throwable> putTemplate(NamedXContentRegistry xContentRegistry, PutRequest request) {
         MetadataCreateIndexService createIndexService = new MetadataCreateIndexService(
                 Settings.EMPTY,