Merge branch 'main' into AddWarmDiskMonitor

Author: Gagan Singh Saini

CHANGELOG.md

@@ -14,6 +14,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Bump OpenSearch Core main branch to 3.0.0 ([#18039](https://github.com/opensearch-project/OpenSearch/pull/18039))
 - [Rule based Auto-tagging] Add wlm `ActionFilter` ([#17791](https://github.com/opensearch-project/OpenSearch/pull/17791))
 - [Rule based auto-tagging] Add update rule API ([#17797](https://github.com/opensearch-project/OpenSearch/pull/17797))
+- [Rule based auto-tagging] Bug fix for update rule api ([#18488](https://github.com/opensearch-project/OpenSearch/pull/18488))
 - Update API of Message in index to add the timestamp for lag calculation in ingestion polling ([#17977](https://github.com/opensearch-project/OpenSearch/pull/17977/))
 - Add Warm Disk Threshold Allocation Decider for Warm shards ([#18082](https://github.com/opensearch-project/OpenSearch/pull/18082))
 - Add support for Warm Indices Write Block on Flood Watermark breach ([#18375](https://github.com/opensearch-project/OpenSearch/pull/18375))
@@ -50,6 +51,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
@@ -100,7 +102,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - DocValues-only IP field supports `terms_query` with more than 1025 IP masks ([#18357](https://github.com/opensearch-project/OpenSearch/pull/18357)
 - Fix MatrixStatsAggregator reuse when mode parameter changes ([#18242](https://github.com/opensearch-project/OpenSearch/issues/18242))
 - Replace the deprecated construction method of TopScoreDocCollectorManager with the new method ([#18395](https://github.com/opensearch-project/OpenSearch/pull/18395))
-- Fixed Approximate Framework regression with Lucene 10.2.1 by updating `intersectRight` BKD walk and `IntRef` visit method ([#18358](https://github.com/opensearch-project/OpenSearch/issues/18358
+- Fixed Approximate Framework regression with Lucene 10.2.1 by updating `intersectRight` BKD walk and `IntRef` visit method ([#18358](https://github.com/opensearch-project/OpenSearch/issues/18358))
+- Add task cancellation checks in aggregators ([#18426](https://github.com/opensearch-project/OpenSearch/pull/18426))
 
 ### Security
 

modules/autotagging-commons/common/src/main/java/org/opensearch/rule/RuleUtils.java

@@ -67,8 +67,12 @@ public static String computeRuleHash(
     public static Optional<String> getDuplicateRuleId(Rule rule, List<Rule> ruleList) {
         Map<Attribute, Set<String>> targetAttributeMap = rule.getAttributeMap();
         for (Rule currRule : ruleList) {
-            Map<Attribute, Set<String>> existingAttributeMap = currRule.getAttributeMap();
+            String currRuleId = currRule.getId();
+            if (currRuleId.equals(rule.getId())) {
+                continue;
+            }
 
+            Map<Attribute, Set<String>> existingAttributeMap = currRule.getAttributeMap();
             if (rule.getFeatureType() != currRule.getFeatureType() || targetAttributeMap.size() != existingAttributeMap.size()) {
                 continue;
             }
@@ -82,7 +86,7 @@ public static Optional<String> getDuplicateRuleId(Rule rule, List<Rule> ruleList
                 }
             }
             if (allAttributesIntersect) {
-                return Optional.of(currRule.getId());
+                return Optional.of(currRuleId);
             }
         }
         return Optional.empty();

modules/autotagging-commons/common/src/test/java/org/opensearch/rule/RuleUtilsTests.java

@@ -19,6 +19,7 @@
 import java.util.Optional;
 import java.util.Set;
 
+import static org.opensearch.rule.utils.RuleTestUtils.ATTRIBUTE_MAP;
 import static org.opensearch.rule.utils.RuleTestUtils.ATTRIBUTE_VALUE_ONE;
 import static org.opensearch.rule.utils.RuleTestUtils.ATTRIBUTE_VALUE_TWO;
 import static org.opensearch.rule.utils.RuleTestUtils.DESCRIPTION_ONE;
@@ -28,15 +29,25 @@
 import static org.opensearch.rule.utils.RuleTestUtils.MockRuleAttributes;
 import static org.opensearch.rule.utils.RuleTestUtils.TIMESTAMP_ONE;
 import static org.opensearch.rule.utils.RuleTestUtils._ID_ONE;
+import static org.opensearch.rule.utils.RuleTestUtils._ID_TWO;
 import static org.opensearch.rule.utils.RuleTestUtils.ruleOne;
 import static org.opensearch.rule.utils.RuleTestUtils.ruleTwo;
 
 public class RuleUtilsTests extends OpenSearchTestCase {
 
     public void testDuplicateRuleFound() {
-        Optional<String> result = RuleUtils.getDuplicateRuleId(ruleOne, List.of(ruleOne, ruleTwo));
+        Rule testRule = Rule.builder()
+            .id(_ID_TWO)
+            .description(DESCRIPTION_ONE)
+            .featureType(RuleTestUtils.MockRuleFeatureType.INSTANCE)
+            .featureValue(FEATURE_VALUE_ONE)
+            .attributeMap(ATTRIBUTE_MAP)
+            .updatedAt(TIMESTAMP_ONE)
+            .build();
+
+        Optional<String> result = RuleUtils.getDuplicateRuleId(ruleOne, List.of(testRule));
         assertTrue(result.isPresent());
-        assertEquals(_ID_ONE, result.get());
+        assertEquals(_ID_TWO, result.get());
     }
 
     public void testNoAttributeIntersection() {
@@ -66,7 +77,7 @@ public void testAttributeSizeMismatch() {
 
     public void testPartialAttributeValueIntersection() {
         Rule ruleWithPartialOverlap = Rule.builder()
-            .id(_ID_ONE)
+            .id(_ID_TWO)
             .description(DESCRIPTION_ONE)
             .featureType(RuleTestUtils.MockRuleFeatureType.INSTANCE)
             .featureValue(FEATURE_VALUE_ONE)
@@ -79,6 +90,11 @@ public void testPartialAttributeValueIntersection() {
         assertEquals(_ID_ONE, result.get());
     }
 
+    public void testDuplicateRuleWithSameId() {
+        Optional<String> result = RuleUtils.getDuplicateRuleId(ruleOne, List.of(ruleOne));
+        assertFalse(result.isPresent());
+    }
+
     public void testDifferentFeatureTypes() {
         Rule differentFeatureTypeRule = Rule.builder()
             .id(_ID_ONE)

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:
@@ -46,6 +53,8 @@
   - is_false: _source.field_auto_added_by_system_ingest_processor
 ---
 "Processor injects field when index is created from matching template":
+  - skip:
+      features: allowed_warnings
   - do:
       indices.put_template:
         name: example-template
@@ -56,6 +65,8 @@
               system_ingest_processor_trigger_field:
                 type: keyword
   - 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
@@ -106,6 +117,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 +129,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 +156,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"