Add data stream template validation

to data stream promotion endpoint

Author: lukewhiting

modules/data-streams/src/main/java/org/elasticsearch/datastreams/action/PromoteDataStreamTransportAction.java

@@ -8,6 +8,8 @@
  */
 package org.elasticsearch.datastreams.action;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.datastreams.PromoteDataStreamAction;
@@ -23,6 +25,8 @@
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
+import org.elasticsearch.common.logging.HeaderWarning;
+import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.indices.SystemIndices;
@@ -31,9 +35,14 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import static org.elasticsearch.core.Strings.format;
+
 public class PromoteDataStreamTransportAction extends AcknowledgedTransportMasterNodeAction<PromoteDataStreamAction.Request> {
 
+    private static final Logger logger = LogManager.getLogger(PromoteDataStreamTransportAction.class);
+
     private final SystemIndices systemIndices;
+    private final ClusterService clusterService;
 
     @Inject
     public PromoteDataStreamTransportAction(
@@ -55,6 +64,7 @@ public PromoteDataStreamTransportAction(
             EsExecutors.DIRECT_EXECUTOR_SERVICE
         );
         this.systemIndices = systemIndices;
+        this.clusterService = clusterService;
     }
 
     @Override
@@ -76,7 +86,7 @@ public void onFailure(Exception e) {
 
                 @Override
                 public ClusterState execute(ClusterState currentState) {
-                    return promoteDataStream(currentState, request);
+                    return promoteDataStream(currentState, request, clusterService);
                 }
 
                 @Override
@@ -92,18 +102,48 @@ private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String
         clusterService.submitUnbatchedStateUpdateTask(source, task);
     }
 
-    static ClusterState promoteDataStream(ClusterState currentState, PromoteDataStreamAction.Request request) {
+    static ClusterState promoteDataStream(
+        ClusterState currentState,
+        PromoteDataStreamAction.Request request,
+        ClusterService clusterService
+    ) {
         DataStream dataStream = currentState.getMetadata().dataStreams().get(request.getName());
+
         if (dataStream == null) {
             throw new ResourceNotFoundException("data stream [" + request.getName() + "] does not exist");
         }
 
+        warnIfTemplateMissingForDatastream(dataStream, clusterService);
+
         DataStream promotedDataStream = dataStream.promoteDataStream();
         Metadata.Builder metadata = Metadata.builder(currentState.metadata());
         metadata.put(promotedDataStream);
         return ClusterState.builder(currentState).metadata(metadata).build();
     }
 
+    private static void warnIfTemplateMissingForDatastream(DataStream dataStream, ClusterService clusterService) {
+        var datastreamName = dataStream.getName();
+
+        var matchingIndex = clusterService.state()
+            .metadata()
+            .templatesV2()
+            .values()
+            .stream()
+            .filter(cit -> cit.getDataStreamTemplate() != null)
+            .flatMap(cit -> cit.indexPatterns().stream())
+            .anyMatch(pattern -> Regex.simpleMatch(pattern, datastreamName));
+
+        if (matchingIndex == false) {
+            String warningMessage = format(
+                "Data stream [%s] does not have a matching index template. This will cause rollover to fail until a matching index "
+                    + "template is created",
+                datastreamName
+            );
+            logger.warn(() -> warningMessage);
+            HeaderWarning.addWarning(warningMessage);
+        }
+    }
+
     @Override
     protected ClusterBlockException checkBlock(PromoteDataStreamAction.Request request, ClusterState state) {
         return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);