Disallow deletion of composable template if in use by data stream (#57957)

Currently a data stream requires a component template in order to be created. This is so that the
stream can be perpetuated when the index is automatically or manually rolled over. This commit
ensures that the composable index template used by the stream is not removed if the data stream
still exists.

Resolves #57004

server/src/internalClusterTest/java/org/elasticsearch/action/bulk/BulkIntegrationIT.java

@@ -24,6 +24,7 @@ import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.admin.indices.alias.Alias;
+import org.elasticsearch.action.admin.indices.datastream.DeleteDataStreamAction;
 import org.elasticsearch.action.admin.indices.datastream.GetDataStreamAction;
 import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
 import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
@@ -270,6 +271,8 @@ public class BulkIntegrationIT extends ESIntegTestCase {
         assertThat(getIndexResponse.getIndices(), hasItemInArray("logs-barfoo2"));
         assertThat(getIndexResponse.getIndices(), hasItemInArray("logs-barfoo3"));
 
+        DeleteDataStreamAction.Request deleteDSReq = new DeleteDataStreamAction.Request("*");
+        client().execute(DeleteDataStreamAction.INSTANCE, deleteDSReq).actionGet();
         DeleteComposableIndexTemplateAction.Request deleteTemplateRequest = new DeleteComposableIndexTemplateAction.Request("*");
         client().execute(DeleteComposableIndexTemplateAction.INSTANCE, deleteTemplateRequest).actionGet();
     }

server/src/internalClusterTest/java/org/elasticsearch/indices/DataStreamIT.java

@@ -18,6 +18,7 @@
  */
 package org.elasticsearch.indices;
 
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.admin.indices.datastream.CreateDataStreamAction;
@@ -60,6 +61,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Optional;
 
 import static org.elasticsearch.indices.IndicesOptionsIntegrationIT._flush;
 import static org.elasticsearch.indices.IndicesOptionsIntegrationIT.clearCache;
@@ -86,6 +88,8 @@ public class DataStreamIT extends ESIntegTestCase {
 
     @After
     public void deleteAllComposableTemplates() {
+        DeleteDataStreamAction.Request deleteDSRequest = new DeleteDataStreamAction.Request("*");
+        client().execute(DeleteDataStreamAction.INSTANCE, deleteDSRequest).actionGet();
         DeleteComposableIndexTemplateAction.Request deleteTemplateRequest = new DeleteComposableIndexTemplateAction.Request("*");
         client().execute(DeleteComposableIndexTemplateAction.INSTANCE, deleteTemplateRequest).actionGet();
     }
@@ -362,6 +366,29 @@ public class DataStreamIT extends ESIntegTestCase {
         verifyResolvability(wildcardExpression, client().admin().indices().prepareGetIndex().addIndices(wildcardExpression), false);
     }
 
+    public void testCannotDeleteComposableTemplateUsedByDataStream() throws Exception {
+        createIndexTemplate("id", "metrics-foobar*", "@timestamp1");
+        String dataStreamName = "metrics-foobar-baz";
+        CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName);
+        client().admin().indices().createDataStream(createDataStreamRequest).get();
+        createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName + "-eggplant");
+        client().admin().indices().createDataStream(createDataStreamRequest).get();
+
+        DeleteComposableIndexTemplateAction.Request req = new DeleteComposableIndexTemplateAction.Request("id");
+        Exception e = expectThrows(Exception.class, () -> client().execute(DeleteComposableIndexTemplateAction.INSTANCE, req).get());
+        Optional<Exception> maybeE = ExceptionsHelper.unwrapCausesAndSuppressed(e, err ->
+                err.getMessage().contains("unable to remove composable templates [id] " +
+                    "as they are in use by a data streams [metrics-foobar-baz, metrics-foobar-baz-eggplant]"));
+        assertTrue(maybeE.isPresent());
+
+        DeleteComposableIndexTemplateAction.Request req2 = new DeleteComposableIndexTemplateAction.Request("i*");
+        Exception e2 = expectThrows(Exception.class, () -> client().execute(DeleteComposableIndexTemplateAction.INSTANCE, req2).get());
+        maybeE = ExceptionsHelper.unwrapCausesAndSuppressed(e2, err ->
+            err.getMessage().contains("unable to remove composable templates [id] " +
+                "as they are in use by a data streams [metrics-foobar-baz, metrics-foobar-baz-eggplant]"));
+        assertTrue(maybeE.isPresent());
+    }
+
     private static void verifyResolvability(String dataStream, ActionRequestBuilder requestBuilder, boolean fail) {
         verifyResolvability(dataStream, requestBuilder, fail, 0);
     }

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

@@ -68,6 +68,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
@@ -609,6 +610,17 @@ public class MetadataIndexTemplateService {
             }
             throw new IndexTemplateMissingException(name);
         }
+
+        Optional<Set<String>> dataStreamsUsingTemplates = templateNames.stream()
+            .map(templateName -> dataStreamsUsingTemplate(currentState, templateName))
+            .reduce(Sets::union);
+        dataStreamsUsingTemplates.ifPresent(set -> {
+            if (set.size() > 0) {
+                throw new IllegalArgumentException("unable to remove composable templates " + new TreeSet<>(templateNames) +
+                    " as they are in use by a data streams " + new TreeSet<>(set));
+            }
+        });
+
         Metadata.Builder metadata = Metadata.builder(currentState.metadata());
         for (String templateName : templateNames) {
             logger.info("removing index template [{}]", templateName);
@@ -617,6 +629,18 @@ public class MetadataIndexTemplateService {
         return ClusterState.builder(currentState).metadata(metadata).build();
     }
 
+    static Set<String> dataStreamsUsingTemplate(final ClusterState state, final String templateName) {
+        final ComposableIndexTemplate template = state.metadata().templatesV2().get(templateName);
+        if (template == null) {
+            return Collections.emptySet();
+        }
+        final Set<String> dataStreams = state.metadata().dataStreams().keySet();
+        Set<String> matches = new HashSet<>();
+        template.indexPatterns().forEach(indexPattern ->
+            matches.addAll(dataStreams.stream().filter(stream -> Regex.simpleMatch(indexPattern, stream)).collect(Collectors.toList())));
+        return matches;
+    }
+
     public void putTemplate(final PutRequest request, final PutListener listener) {
         Settings.Builder updatedSettingsBuilder = Settings.builder();
         updatedSettingsBuilder.put(request.settings).normalizePrefix(IndexMetadata.INDEX_SETTING_PREFIX);