SNAPSHOT - Refactor for ingest service to handle reroute path checks

Author: lukewhiting

.idea/runConfigurations/Debug_Elasticsearch.xml

@@ -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(parameters.ingestService)),
+            entry(RerouteProcessor.TYPE, new RerouteProcessor.Factory()),
             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/IngestCommonPlugin.java

@@ -10,14 +10,10 @@
 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;
@@ -49,20 +45,14 @@ 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,
-        ClusterService clusterService,
-        ProjectId projectId,
-        StreamsPermissionsUtils streamsPermissionsUtils
+        String destination
     ) {
         super(tag, description);
         if (type.isEmpty()) {
@@ -81,16 +71,11 @@ 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.throwIfRerouteToSubstreamNotAllowed(projectMetadata, ingestDocument.getIndexHistory(), destination);
             ingestDocument.reroute(destination);
             return ingestDocument;
         }
@@ -186,12 +171,6 @@ 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,
@@ -233,17 +212,7 @@ 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,
-                ingestService.getClusterService(),
-                projectId,
-                StreamsPermissionsUtils.getInstance()
-            );
+            return new RerouteProcessor(tag, description, type, dataset, namespace, destination);
         }
     }
 

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

@@ -11,7 +11,6 @@
 
 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;
 
@@ -20,8 +19,6 @@
 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 {
 
@@ -78,7 +75,6 @@ private static RerouteProcessor create(String dataset, String namespace) throws
     }
 
     private static RerouteProcessor create(Map<String, Object> config) throws Exception {
-        IngestService ingestService = mock(IngestService.class, RETURNS_SMART_NULLS);
-        return new RerouteProcessor.Factory(ingestService).create(null, null, null, new HashMap<>(config), ProjectId.DEFAULT);
+        return new RerouteProcessor.Factory().create(null, null, null, new HashMap<>(config), ProjectId.DEFAULT);
     }
 }

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

@@ -9,45 +9,21 @@
 
 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).throwIfRerouteToSubstreamNotAllowed(any(), any(), anyString());
-    }
-
     public void testDefaults() throws Exception {
         IngestDocument ingestDocument = createIngestDocument("logs-generic-default");
 
@@ -315,25 +291,12 @@ 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,
-            clusterServiceMock,
-            ProjectId.DEFAULT,
-            streamsPermissionsUtilsMock
+            null
         );
     }
 
     private RerouteProcessor createRerouteProcessor(String destination) {
-        return new RerouteProcessor(
-            null,
-            null,
-            List.of(),
-            List.of(),
-            List.of(),
-            destination,
-            clusterServiceMock,
-            ProjectId.DEFAULT,
-            streamsPermissionsUtilsMock
-        );
+        return new RerouteProcessor(null, null, List.of(), List.of(), List.of(), destination);
     }
 
     private void assertDataSetFields(IngestDocument ingestDocument, String type, String dataset, String namespace) {

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

@@ -26,6 +26,7 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.ingest.DeletePipelineRequest;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
+import org.elasticsearch.action.ingest.ReservedPipelineAction;
 import org.elasticsearch.action.support.RefCountingRunnable;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
@@ -55,6 +56,8 @@
 import org.elasticsearch.common.logging.DeprecationLogger;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.streams.StreamType;
+import org.elasticsearch.common.streams.StreamsPermissionsUtils;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
@@ -75,6 +78,7 @@
 import org.elasticsearch.node.ReportingService;
 import org.elasticsearch.plugins.IngestPlugin;
 import org.elasticsearch.plugins.internal.XContentParserDecorator;
+import org.elasticsearch.script.Metadata;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.threadpool.Scheduler;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -154,6 +158,7 @@ public static boolean locallySupportedIngestFeature(NodeFeature nodeFeature) {
     private volatile ClusterState state;
     private final ProjectResolver projectResolver;
     private final FeatureService featureService;
+    private final StreamsPermissionsUtils streamsPermissionsUtils;
 
     private static BiFunction<Long, Runnable, Scheduler.ScheduledCancellable> createScheduler(ThreadPool threadPool) {
         return (delay, command) -> threadPool.schedule(command, TimeValue.timeValueMillis(delay), threadPool.generic());
@@ -241,7 +246,8 @@ public IngestService(
         MatcherWatchdog matcherWatchdog,
         FailureStoreMetrics failureStoreMetrics,
         ProjectResolver projectResolver,
-        FeatureService featureService
+        FeatureService featureService,
+        StreamsPermissionsUtils streamsPermissionsUtils
     ) {
         this.clusterService = clusterService;
         this.scriptService = scriptService;
@@ -265,6 +271,7 @@ public IngestService(
         this.failureStoreMetrics = failureStoreMetrics;
         this.projectResolver = projectResolver;
         this.featureService = featureService;
+        this.streamsPermissionsUtils = streamsPermissionsUtils;
     }
 
     /**
@@ -283,6 +290,7 @@ public IngestService(
         this.failureStoreMetrics = ingestService.failureStoreMetrics;
         this.projectResolver = ingestService.projectResolver;
         this.featureService = ingestService.featureService;
+        streamsPermissionsUtils = ingestService.streamsPermissionsUtils;
     }
 
     private static Map<String, Processor.Factory> processorFactories(List<IngestPlugin> ingestPlugins, Processor.Parameters parameters) {
@@ -301,15 +309,15 @@ private static Map<String, Processor.Factory> processorFactories(List<IngestPlug
 
     /**
      * Resolves the potential pipelines (default and final) from the requests or templates associated to the index and then **mutates**
-     * the {@link org.elasticsearch.action.index.IndexRequest} passed object with the pipeline information.
+     * the {@link IndexRequest} passed object with the pipeline information.
      * <p>
      * Also, this method marks the request as `isPipelinesResolved = true`: Due to the request could be rerouted from a coordinating node
      * to an ingest node, we have to be able to avoid double resolving the pipelines and also able to distinguish that either the pipeline
      * comes as part of the request or resolved from this method. All this is made to later be able to reject the request in case the
      * pipeline was set by a required pipeline **and** the request also has a pipeline request too.
      *
      * @param originalRequest Original write request received.
-     * @param indexRequest    The {@link org.elasticsearch.action.index.IndexRequest} object to update.
+     * @param indexRequest    The {@link IndexRequest} object to update.
      * @param projectMetadata Project metadata from the cluster state from where the pipeline information is derived.
      */
     public static void resolvePipelinesAndUpdateIndexRequest(
@@ -411,7 +419,7 @@ public void delete(ProjectId projectId, DeletePipelineRequest request, ActionLis
     }
 
     /**
-     * Used by this class and {@link org.elasticsearch.action.ingest.ReservedPipelineAction}
+     * Used by this class and {@link ReservedPipelineAction}
      */
     public static class DeletePipelineClusterStateUpdateTask extends PipelineClusterStateUpdateTask {
         private final DeletePipelineRequest request;
@@ -672,7 +680,7 @@ private static void collectProcessorMetrics(
     }
 
     /**
-     * Used in this class and externally by the {@link org.elasticsearch.action.ingest.ReservedPipelineAction}
+     * Used in this class and externally by the {@link ReservedPipelineAction}
      */
     public static class PutPipelineClusterStateUpdateTask extends PipelineClusterStateUpdateTask {
         private final PutPipelineRequest request;
@@ -683,7 +691,7 @@ public static class PutPipelineClusterStateUpdateTask extends PipelineClusterSta
         }
 
         /**
-         * Used by {@link org.elasticsearch.action.ingest.ReservedPipelineAction}
+         * Used by {@link ReservedPipelineAction}
          */
         public PutPipelineClusterStateUpdateTask(ProjectId projectId, PutPipelineRequest request) {
             this(projectId, null, request);
@@ -904,7 +912,7 @@ protected void doRun() {
                         final int slot = i;
                         final Releasable ref = refs.acquire();
                         final IngestDocument ingestDocument = newIngestDocument(indexRequest);
-                        final org.elasticsearch.script.Metadata originalDocumentMetadata = ingestDocument.getMetadata().clone();
+                        final Metadata originalDocumentMetadata = ingestDocument.getMetadata().clone();
                         // the document listener gives us three-way logic: a document can fail processing (1), or it can
                         // be successfully processed. a successfully processed document can be kept (2) or dropped (3).
                         final ActionListener<IngestPipelinesExecutionResult> documentListener = ActionListener.runAfter(
@@ -1198,6 +1206,31 @@ private void executePipelines(
                         return; // document failed!
                     }
 
+                    StreamsPermissionsUtils permissionsUtils = StreamsPermissionsUtils.getInstance();
+                    for (StreamType streamType : StreamType.values()) {
+                        if (permissionsUtils.streamTypeIsEnabled(streamType, project)) {
+                            if (newIndex.startsWith(streamType.getStreamName() + ".")
+                                && ingestDocument.getIndexHistory().stream().noneMatch(s -> s.equals(streamType.getStreamName()))) {
+                                exceptionHandler.accept(
+                                    new IngestPipelineException(
+                                        pipelineId,
+                                        new IllegalArgumentException(
+                                            format(
+                                                "Pipelines can't re-route documents to child streams, but pipeline [%s] tried to reroute "
+                                                    + "this document from index [%s] to index [%s]. Reroute history: %s",
+                                                pipelineId,
+                                                originalIndex,
+                                                newIndex,
+                                                String.join(" -> ", ingestDocument.getIndexHistory())
+                                            )
+                                        )
+                                    )
+                                );
+                                return; // document failed!
+                            }
+                        }
+                    }
+
                     // add the index to the document's index history, and check for cycles in the visited indices
                     boolean cycle = ingestDocument.updateIndexHistory(newIndex) == false;
                     if (cycle) {
@@ -1352,7 +1385,7 @@ private static IngestDocument newIngestDocument(final IndexRequest request) {
     /**
      * Updates an index request based on the metadata of an ingest document.
      */
-    private static void updateIndexRequestMetadata(final IndexRequest request, final org.elasticsearch.script.Metadata metadata) {
+    private static void updateIndexRequestMetadata(final IndexRequest request, final Metadata metadata) {
         // it's fine to set all metadata fields all the time, as ingest document holds their starting values
         // before ingestion, which might also get modified during ingestion.
         request.index(metadata.getIndex());

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -77,6 +77,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsModule;
+import org.elasticsearch.common.streams.StreamsPermissionsUtils;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.PageCacheRecycler;
 import org.elasticsearch.common.util.set.Sets;
@@ -716,7 +717,8 @@ private void construct(
             IngestService.createGrokThreadWatchdog(environment, threadPool),
             failureStoreMetrics,
             projectResolver,
-            featureService
+            featureService,
+            StreamsPermissionsUtils.getInstance()
         );
 
         SystemIndices systemIndices = createSystemIndices(settings);