Make enrich project-aware (#124099)

Makes the execution and use of enrich policies project-aware.
Note: this does not make the enrich cache project-aware. That is to be
handled in a follow-up PR.

Author: nielsbauman

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/ingest/PipelineBridge.java

@@ -8,6 +8,7 @@
  */
 package org.elasticsearch.logstashbridge.ingest;
 
+import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.ingest.Pipeline;
 import org.elasticsearch.logstashbridge.StableBridgeAPI;
 import org.elasticsearch.logstashbridge.script.ScriptServiceBridge;
@@ -20,14 +21,21 @@ public static PipelineBridge wrap(final Pipeline pipeline) {
         return new PipelineBridge(pipeline);
     }
 
+    @FixForMultiProject(description = "should we pass a non-null project ID here?")
     public static PipelineBridge create(
         String id,
         Map<String, Object> config,
         Map<String, ProcessorBridge.Factory> processorFactories,
         ScriptServiceBridge scriptServiceBridge
     ) throws Exception {
         return wrap(
-            Pipeline.create(id, config, StableBridgeAPI.unwrap(processorFactories), StableBridgeAPI.unwrapNullable(scriptServiceBridge))
+            Pipeline.create(
+                id,
+                config,
+                StableBridgeAPI.unwrap(processorFactories),
+                StableBridgeAPI.unwrapNullable(scriptServiceBridge),
+                null
+            )
         );
     }
 

libs/logstash-bridge/src/main/java/org/elasticsearch/logstashbridge/ingest/ProcessorBridge.java

@@ -8,6 +8,7 @@
  */
 package org.elasticsearch.logstashbridge.ingest;
 
+import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.ingest.Processor;
@@ -118,7 +119,7 @@ static Factory wrap(final Processor.Factory delegate) {
         @Override
         default Processor.Factory unwrap() {
             final Factory stableAPIFactory = this;
-            return (registry, tag, description, config) -> stableAPIFactory.create(
+            return (registry, tag, description, config, projectId) -> stableAPIFactory.create(
                 StableBridgeAPI.wrap(registry, Factory::wrap),
                 tag,
                 description,
@@ -131,14 +132,17 @@ private Wrapped(final Processor.Factory delegate) {
                 super(delegate);
             }
 
+            @FixForMultiProject(description = "should we pass a non-null project ID here?")
             @Override
             public ProcessorBridge create(
                 final Map<String, Factory> registry,
                 final String processorTag,
                 final String description,
                 final Map<String, Object> config
             ) throws Exception {
-                return ProcessorBridge.wrap(this.delegate.create(StableBridgeAPI.unwrap(registry), processorTag, description, config));
+                return ProcessorBridge.wrap(
+                    this.delegate.create(StableBridgeAPI.unwrap(registry), processorTag, description, config, null)
+                );
             }
 
             @Override

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/IngestFailureStoreMetricsIT.java

@@ -403,9 +403,9 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
             Map<String, Processor.Factory> processors = new HashMap<>();
             processors.put(
                 "drop",
-                (factories, tag, description, config) -> new TestProcessor(tag, "drop", description, ingestDocument -> null)
+                (factories, tag, description, config, projectId) -> new TestProcessor(tag, "drop", description, ingestDocument -> null)
             );
-            processors.put("reroute", (factories, tag, description, config) -> {
+            processors.put("reroute", (factories, tag, description, config, projectId) -> {
                 String destination = (String) config.remove("destination");
                 return new TestProcessor(
                     tag,
@@ -416,7 +416,12 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
             });
             processors.put(
                 "fail",
-                (processorFactories, tag, description, config) -> new TestProcessor(tag, "fail", description, new RuntimeException())
+                (processorFactories, tag, description, config, projectId) -> new TestProcessor(
+                    tag,
+                    "fail",
+                    description,
+                    new RuntimeException()
+                )
             );
             return processors;
         }

modules/ingest-attachment/src/main/java/org/elasticsearch/ingest/attachment/AttachmentProcessor.java

@@ -15,6 +15,7 @@
 import org.apache.tika.metadata.Office;
 import org.apache.tika.metadata.TikaCoreProperties;
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.logging.DeprecationCategory;
 import org.elasticsearch.common.logging.DeprecationLogger;
@@ -232,7 +233,8 @@ public AttachmentProcessor create(
             Map<String, Processor.Factory> registry,
             String processorTag,
             String description,
-            Map<String, Object> config
+            Map<String, Object> config,
+            ProjectId projectId
         ) {
             String field = readStringProperty(TYPE, processorTag, config, "field");
             String resourceName = readOptionalStringProperty(TYPE, processorTag, config, "resource_name");

modules/ingest-attachment/src/test/java/org/elasticsearch/ingest/attachment/AttachmentProcessorFactoryTests.java

@@ -36,7 +36,7 @@ public void testBuildDefaults() throws Exception {
 
         String processorTag = randomAlphaOfLength(10);
 
-        AttachmentProcessor processor = factory.create(null, processorTag, null, config);
+        AttachmentProcessor processor = factory.create(null, processorTag, null, config, null);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("attachment"));
@@ -57,7 +57,7 @@ public void testConfigureIndexedChars() throws Exception {
         config.put("indexed_chars", indexedChars);
 
         String processorTag = randomAlphaOfLength(10);
-        AttachmentProcessor processor = factory.create(null, processorTag, null, config);
+        AttachmentProcessor processor = factory.create(null, processorTag, null, config, null);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getIndexedChars(), is(indexedChars));
         assertFalse(processor.isIgnoreMissing());
@@ -73,7 +73,7 @@ public void testBuildTargetField() throws Exception {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("target_field", "_field");
-        AttachmentProcessor processor = factory.create(null, null, null, config);
+        AttachmentProcessor processor = factory.create(null, null, null, config, null);
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("_field"));
         assertFalse(processor.isIgnoreMissing());
@@ -97,7 +97,7 @@ public void testBuildFields() throws Exception {
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
         config.put("properties", fieldNames);
-        AttachmentProcessor processor = factory.create(null, null, null, config);
+        AttachmentProcessor processor = factory.create(null, null, null, config, null);
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getProperties(), equalTo(properties));
         assertFalse(processor.isIgnoreMissing());
@@ -114,7 +114,7 @@ public void testBuildIllegalFieldOption() throws Exception {
         config.put("field", "_field");
         config.put("properties", Collections.singletonList("invalid"));
         try {
-            factory.create(null, null, null, config);
+            factory.create(null, null, null, config, null);
             fail("exception expected");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), containsString("[properties] illegal field option [invalid]"));
@@ -128,7 +128,7 @@ public void testBuildIllegalFieldOption() throws Exception {
         config.put("field", "_field");
         config.put("properties", "invalid");
         try {
-            factory.create(null, null, null, config);
+            factory.create(null, null, null, config, null);
             fail("exception expected");
         } catch (ElasticsearchParseException e) {
             assertThat(e.getMessage(), equalTo("[properties] property isn't a list, but of type [java.lang.String]"));
@@ -148,7 +148,7 @@ public void testIgnoreMissing() throws Exception {
 
         String processorTag = randomAlphaOfLength(10);
 
-        AttachmentProcessor processor = factory.create(null, processorTag, null, config);
+        AttachmentProcessor processor = factory.create(null, processorTag, null, config, null);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("attachment"));
@@ -169,7 +169,7 @@ public void testRemoveBinary() throws Exception {
 
         String processorTag = randomAlphaOfLength(10);
 
-        AttachmentProcessor processor = factory.create(null, processorTag, null, config);
+        AttachmentProcessor processor = factory.create(null, processorTag, null, config, null);
         assertThat(processor.getTag(), equalTo(processorTag));
         assertThat(processor.getField(), equalTo("_field"));
         assertThat(processor.getTargetField(), equalTo("attachment"));

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest.common;
 
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
@@ -108,7 +109,8 @@ public AbstractStringProcessor<?> create(
             Map<String, Processor.Factory> registry,
             String tag,
             String description,
-            Map<String, Object> config
+            Map<String, Object> config,
+            ProjectId projectId
         ) throws Exception {
             String field = ConfigurationUtils.readStringProperty(processorType, tag, config, "field");
             boolean ignoreMissing = ConfigurationUtils.readBooleanProperty(processorType, tag, config, "ignore_missing", false);

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest.common;
 
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
@@ -73,7 +74,8 @@ public AppendProcessor create(
             Map<String, Processor.Factory> registry,
             String processorTag,
             String description,
-            Map<String, Object> config
+            Map<String, Object> config,
+            ProjectId projectId
         ) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest.common;
 
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
@@ -297,7 +298,8 @@ public CommunityIdProcessor create(
             Map<String, Processor.Factory> registry,
             String processorTag,
             String description,
-            Map<String, Object> config
+            Map<String, Object> config,
+            ProjectId projectId
         ) throws Exception {
             String sourceIpField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "source_ip", DEFAULT_SOURCE_IP);
             String sourcePortField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "source_port", DEFAULT_SOURCE_PORT);

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest.common;
 
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
@@ -216,7 +217,8 @@ public ConvertProcessor create(
             Map<String, Processor.Factory> registry,
             String processorTag,
             String description,
-            Map<String, Object> config
+            Map<String, Object> config,
+            ProjectId projectId
         ) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             String typeProperty = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "type");

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest.common;
 
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
@@ -93,7 +94,8 @@ public CsvProcessor create(
             Map<String, Processor.Factory> registry,
             String processorTag,
             String description,
-            Map<String, Object> config
+            Map<String, Object> config,
+            ProjectId projectId
         ) {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             String quote = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "quote", "\"");