Pass index settings to system ingest processor factories. (opensearch-project#18708)

Signed-off-by: Bo Zhang <bzhangam@amazon.com>

Author: bzhangam

CHANGELOG.md

@@ -23,6 +23,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support for non-timing info in profiler ([#18460](https://github.com/opensearch-project/OpenSearch/issues/18460))
 - Extend Approximation Framework to other numeric types ([#18530](https://github.com/opensearch-project/OpenSearch/issues/18530))
 - Add Semantic Version field type mapper and extensive unit tests([#18454](https://github.com/opensearch-project/OpenSearch/pull/18454))
+- Pass index settings to system ingest processor factories. ([#18708](https://github.com/opensearch-project/OpenSearch/pull/18708))
 
 ### Changed
 - Update Subject interface to use CheckedRunnable ([#18570](https://github.com/opensearch-project/OpenSearch/issues/18570))

plugins/examples/system-ingest-processor/src/main/java/org/opensearch/example/systemingestprocessor/ExampleSystemIngestProcessorFactory.java

@@ -8,15 +8,20 @@
 
 package org.opensearch.example.systemingestprocessor;
 
+import org.opensearch.common.settings.Settings;
 import org.opensearch.ingest.AbstractBatchingSystemProcessor;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.opensearch.example.systemingestprocessor.ExampleSystemIngestProcessorPlugin.TRIGGER_SETTING;
 import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_MAPPINGS;
+import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_SETTINGS;
 import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_TEMPLATE_MAPPINGS;
+import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_TEMPLATE_SETTINGS;
 
 /**
  * A factory to create the example system ingest processor
@@ -55,17 +60,26 @@ protected ExampleSystemIngestProcessorFactory() {
     @Override
     protected AbstractBatchingSystemProcessor newProcessor(String tag, String description, Map<String, Object> config) {
         final List<Map<String, Object>> mappings = new ArrayList<>();
+        final List<Settings> settings = new ArrayList<>();
         final Object mappingFromIndex = config.get(INDEX_MAPPINGS);
         final Object mappingFromTemplates = config.get(INDEX_TEMPLATE_MAPPINGS);
+        final Object settingsFromIndex = config.get(INDEX_SETTINGS);
+        final Object settingsFromTemplates = config.get(INDEX_TEMPLATE_SETTINGS);
         if (mappingFromTemplates instanceof List) {
             mappings.addAll((List<Map<String, Object>>) mappingFromTemplates);
         }
         if (mappingFromIndex instanceof Map) {
             mappings.add((Map<String, Object>) mappingFromIndex);
         }
+        if (settingsFromTemplates instanceof List) {
+            settings.addAll((Collection<? extends Settings>) settingsFromTemplates);
+        }
+        if (settingsFromIndex instanceof Settings) {
+            settings.add((Settings) settingsFromIndex);
+        }
 
         // If no config we are not able to create a processor so simply return a null to show no processor created
-        if (mappings.isEmpty()) {
+        if (mappings.isEmpty() && settings.isEmpty()) {
             return null;
         }
 
@@ -87,6 +101,15 @@ protected AbstractBatchingSystemProcessor newProcessor(String tag, String descri
             }
         }
 
+        // If the trigger setting is configured then use it directly.
+        // When we rely on the v1 template to create the index there can be multiple settings and the later one can
+        // override the previous one so we need to loop through all the settings.
+        for (final Settings setting : settings) {
+            if (setting.hasValue(TRIGGER_SETTING.getKey())) {
+                isTriggerFieldFound = TRIGGER_SETTING.get(setting);
+            }
+        }
+
         return isTriggerFieldFound ? new ExampleSystemIngestProcessor(tag, description, DEFAULT_BATCH_SIZE) : null;
     }
 

plugins/examples/system-ingest-processor/src/main/java/org/opensearch/example/systemingestprocessor/ExampleSystemIngestProcessorPlugin.java

@@ -8,10 +8,12 @@
 
 package org.opensearch.example.systemingestprocessor;
 
+import org.opensearch.common.settings.Setting;
 import org.opensearch.ingest.Processor;
 import org.opensearch.plugins.IngestPlugin;
 import org.opensearch.plugins.Plugin;
 
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -23,8 +25,23 @@ public class ExampleSystemIngestProcessorPlugin extends Plugin implements Ingest
      */
     public ExampleSystemIngestProcessorPlugin() {}
 
+    /**
+     * A custom index setting which is used to control if we should create the example system ingest processor.
+     */
+    public static final Setting<Boolean> TRIGGER_SETTING = Setting.boolSetting(
+        "index.example_system_ingest_processor_plugin.trigger_setting",
+        false,
+        Setting.Property.IndexScope,
+        Setting.Property.Dynamic
+    );
+
     @Override
     public Map<String, Processor.Factory> getSystemIngestProcessors(Processor.Parameters parameters) {
         return Map.of(ExampleSystemIngestProcessorFactory.TYPE, new ExampleSystemIngestProcessorFactory());
     }
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        return List.of(TRIGGER_SETTING);
+    }
 }

plugins/examples/system-ingest-processor/src/test/java/org/opensearch/example/systemingestprocessor/ExampleSystemIngestProcessorFactoryTests.java

@@ -8,6 +8,7 @@
 
 package org.opensearch.example.systemingestprocessor;
 
+import org.opensearch.common.settings.Settings;
 import org.opensearch.ingest.AbstractBatchingSystemProcessor;
 import org.opensearch.test.OpenSearchTestCase;
 
@@ -18,8 +19,11 @@
 import static org.opensearch.example.systemingestprocessor.ExampleSystemIngestProcessorFactory.DOC;
 import static org.opensearch.example.systemingestprocessor.ExampleSystemIngestProcessorFactory.PROPERTIES;
 import static org.opensearch.example.systemingestprocessor.ExampleSystemIngestProcessorFactory.TRIGGER_FIELD_NAME;
+import static org.opensearch.example.systemingestprocessor.ExampleSystemIngestProcessorPlugin.TRIGGER_SETTING;
 import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_MAPPINGS;
+import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_SETTINGS;
 import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_TEMPLATE_MAPPINGS;
+import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_TEMPLATE_SETTINGS;
 
 public class ExampleSystemIngestProcessorFactoryTests extends OpenSearchTestCase {
     public void testNewProcessor_whenWithTriggerField_thenReturnProcessor() {
@@ -55,4 +59,37 @@ public void testNewProcessor_whenNoMapping_thenReturnNull() {
 
         assertNull("Should not create an example system ingest processor when the mapping is not found.", processor);
     }
+
+    public void testNewProcessor_whenWithTriggerSettingFromIndex_thenReturnProcessor() {
+        final ExampleSystemIngestProcessorFactory factory = new ExampleSystemIngestProcessorFactory();
+        Settings triggerEnabled = Settings.builder().put(TRIGGER_SETTING.getKey(), true).build();
+
+        AbstractBatchingSystemProcessor processor = factory.newProcessor("tag", "description", Map.of(INDEX_SETTINGS, triggerEnabled));
+
+        assertNotNull("Should create an example system ingest processor when the trigger_setting is true.", processor);
+        assertTrue(processor instanceof ExampleSystemIngestProcessor);
+    }
+
+    public void testNewProcessor_whenWithTriggerSettingFromTemplate_thenReturnProcessor() {
+        final ExampleSystemIngestProcessorFactory factory = new ExampleSystemIngestProcessorFactory();
+        Settings triggerEnabled = Settings.builder().put(TRIGGER_SETTING.getKey(), true).build();
+
+        AbstractBatchingSystemProcessor processor = factory.newProcessor(
+            "tag",
+            "description",
+            Map.of(INDEX_TEMPLATE_SETTINGS, List.of(triggerEnabled))
+        );
+
+        assertNotNull("Should create an example system ingest processor when the trigger_setting is true.", processor);
+        assertTrue(processor instanceof ExampleSystemIngestProcessor);
+    }
+
+    public void testNewProcessor_whenWithTriggerSettingDisabled_thenReturnProcessor() {
+        final ExampleSystemIngestProcessorFactory factory = new ExampleSystemIngestProcessorFactory();
+        Settings triggerDisabled = Settings.builder().put(TRIGGER_SETTING.getKey(), false).build();
+
+        AbstractBatchingSystemProcessor processor = factory.newProcessor("tag", "description", Map.of(INDEX_SETTINGS, triggerDisabled));
+
+        assertNull("Should not create an example system ingest processor when the trigger_setting is false.", processor);
+    }
 }

plugins/examples/system-ingest-processor/src/test/java/org/opensearch/example/systemingestprocessor/ExampleSystemIngestProcessorPluginTests.java

@@ -11,8 +11,10 @@
 import org.opensearch.ingest.Processor;
 import org.opensearch.test.OpenSearchTestCase;
 
+import java.util.List;
 import java.util.Map;
 
+import static org.opensearch.example.systemingestprocessor.ExampleSystemIngestProcessorPlugin.TRIGGER_SETTING;
 import static org.mockito.Mockito.mock;
 
 public class ExampleSystemIngestProcessorPluginTests extends OpenSearchTestCase {
@@ -27,4 +29,8 @@ public void testGetSystemIngestProcessors() {
             factories.get(ExampleSystemIngestProcessorFactory.TYPE) instanceof ExampleSystemIngestProcessorFactory
         );
     }
+
+    public void testGetSettings() {
+        assertEquals(List.of(TRIGGER_SETTING), plugin.getSettings());
+    }
 }

plugins/examples/system-ingest-processor/src/yamlRestTest/resources/rest-api-spec/test/example-system-ingest-processor/20_system_ingest_processor.yml

@@ -29,6 +29,28 @@ teardown:
   - match:
       _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
 
+---
+"Processor injects a field on indexing a doc to existing index when trigger_setting is true":
+  - do:
+      indices.create:
+        index: test-index
+        body:
+          settings:
+            index.example_system_ingest_processor_plugin.trigger_setting: true
+  - do:
+      index:
+        index: test-index
+        id: 1
+        body:
+          system_ingest_processor_trigger_field: "dummy value"
+        refresh: true
+  - do:
+      get:
+        index: test-index
+        id: 1
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
+
 ---
 "Processor should not inject a field on indexing a doc to existing index when trigger field is not defined in the index mapping":
   - do:
@@ -56,6 +78,7 @@ teardown:
   - skip:
       features: allowed_warnings
   - do:
+      # test v1 template
       indices.put_template:
         name: example-template
         body:
@@ -79,7 +102,41 @@ teardown:
         id: 1
   - match:
       _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
+  - do:
+      indices.delete_template:
+        name: example-template
+---
 
+"Processor injects field when index is created from matching template where trigger_setting is true":
+  - skip:
+      features: allowed_warnings
+  - do:
+      # test v2 template
+      indices.put_index_template:
+        name: example-template
+        body:
+          index_patterns: ["template-*"]
+          template:
+            settings:
+              index.example_system_ingest_processor_plugin.trigger_setting: true
+  - do:
+      allowed_warnings:
+        - "index [template-index-1] matches multiple legacy templates [example-template, global], composable templates will only match a single template"
+      index:
+        index: template-index-1
+        id: 1
+        body:
+          system_ingest_processor_trigger_field: "dummy value"
+        refresh: true
+  - do:
+      get:
+        index: template-index-1
+        id: 1
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
+  - do:
+      indices.delete_index_template:
+        name: example-template
 ---
 "Processor injects field on bulk indexing to existing index":
   - do:
@@ -156,7 +213,9 @@ teardown:
         id: 2
   - match:
       _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
-
+  - do:
+      indices.delete_template:
+        name: bulk-template
 ---
 "Processor injects field on bulk update, upsert on existing/new documents":
   # Temporarily disable system ingest pipelines to insert without triggering the system ingest field

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

@@ -107,7 +107,9 @@
 import static org.opensearch.cluster.service.ClusterManagerTask.DELETE_PIPELINE;
 import static org.opensearch.cluster.service.ClusterManagerTask.PUT_PIPELINE;
 import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_MAPPINGS;
+import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_SETTINGS;
 import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_TEMPLATE_MAPPINGS;
+import static org.opensearch.plugins.IngestPlugin.SystemIngestPipelineConfigKeys.INDEX_TEMPLATE_SETTINGS;
 
 /**
  * Holder class for several ingest related services.
@@ -272,7 +274,8 @@ public boolean resolveSystemIngestPipeline(
                 // precedence), or if a V2 template does not match, any V1 templates
                 String v2Template = MetadataIndexTemplateService.findV2Template(metadata, indexRequest.index(), false);
                 if (v2Template != null) {
-                    systemIngestPipelineId = getSystemIngestPipelineForTemplateV2(v2Template, indexRequest);
+                    Settings settings = MetadataIndexTemplateService.resolveSettings(metadata, v2Template);
+                    systemIngestPipelineId = getSystemIngestPipelineForTemplateV2(v2Template, indexRequest, settings);
                 } else {
                     List<IndexTemplateMetadata> templates = MetadataIndexTemplateService.findV1Templates(
                         metadata,
@@ -344,7 +347,7 @@ public boolean resolvePipelines(final DocWriteRequest<?> originalRequest, final
                     }
 
                     if (this.isSystemIngestPipelineEnabled) {
-                        systemIngestPipelineId = getSystemIngestPipelineForTemplateV2(v2Template, indexRequest);
+                        systemIngestPipelineId = getSystemIngestPipelineForTemplateV2(v2Template, indexRequest, settings);
                     }
                 } else {
                     List<IndexTemplateMetadata> templates = MetadataIndexTemplateService.findV1Templates(
@@ -433,9 +436,13 @@ private String getSystemIngestPipelineForTemplateV1(
         final String indexId = createIndexIdWithTemplateSuffix(indexRequest.index());
         Pipeline ingestPipeline = systemIngestPipelineCache.getSystemIngestPipeline(indexId);
         if (ingestPipeline == null) {
+            final List<Settings> settingsList = new ArrayList<>();
             final List<Map<String, Object>> mappingsMap = new ArrayList<>();
             final Map<String, Object> pipelineConfig = new HashMap<>();
             for (final IndexTemplateMetadata template : templates) {
+                if (template.settings() != null) {
+                    settingsList.add(template.settings());
+                }
                 if (template.mappings() != null) {
                     try {
                         mappingsMap.add(MapperService.parseMapping(xContentRegistry, template.mappings().string()));
@@ -453,6 +460,7 @@ private String getSystemIngestPipelineForTemplateV1(
             }
 
             pipelineConfig.put(INDEX_TEMPLATE_MAPPINGS, mappingsMap);
+            pipelineConfig.put(INDEX_TEMPLATE_SETTINGS, settingsList);
             ingestPipeline = createSystemIngestPipeline(indexId, pipelineConfig);
         }
 
@@ -461,7 +469,11 @@ private String getSystemIngestPipelineForTemplateV1(
         return ingestPipeline.getProcessors().isEmpty() ? null : indexId;
     }
 
-    private String getSystemIngestPipelineForTemplateV2(@NonNull final String templateName, @NonNull final IndexRequest indexRequest) {
+    private String getSystemIngestPipelineForTemplateV2(
+        @NonNull final String templateName,
+        @NonNull final IndexRequest indexRequest,
+        final Settings settings
+    ) {
         // Here we cache it with index name + template as the suffix since currently we don't have the uuid.
         // We need to cache it so that later during execution we can find it by indexId to reuse it.
         final String indexId = createIndexIdWithTemplateSuffix(indexRequest.index());
@@ -491,6 +503,7 @@ private String getSystemIngestPipelineForTemplateV2(@NonNull final String templa
             }
 
             pipelineConfig.put(INDEX_TEMPLATE_MAPPINGS, mappingsMap);
+            pipelineConfig.put(INDEX_TEMPLATE_SETTINGS, settings == null ? Collections.emptyList() : List.of(settings));
             ingestPipeline = createSystemIngestPipeline(indexId, pipelineConfig);
         }
 
@@ -515,10 +528,14 @@ private String getSystemIngestPipelineForExistingIndex(@NonNull final IndexMetad
         if (ingestPipeline == null) {
             // no cache we will try to resolve the ingest pipeline based on the index configuration
             final MappingMetadata mappingMetadata = indexMetadata.mapping();
+            final Settings settings = indexMetadata.getSettings();
             final Map<String, Object> pipelineConfig = new HashMap<>();
             if (mappingMetadata != null) {
                 pipelineConfig.put(INDEX_MAPPINGS, mappingMetadata.getSourceAsMap());
             }
+            if (settings != null) {
+                pipelineConfig.put(INDEX_SETTINGS, settings);
+            }
             ingestPipeline = createSystemIngestPipeline(indexId, pipelineConfig);
         }
         // we can get an empty pipeline from the cache

server/src/main/java/org/opensearch/plugins/IngestPlugin.java

@@ -109,5 +109,17 @@ class SystemIngestPipelineConfigKeys {
          *  ]
          */
         public static final String INDEX_TEMPLATE_MAPPINGS = "index_template_mappings";
+
+        /**
+         * Use this key to access the settings{@link org.opensearch.common.settings.Settings} of the index from the config.
+         */
+        public static final String INDEX_SETTINGS = "index_settings";
+
+        /**
+         * Use this key to access the settings{@link org.opensearch.common.settings.Settings} of the matched templates
+         * of the index from the config. If there are multiple matched templates the later one can override the setting of the previous one if merge
+         * rules are allowed. So this will be a list of settings.
+         */
+        public static final String INDEX_TEMPLATE_SETTINGS = "index_template_settings";
     }
 }