Reroute code complete

Author: lukewhiting

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java

@@ -65,7 +65,7 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
             entry(RegisteredDomainProcessor.TYPE, new RegisteredDomainProcessor.Factory()),
             entry(RemoveProcessor.TYPE, new RemoveProcessor.Factory(parameters.scriptService)),
             entry(RenameProcessor.TYPE, new RenameProcessor.Factory(parameters.scriptService)),
-            entry(RerouteProcessor.TYPE, new RerouteProcessor.Factory()),
+            entry(RerouteProcessor.TYPE, new RerouteProcessor.Factory(parameters.ingestService)),
             entry(ScriptProcessor.TYPE, new ScriptProcessor.Factory(parameters.scriptService)),
             entry(SetProcessor.TYPE, new SetProcessor.Factory(parameters.scriptService)),
             entry(SortProcessor.TYPE, new SortProcessor.Factory()),

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RerouteProcessor.java

@@ -10,10 +10,14 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.streams.StreamsPermissionsUtils;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.ingest.Processor;
 
 import java.util.List;
@@ -45,14 +49,20 @@ public final class RerouteProcessor extends AbstractProcessor {
     private final List<DataStreamValueSource> dataset;
     private final List<DataStreamValueSource> namespace;
     private final String destination;
+    private final ClusterService clusterService;
+    private final ProjectId projectId;
+    private final StreamsPermissionsUtils streamsPermissionsUtils;
 
     RerouteProcessor(
         String tag,
         String description,
         List<DataStreamValueSource> type,
         List<DataStreamValueSource> dataset,
         List<DataStreamValueSource> namespace,
-        String destination
+        String destination,
+        ClusterService clusterService,
+        ProjectId projectId,
+        StreamsPermissionsUtils streamsPermissionsUtils
     ) {
         super(tag, description);
         if (type.isEmpty()) {
@@ -71,11 +81,16 @@ public final class RerouteProcessor extends AbstractProcessor {
             this.namespace = namespace;
         }
         this.destination = destination;
+        this.clusterService = clusterService;
+        this.projectId = projectId;
+        this.streamsPermissionsUtils = streamsPermissionsUtils;
     }
 
     @Override
     public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
         if (destination != null) {
+            ProjectMetadata projectMetadata = clusterService.state().projectState(projectId).metadata();
+            streamsPermissionsUtils.throwIfRetrouteToSubstreamNotAllowed(projectMetadata, ingestDocument.getIndexHistory(), destination);
             ingestDocument.reroute(destination);
             return ingestDocument;
         }
@@ -171,6 +186,12 @@ String getDestination() {
 
     public static final class Factory implements Processor.Factory {
 
+        private final IngestService ingestService;
+
+        public Factory(IngestService ingestService) {
+            this.ingestService = ingestService;
+        }
+
         @Override
         public RerouteProcessor create(
             Map<String, Processor.Factory> processorFactories,
@@ -212,7 +233,17 @@ public RerouteProcessor create(
                 throw newConfigurationException(TYPE, tag, "destination", "can only be set if type, dataset, and namespace are not set");
             }
 
-            return new RerouteProcessor(tag, description, type, dataset, namespace, destination);
+            return new RerouteProcessor(
+                tag,
+                description,
+                type,
+                dataset,
+                namespace,
+                destination,
+                ingestService.getClusterService(),
+                projectId,
+                StreamsPermissionsUtils.getInstance()
+            );
         }
     }
 

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

@@ -10,6 +10,8 @@
 package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.ingest.common.RerouteProcessor.DataStreamValueSource;
 import org.elasticsearch.test.ESTestCase;
 
@@ -18,6 +20,8 @@
 import java.util.Map;
 
 import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Answers.RETURNS_SMART_NULLS;
+import static org.mockito.Mockito.mock;
 
 public class RerouteProcessorFactoryTests extends ESTestCase {
 
@@ -74,6 +78,7 @@ 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);
+        IngestService ingestService = mock(IngestService.class, RETURNS_SMART_NULLS);
+        return new RerouteProcessor.Factory(ingestService).create(null, null, null, new HashMap<>(config), ProjectId.DEFAULT);
     }
 }

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

@@ -9,21 +9,45 @@
 
 package org.elasticsearch.ingest.common;
 
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.streams.StreamsPermissionsUtils;
 import org.elasticsearch.ingest.CompoundProcessor;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.ingest.TestProcessor;
 import org.elasticsearch.ingest.WrappingProcessor;
 import org.elasticsearch.test.ESTestCase;
+import org.junit.Before;
 
 import java.util.List;
 import java.util.Map;
 
 import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class RerouteProcessorTests extends ESTestCase {
 
+    private final StreamsPermissionsUtils streamsPermissionsUtilsMock = mock(StreamsPermissionsUtils.class);
+    private final ClusterService clusterServiceMock = mock(ClusterService.class, RETURNS_DEEP_STUBS);
+
+    @Before
+    public void setUpStreamsPermissionsUtils() {
+        ClusterState clusterState = ClusterState.builder(ClusterState.EMPTY_STATE)
+            .putProjectMetadata(ProjectMetadata.builder(ProjectId.DEFAULT).build())
+            .build();
+        when(clusterServiceMock.state()).thenReturn(clusterState);
+        doNothing().when(streamsPermissionsUtilsMock).throwIfRetrouteToSubstreamNotAllowed(any(), any(), anyString());
+    }
+
     public void testDefaults() throws Exception {
         IngestDocument ingestDocument = createIngestDocument("logs-generic-default");
 
@@ -291,12 +315,25 @@ private RerouteProcessor createRerouteProcessor(List<String> type, List<String>
             type.stream().map(RerouteProcessor.DataStreamValueSource::type).toList(),
             dataset.stream().map(RerouteProcessor.DataStreamValueSource::dataset).toList(),
             namespace.stream().map(RerouteProcessor.DataStreamValueSource::namespace).toList(),
-            null
+            null,
+            clusterServiceMock,
+            ProjectId.DEFAULT,
+            streamsPermissionsUtilsMock
         );
     }
 
     private RerouteProcessor createRerouteProcessor(String destination) {
-        return new RerouteProcessor(null, null, List.of(), List.of(), List.of(), destination);
+        return new RerouteProcessor(
+            null,
+            null,
+            List.of(),
+            List.of(),
+            List.of(),
+            destination,
+            clusterServiceMock,
+            ProjectId.DEFAULT,
+            streamsPermissionsUtilsMock
+        );
     }
 
     private void assertDataSetFields(IngestDocument ingestDocument, String type, String dataset, String namespace) {

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/common/streams/StreamTypes.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 StreamTypes {
+
+    LOGS("logs");
+
+    private final String streamName;
+
+    StreamTypes(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 throwIfRetrouteToSubstreamNotAllowed(ProjectMetadata projectMetadata, Set<String> indexHistory, String destination)
+        throws IllegalArgumentException {
+        for (StreamTypes streamType : StreamTypes.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(StreamTypes streamType, ProjectMetadata projectMetadata) {
+        StreamsMetadata metadata = projectMetadata.custom(StreamsMetadata.TYPE, StreamsMetadata.EMPTY);
+        return switch (streamType) {
+            case LOGS -> metadata.isLogsEnabled();
+        };
+    }
+
+}