Transport action changes for streams write restrictions

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RerouteProcessorFactoryTests.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.ingest.common.RerouteProcessor.DataStreamValueSource;
 import org.elasticsearch.test.ESTestCase;
 
@@ -74,6 +75,6 @@ private static RerouteProcessor create(String dataset, String namespace) throws
     }
 
     private static RerouteProcessor create(Map<String, Object> config) throws Exception {
-        return new RerouteProcessor.Factory().create(null, null, null, new HashMap<>(config), null);
+        return new RerouteProcessor.Factory().create(null, null, null, new HashMap<>(config), ProjectId.DEFAULT);
     }
 }

modules/streams/build.gradle

@@ -20,7 +20,7 @@ esplugin {
 
 restResources {
   restApi {
-    include '_common', 'streams'
+    include '_common', 'streams', "bulk", "index", "ingest", "indices"
   }
 }
 
@@ -38,4 +38,5 @@ artifacts {
 
 dependencies {
   testImplementation project(path: ':test:test-clusters')
+  clusterModules project(':modules:ingest-common')
 }

modules/streams/src/yamlRestTest/java/org/elasticsearch/streams/StreamsYamlTestSuiteIT.java

@@ -29,7 +29,11 @@ public static Iterable<Object[]> parameters() throws Exception {
     }
 
     @ClassRule
-    public static ElasticsearchCluster cluster = ElasticsearchCluster.local().module("streams").feature(FeatureFlag.LOGS_STREAM).build();
+    public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
+        .module("streams")
+        .module("ingest-common")
+        .feature(FeatureFlag.LOGS_STREAM)
+        .build();
 
     @Override
     protected String getTestRestCluster() {

modules/streams/src/yamlRestTest/resources/rest-api-spec/test/streams/logs/20_substream_restrictions.yml

@@ -0,0 +1,72 @@
+---
+"Check User Can't Write To Substream Directly":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      streams.status: { }
+  - is_true: logs.enabled
+
+  - do:
+      bulk:
+        body: |
+          { "index": { "_index": "logs.foo" } }
+          { "foo": "bar" }
+  - match: { errors: true }
+  - match: { items.0.index.status: 400 }
+  - match: { items.0.index.error.type: "illegal_argument_exception" }
+  - match: { items.0.index.error.reason: "Writes to child stream [logs.foo] are not allowed, use the parent stream instead: [logs]" }
+
+---
+"Check User Can't Write To Substream Directly With Single Doc":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      streams.status: { }
+  - is_true: logs.enabled
+
+  - do:
+      catch: bad_request
+      index:
+        index: logs.foo
+        id: "1"
+        body:
+          foo: bar
+  - match: { error.type: "illegal_argument_exception" }
+  - match: { error.reason: "Writes to child stream [logs.foo] are not allowed, use the parent stream instead: [logs]" }
+
+---
+"Check Bulk Index With Reroute Processor To Substream Is Rejected":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      streams.status: { }
+  - is_true: logs.enabled
+
+  - do:
+      ingest.put_pipeline:
+        id: "reroute-to-logs-foo"
+        body:
+          processors:
+            - reroute:
+                destination: "logs.foo"
+  - do:
+      indices.create:
+        index: "bad-index"
+        body:
+          settings:
+            index.default_pipeline: "reroute-to-logs-foo"
+  - do:
+      bulk:
+        body: |
+          { "index": { "_index": "bad-index" } }
+          { "foo": "bar" }
+  - match: { errors: true }
+  - match: { items.0.index.status: 400 }
+  - match: { items.0.index.error.type: "illegal_argument_exception" }
+  - match: { items.0.index.error.reason: "Cannot reroute to substream [logs.foo] as only the stream itself can reroute to substreams. Please reroute to the stream [logs] instead." }

server/src/main/java/module-info.java

@@ -213,6 +213,7 @@
     exports org.elasticsearch.common.regex;
     exports org.elasticsearch.common.scheduler;
     exports org.elasticsearch.common.settings;
+    exports org.elasticsearch.common.streams;
     exports org.elasticsearch.common.text;
     exports org.elasticsearch.common.time;
     exports org.elasticsearch.common.transport;

server/src/main/java/org/elasticsearch/action/bulk/BulkOperation.java

@@ -43,6 +43,7 @@
 import org.elasticsearch.cluster.routing.IndexRouting;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.collect.Iterators;
+import org.elasticsearch.common.streams.StreamsPermissionsUtils;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
@@ -104,6 +105,7 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
     private final FailureStoreMetrics failureStoreMetrics;
     private final DataStreamFailureStoreSettings dataStreamFailureStoreSettings;
     private final boolean clusterHasFailureStoreFeature;
+    private final StreamsPermissionsUtils streamsPermissionsUtils;
 
     BulkOperation(
         Task task,
@@ -139,7 +141,8 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
             new FailureStoreDocumentConverter(),
             failureStoreMetrics,
             dataStreamFailureStoreSettings,
-            clusterHasFailureStoreFeature
+            clusterHasFailureStoreFeature,
+            StreamsPermissionsUtils.getInstance()
         );
     }
 
@@ -160,7 +163,8 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
         FailureStoreDocumentConverter failureStoreDocumentConverter,
         FailureStoreMetrics failureStoreMetrics,
         DataStreamFailureStoreSettings dataStreamFailureStoreSettings,
-        boolean clusterHasFailureStoreFeature
+        boolean clusterHasFailureStoreFeature,
+        StreamsPermissionsUtils streamsPermissionsUtils
     ) {
         super(listener);
         this.task = task;
@@ -182,6 +186,7 @@ final class BulkOperation extends ActionRunnable<BulkResponse> {
         this.failureStoreMetrics = failureStoreMetrics;
         this.dataStreamFailureStoreSettings = dataStreamFailureStoreSettings;
         this.clusterHasFailureStoreFeature = clusterHasFailureStoreFeature;
+        this.streamsPermissionsUtils = streamsPermissionsUtils;
     }
 
     @Override
@@ -274,6 +279,30 @@ private long buildTookInMillis(long startTimeNanos) {
     }
 
     private Map<ShardId, List<BulkItemRequest>> groupBulkRequestsByShards(ClusterState clusterState) {
+        // ProjectMetadata projectMetadata = projectResolver.getProjectMetadata(clusterState);
+        //
+        // Set<StreamType> enabledStreamTypes = Arrays.stream(StreamType.values())
+        // .filter(t -> streamsPermissionsUtils.streamTypeIsEnabled(t, projectMetadata))
+        // .collect(Collectors.toCollection(() -> EnumSet.noneOf(StreamType.class)));
+        //
+        // for (StreamType streamType : enabledStreamTypes) {
+        // for (int i = 0; i < bulkRequest.requests.size(); i++) {
+        // DocWriteRequest<?> req = bulkRequest.requests.get(i);
+        // String prefix = streamType.getStreamName() + ".";
+        // if (req != null && req.index().startsWith(prefix)) {
+        // IllegalArgumentException exception = new IllegalArgumentException(
+        // "Writes to child stream ["
+        // + req.index()
+        // + "] are not allowed, use the parent stream instead: ["
+        // + streamType.getStreamName()
+        // + "]"
+        // );
+        // IndexDocFailureStoreStatus failureStoreStatus = processFailure(new BulkItemRequest(i, req), projectMetadata, exception);
+        // addFailureAndDiscardRequest(req, i, req.index(), exception, failureStoreStatus);
+        // }
+        // }
+        // }
+
         return groupRequestsByShards(
             clusterState,
             Iterators.enumerate(bulkRequest.requests.iterator(), BulkItemRequest::new),

server/src/main/java/org/elasticsearch/action/bulk/TransportAbstractBulkAction.java

@@ -32,6 +32,8 @@
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.streams.StreamType;
+import org.elasticsearch.common.streams.StreamsPermissionsUtils;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Assertions;
 import org.elasticsearch.core.FixForMultiProject;
@@ -46,12 +48,16 @@
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.function.LongSupplier;
+import java.util.stream.Collectors;
 
 /**
  * This is an abstract base class for bulk actions. It traverses all indices that the request gets routed to, executes all applicable
@@ -400,6 +406,31 @@ private void applyPipelinesAndDoInternalExecute(
         ActionListener<BulkResponse> listener
     ) throws IOException {
         final long relativeStartTimeNanos = relativeTimeNanos();
+
+        // Validate child stream writes before processing pipelines
+        ProjectMetadata projectMetadata = projectResolver.getProjectMetadata(clusterService.state());
+        Set<StreamType> enabledStreamTypes = Arrays.stream(StreamType.values())
+            .filter(t -> StreamsPermissionsUtils.getInstance().streamTypeIsEnabled(t, projectMetadata))
+            .collect(Collectors.toCollection(() -> EnumSet.noneOf(StreamType.class)));
+
+        BulkRequestModifier bulkRequestModifier = new BulkRequestModifier(bulkRequest);
+
+        for (StreamType streamType : enabledStreamTypes) {
+            for (int i = 0; i < bulkRequest.requests.size(); i++) {
+                DocWriteRequest<?> req = bulkRequest.requests.get(i);
+                String prefix = streamType.getStreamName() + ".";
+                if (req != null && req.index() != null && req.index().startsWith(prefix)) {
+                    IllegalArgumentException e = new IllegalArgumentException("Can't write to child stream");
+                    Boolean failureStore = resolveFailureStore(req.index(), projectMetadata, threadPool.absoluteTimeInMillis());
+                    if (failureStore != null && failureStore) {
+                        bulkRequestModifier.markItemForFailureStore(i, req.index(), e);
+                    } else {
+                        bulkRequestModifier.markItemAsFailed(i, e, IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN);
+                    }
+                }
+            }
+        }
+
         if (applyPipelines(task, bulkRequest, executor, listener) == false) {
             doInternalExecute(task, bulkRequest, executor, listener, relativeStartTimeNanos);
         }

server/src/main/java/org/elasticsearch/common/streams/StreamType.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.common.streams;
+
+public enum StreamType {
+
+    LOGS("logs");
+
+    private final String streamName;
+
+    StreamType(String streamName) {
+        this.streamName = streamName;
+    }
+
+    public String getStreamName() {
+        return streamName;
+    }
+}

server/src/main/java/org/elasticsearch/common/streams/StreamsPermissionsUtils.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.common.streams;
+
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.cluster.metadata.StreamsMetadata;
+
+import java.util.Set;
+
+public class StreamsPermissionsUtils {
+
+    private static volatile StreamsPermissionsUtils INSTANCE = null;
+
+    // Visible for testing only
+    StreamsPermissionsUtils() {}
+
+    public static StreamsPermissionsUtils getInstance() {
+        if (INSTANCE == null) {
+            synchronized (StreamsPermissionsUtils.class) {
+                if (INSTANCE == null) {
+                    INSTANCE = new StreamsPermissionsUtils();
+                }
+            }
+        }
+        return INSTANCE;
+    }
+
+    public void throwIfRerouteToSubstreamNotAllowed(ProjectMetadata projectMetadata, Set<String> indexHistory, String destination)
+        throws IllegalArgumentException {
+        for (StreamType streamType : StreamType.values()) {
+            String streamName = streamType.getStreamName();
+            if (streamTypeIsEnabled(streamType, projectMetadata)
+                && destination.startsWith(streamName + ".")
+                && indexHistory.contains(streamName) == false) {
+                throw new IllegalArgumentException(
+                    "Cannot reroute to substream ["
+                        + destination
+                        + "] as only the stream itself can reroute to substreams. "
+                        + "Please reroute to the stream ["
+                        + streamName
+                        + "] instead."
+                );
+            }
+        }
+    }
+
+    public boolean streamTypeIsEnabled(StreamType streamType, ProjectMetadata projectMetadata) {
+        StreamsMetadata metadata = projectMetadata.custom(StreamsMetadata.TYPE, StreamsMetadata.EMPTY);
+        return switch (streamType) {
+            case LOGS -> metadata.isLogsEnabled();
+        };
+    }
+
+}