Passing IngestService to processor factory with processor params

Signed-off-by: Martin Gaievski <gaievski@amazon.com>
opensearch-project · Oct 2, 2023 · cb9496d · cb9496d
1 parent 05ee0c3
commit cb9496d
Show file tree

Hide file tree

Showing 6 changed files with 35 additions and 18 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Introduce new dynamic cluster setting to control slice computation for concurrent segment search ([#9107](https://github.com/opensearch-project/OpenSearch/pull/9107))
 - Implement on behalf of token passing for extensions ([#8679](https://github.com/opensearch-project/OpenSearch/pull/8679))
 - Implement Visitor Design pattern in QueryBuilder to enable the capability to traverse through the complex QueryBuilder tree. ([#10110](https://github.com/opensearch-project/OpenSearch/pull/10110))
+- Allowing pipeline processors to access index maping info by passing ingest service ref as part of the processor factory parameters ([#X](https://github.com/opensearch-project/OpenSearch/pull/X))
 
 ### Dependencies
 - Bump `log4j-core` from 2.18.0 to 2.19.0

diff --git a/server/src/main/java/org/opensearch/ingest/IngestService.java b/server/src/main/java/org/opensearch/ingest/IngestService.java
@@ -72,6 +72,7 @@
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.VersionType;
 import org.opensearch.index.analysis.AnalysisRegistry;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.plugins.IngestPlugin;
 import org.opensearch.script.ScriptService;
 import org.opensearch.threadpool.ThreadPool;
@@ -128,7 +129,8 @@ public IngestService(
         ScriptService scriptService,
         AnalysisRegistry analysisRegistry,
         List<IngestPlugin> ingestPlugins,
-        Client client
+        Client client,
+        IndicesService indicesService
     ) {
         this.clusterService = clusterService;
         this.scriptService = scriptService;
@@ -143,7 +145,8 @@ public IngestService(
                 (delay, command) -> threadPool.schedule(command, TimeValue.timeValueMillis(delay), ThreadPool.Names.GENERIC),
                 this,
                 client,
-                threadPool.generic()::execute
+                threadPool.generic()::execute,
+                indicesService
             )
         );
         this.threadPool = threadPool;

diff --git a/server/src/main/java/org/opensearch/ingest/Processor.java b/server/src/main/java/org/opensearch/ingest/Processor.java
@@ -36,6 +36,7 @@
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.env.Environment;
 import org.opensearch.index.analysis.AnalysisRegistry;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.script.ScriptService;
 import org.opensearch.threadpool.Scheduler;
 
@@ -156,6 +157,8 @@ class Parameters {
          */
         public final Client client;
 
+        public final IndicesService indicesService;
+
         public Parameters(
             Environment env,
             ScriptService scriptService,
@@ -165,7 +168,8 @@ public Parameters(
             BiFunction<Long, Runnable, Scheduler.ScheduledCancellable> scheduler,
             IngestService ingestService,
             Client client,
-            Consumer<Runnable> genericExecutor
+            Consumer<Runnable> genericExecutor,
+            IndicesService indicesService
         ) {
             this.env = env;
             this.scriptService = scriptService;
@@ -176,6 +180,7 @@ public Parameters(
             this.ingestService = ingestService;
             this.client = client;
             this.genericExecutor = genericExecutor;
+            this.indicesService = indicesService;
         }
 
     }

diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java
@@ -602,15 +602,6 @@ protected Node(
 
             tracer = tracerFactory.getTracer();
             resourcesToClose.add(tracer::close);
-            final IngestService ingestService = new IngestService(
-                clusterService,
-                threadPool,
-                this.environment,
-                scriptService,
-                analysisModule.getAnalysisRegistry(),
-                pluginsService.filterPlugins(IngestPlugin.class),
-                client
-            );
 
             final ClusterInfoService clusterInfoService = newClusterInfoService(settings, clusterService, threadPool, client);
             final UsageService usageService = new UsageService();
@@ -793,6 +784,17 @@ protected Node(
                 remoteStoreStatsTrackerFactory
             );
 
+            final IngestService ingestService = new IngestService(
+                clusterService,
+                threadPool,
+                this.environment,
+                scriptService,
+                analysisModule.getAnalysisRegistry(),
+                pluginsService.filterPlugins(IngestPlugin.class),
+                client,
+                indicesService
+            );
+
             final AliasValidator aliasValidator = new AliasValidator();
 
             final ShardLimitValidator shardLimitValidator = new ShardLimitValidator(settings, clusterService, systemIndices);

diff --git a/server/src/test/java/org/opensearch/ingest/IngestServiceTests.java b/server/src/test/java/org/opensearch/ingest/IngestServiceTests.java
@@ -67,6 +67,7 @@
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.VersionType;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.plugins.IngestPlugin;
 import org.opensearch.script.MockScriptEngine;
 import org.opensearch.script.Script;
@@ -149,7 +150,8 @@ public void testIngestPlugin() {
             null,
             null,
             Collections.singletonList(DUMMY_PLUGIN),
-            client
+            client,
+            mock(IndicesService.class)
         );
         Map<String, Processor.Factory> factories = ingestService.getProcessorFactories();
         assertTrue(factories.containsKey("foo"));
@@ -167,7 +169,8 @@ public void testIngestPluginDuplicate() {
                 null,
                 null,
                 Arrays.asList(DUMMY_PLUGIN, DUMMY_PLUGIN),
-                client
+                client,
+                mock(IndicesService.class)
             )
         );
         assertTrue(e.getMessage(), e.getMessage().contains("already registered"));
@@ -182,7 +185,8 @@ public void testExecuteIndexPipelineDoesNotExist() {
             null,
             null,
             Collections.singletonList(DUMMY_PLUGIN),
-            client
+            client,
+            mock(IndicesService.class)
         );
         final IndexRequest indexRequest = new IndexRequest("_index").id("_id")
             .source(emptyMap())
@@ -1485,7 +1489,8 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
             null,
             null,
             Arrays.asList(testPlugin),
-            client
+            client,
+            mock(IndicesService.class)
         );
         ingestService.addIngestClusterStateListener(ingestClusterStateListener);
 
@@ -1702,7 +1707,7 @@ private static IngestService createWithProcessors(Map<String, Processor.Factory>
             public Map<String, Processor.Factory> getProcessors(final Processor.Parameters parameters) {
                 return processors;
             }
-        }), client);
+        }), client, mock(IndicesService.class));
     }
 
     private CompoundProcessor mockCompoundProcessor() {

diff --git a/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java b/server/src/test/java/org/opensearch/snapshots/SnapshotResiliencyTests.java
@@ -2202,7 +2202,8 @@ public void onFailure(final Exception e) {
                             scriptService,
                             new AnalysisModule(environment, Collections.emptyList()).getAnalysisRegistry(),
                             Collections.emptyList(),
-                            client
+                            client,
+                            indicesService
                         ),
                         transportShardBulkAction,
                         client,