Support system ingest pipelines for bulk update operations (opensearch-project#18277)

* Support system ingest pipelines for bulk update operations

Signed-off-by: Andy Qin <[email protected]>

* Spotless

Signed-off-by: Andy Qin <[email protected]>

# Conflicts:
#	CHANGELOG.md

* Refactor UpdateRequest type checking

Signed-off-by: Andy Qin <[email protected]>

* Fix integ tests after rebase

Signed-off-by: Andy Qin <[email protected]>

* Rename innerSlot to childSlot

Signed-off-by: Andy Qin <[email protected]>

* Update yaml tests

Signed-off-by: Andy Qin <[email protected]>

* Update comments

Signed-off-by: Andy Qin <[email protected]>

* Add allowed warning header for yaml tests

Signed-off-by: Andy Qin <[email protected]>

* Update warning headers

Signed-off-by: Andy Qin <[email protected]>

---------

Signed-off-by: Andy Qin <[email protected]>
Signed-off-by: Sarat Vemulapalli <[email protected]>
Co-authored-by: Sarat Vemulapalli <[email protected]>

Author: q-andy

CHANGELOG.md

@@ -49,6 +49,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Added node-left metric to cluster manager ([#18421](https://github.com/opensearch-project/OpenSearch/pull/18421))
 - [Star tree] Remove star tree feature flag and add index setting to configure star tree search on index basis ([#18070](https://github.com/opensearch-project/OpenSearch/pull/18070))
 - Approximation Framework Enhancement: Update the BKD traversal logic to improve the performance on skewed data ([#18439](https://github.com/opensearch-project/OpenSearch/issues/18439))
+- Support system generated ingest pipelines for bulk update operations ([#18277](https://github.com/opensearch-project/OpenSearch/pull/18277)))
 - Added FS Health Check Failure metric ([#18435](https://github.com/opensearch-project/OpenSearch/pull/18435))
 
 ### Changed

plugins/examples/mapping-transformer/src/yamlRestTest/resources/rest-api-spec/test/example-mapping-transformer/20_mapping_transformer.yml

@@ -39,7 +39,11 @@
 
 ---
 "Test auto-injected field via index template":
+  - skip:
+      features: allowed_warnings
   - do:
+      allowed_warnings:
+        - "index template [example_template] has index patterns [auto-template-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [example_template] will take precedence during new index creation"
       indices.put_index_template:
         name: example_template
         body:
@@ -62,6 +66,8 @@
 
 ---
 "Test auto-injected field via legacy create template API":
+  - skip:
+      features: allowed_warnings
   - do:
       indices.put_template:
         name: legacy_template
@@ -72,6 +78,8 @@
               mapping_transform_trigger_field:
                 type: keyword
   - do:
+      allowed_warnings:
+        - "index [legacy-1] matches multiple legacy templates [global, legacy_template], composable templates will only match a single template"
       indices.create:
         index: legacy-1
   - do:

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

@@ -44,7 +44,7 @@ public void testSubBatchExecute() {
 
     private IngestDocumentWrapper createIngestDocumentWrapper(int slot) {
         final IngestDocument ingestDocument = new IngestDocument(new HashMap<>(), new HashMap<>());
-        return new IngestDocumentWrapper(slot, ingestDocument, null);
+        return new IngestDocumentWrapper(slot, 0, ingestDocument, null);
     }
 
     public void testGetType() {

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

@@ -1,3 +1,10 @@
+teardown:
+  - do:
+      cluster.put_settings:
+        body:
+          persistent:
+            cluster.ingest.system_pipeline_enabled: "true"
+
 ---
 "Processor injects a field on indexing a doc to existing index":
   - do:
@@ -106,6 +113,8 @@
 
 ---
 "Processor injects field on bulk index to non-existent index matching template":
+  - skip:
+        features: allowed_warnings
   - do:
       indices.put_template:
         name: bulk-template
@@ -116,6 +125,8 @@
               system_ingest_processor_trigger_field:
                 type: keyword
   - do:
+      allowed_warnings:
+        - "index [bulk-template-index-1] matches multiple legacy templates [bulk-template, global], composable templates will only match a single template"
       bulk:
         refresh: true
         body:
@@ -141,3 +152,135 @@
         id: 2
   - match:
       _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
+
+---
+"Processor injects field on bulk update, upsert on existing/new documents":
+  # Temporarily disable system ingest pipelines to insert without triggering the system ingest field
+  - do:
+      cluster.put_settings:
+        body:
+          persistent:
+            cluster.ingest.system_pipeline_enabled: "false"
+
+  - do:
+      indices.create:
+        index: bulk-index
+        body:
+          mappings:
+            properties:
+              system_ingest_processor_trigger_field:
+                type: keyword
+
+# Index documents to be updated on later step
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - index:
+              _index: bulk-index
+              _id: "1"
+          - system_ingest_processor_trigger_field: "bulk1"
+          - index:
+              _index: bulk-index
+              _id: "2"
+          - system_ingest_processor_trigger_field: "bulk2"
+          - index:
+              _index: bulk-index
+              _id: "3"
+          - system_ingest_processor_trigger_field: "bulk3"
+
+# Verify system ingest pipelines weren't triggered on regular index operations
+  - do:
+      get:
+        index: bulk-index
+        id: 1
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: null
+
+  - do:
+      get:
+        index: bulk-index
+        id: 2
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: null
+
+  - do:
+      get:
+        index: bulk-index
+        id: 2
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: null
+
+# Reenable system ingest pipeline
+  - do:
+      cluster.put_settings:
+        body:
+          persistent:
+            cluster.ingest.system_pipeline_enabled: "true"
+
+# Bulk update, bulk upsert existing, and bulk upsert new should all trigger ingest pipelines
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - update:
+              _index: bulk-index
+              _id: "1"
+          - doc:
+              system_ingest_processor_trigger_field: "update bulk1"
+          - update:
+              _index: bulk-index
+              _id: "2"
+          - doc:
+              system_ingest_processor_trigger_field: "update bulk2"
+            upsert:
+              system_ingest_processor_trigger_field: "new upsert bulk2"
+          - update:
+              _index: bulk-index
+              _id: "3"
+          - doc:
+              system_ingest_processor_trigger_field: "doc as upsert bulk3"
+            doc_as_upsert: true
+          - update:
+              _index: bulk-index
+              _id: "4"
+          - doc:
+              system_ingest_processor_trigger_field: "update bulk4"
+            upsert:
+              system_ingest_processor_trigger_field: "new upsert bulk4"
+
+  - do:
+      get:
+        index: bulk-index
+        id: 1
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
+  - match:
+      _source.system_ingest_processor_trigger_field: "update bulk1"
+
+  - do:
+      get:
+        index: bulk-index
+        id: 2
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
+  - match:
+      _source.system_ingest_processor_trigger_field: "update bulk2"
+
+  - do:
+      get:
+        index: bulk-index
+        id: 3
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
+  - match:
+      _source.system_ingest_processor_trigger_field: "doc as upsert bulk3"
+
+  - do:
+      get:
+        index: bulk-index
+        id: 4
+  - match:
+      _source.field_auto_added_by_system_ingest_processor: "This field is auto added by the example system ingest processor."
+  - match:
+      _source.system_ingest_processor_trigger_field: "new upsert bulk4"