New logic to check for conflicting indices before enabling streams plus tests

Author: lukewhiting

modules/streams/build.gradle

@@ -20,7 +20,7 @@ esplugin {
 
 restResources {
   restApi {
-    include '_common', 'streams'
+    include '_common', 'streams', 'indices'
   }
 }
 

modules/streams/src/main/java/org/elasticsearch/rest/streams/logs/TransportLogsStreamsToggleActivation.java

@@ -11,6 +11,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.AcknowledgedRequest;
@@ -22,16 +23,19 @@
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.metadata.StreamsMetadata;
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
+import java.util.Arrays;
 import java.util.Locale;
 
 /**
@@ -77,17 +81,34 @@ protected void masterOperation(
         ActionListener<AcknowledgedResponse> listener
     ) throws Exception {
         ProjectId projectId = projectResolver.getProjectId();
-        StreamsMetadata streamsState = state.metadata().getProject(projectId).custom(StreamsMetadata.TYPE, StreamsMetadata.EMPTY);
+        ProjectMetadata projectMetadata = state.metadata().getProject(projectId);
+        StreamsMetadata streamsState = projectMetadata.custom(StreamsMetadata.TYPE, StreamsMetadata.EMPTY);
         boolean currentlyEnabled = streamsState.isLogsEnabled();
         boolean shouldEnable = request.shouldEnable();
-        if (shouldEnable != currentlyEnabled) {
-            StreamsMetadataUpdateTask updateTask = new StreamsMetadataUpdateTask(request, listener, projectId, shouldEnable);
-            String taskName = String.format(Locale.ROOT, "enable-streams-logs-[%s]", shouldEnable ? "enable" : "disable");
-            taskQueue.submitTask(taskName, updateTask, updateTask.timeout());
-        } else {
+
+        if (shouldEnable == currentlyEnabled) {
             logger.debug("Logs streams are already in the requested state: {}", shouldEnable);
             listener.onResponse(AcknowledgedResponse.TRUE);
+            return;
+        }
+
+        if (shouldEnable && logsIndexExists(projectMetadata)) {
+            listener.onFailure(
+                new ElasticsearchStatusException(
+                    "Cannot enable logs streams: indices named 'logs' or starting with 'logs.' already exist.",
+                    RestStatus.CONFLICT
+                )
+            );
+            return;
         }
+
+        StreamsMetadataUpdateTask updateTask = new StreamsMetadataUpdateTask(request, listener, projectId, shouldEnable);
+        String taskName = String.format(Locale.ROOT, "enable-streams-logs-[%s]", shouldEnable ? "enable" : "disable");
+        taskQueue.submitTask(taskName, updateTask, updateTask.timeout());
+    }
+
+    private boolean logsIndexExists(ProjectMetadata projectMetadata) {
+        return Arrays.stream(projectMetadata.getConcreteAllIndices()).anyMatch(name -> name.equals("logs") || name.startsWith("logs."));
     }
 
     @Override

modules/streams/src/yamlRestTest/resources/rest-api-spec/test/streams/logs/10_basic.yml

@@ -41,3 +41,29 @@
   - do:
       streams.status: { }
   - is_true: logs.enabled
+
+---
+"Check streams can't be enabled with existing logs indices":
+  - do:
+      indices.create:
+        index: logs
+  - do:
+      catch: conflict
+      streams.logs_enable: { }
+  - match: { error.reason: "Cannot enable logs streams: indices named 'logs' or starting with 'logs.' already exist." }
+
+  - do:
+      indices.delete:
+        index: logs
+
+  - do:
+      indices.create:
+        index: logs.test
+  - do:
+      catch: conflict
+      streams.logs_enable: { }
+  - match: { error.reason: "Cannot enable logs streams: indices named 'logs' or starting with 'logs.' already exist." }
+
+  - do:
+      indices.delete:
+        index: logs.test