chore: take update from upstream

Signed-off-by: Mohammad Hasnain <[email protected]>
opensearch-project · Mar 27, 2024 · 1042982 · 1042982
2 parents 7fd7c1f + 5db84d1
commit 1042982
Show file tree

Hide file tree

Showing 22 changed files with 322 additions and 61 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -103,8 +103,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased 2.x]
 ### Added
-- Constant Keyword Field ([#12285](https://github.com/opensearch-project/OpenSearch/pull/12285))
+- Convert ingest processor supports ip type ([#12818](https://github.com/opensearch-project/OpenSearch/pull/12818))
 - Add a counter to node stat api to track shard going from idle to non-idle ([#12768](https://github.com/opensearch-project/OpenSearch/pull/12768))
+- Allow setting KEYSTORE_PASSWORD through env variable ([#12865](https://github.com/opensearch-project/OpenSearch/pull/12865))
+- [Concurrent Segment Search] Perform buildAggregation concurrently and support Composite Aggregations ([#12697](https://github.com/opensearch-project/OpenSearch/pull/12697))
+- Constant Keyword Field ([#12285](https://github.com/opensearch-project/OpenSearch/pull/12285))
 
 ### Dependencies
 - Bump `org.apache.commons:commons-configuration2` from 2.10.0 to 2.10.1 ([#12896](https://github.com/opensearch-project/OpenSearch/pull/12896))

diff --git a/distribution/src/bin/opensearch b/distribution/src/bin/opensearch
@@ -36,14 +36,16 @@ fi
 
 # get keystore password before setting java options to avoid
 # conflicting GC configurations for the keystore tools
-unset KEYSTORE_PASSWORD
-KEYSTORE_PASSWORD=
 if [[ $CHECK_KEYSTORE = true ]] \
     && bin/opensearch-keystore has-passwd --silent
 then
-  if ! read -s -r -p "OpenSearch keystore password: " KEYSTORE_PASSWORD ; then
-    echo "Failed to read keystore password on console" 1>&2
-    exit 1
+  if [[ ! -z "${KEYSTORE_PASSWORD}" ]]; then
+    echo "Using value of KEYSTORE_PASSWORD from the environment"
+  else
+    if ! read -s -r -p "OpenSearch keystore password: " KEYSTORE_PASSWORD ; then
+      echo "Failed to read keystore password on console" 1>&2
+      exit 1
+    fi
   fi
 fi
 

diff --git a/distribution/src/bin/opensearch.bat b/distribution/src/bin/opensearch.bat
@@ -62,14 +62,17 @@ if not exist "%SERVICE_LOG_DIR%" (
 	mkdir "%SERVICE_LOG_DIR%"
 )
 
-SET KEYSTORE_PASSWORD=
 IF "%checkpassword%"=="Y" (
   CALL "%~dp0opensearch-keystore.bat" has-passwd --silent
   IF !ERRORLEVEL! EQU 0 (
-    SET /P KEYSTORE_PASSWORD=OpenSearch keystore password:
-    IF !ERRORLEVEL! NEQ 0 (
-      ECHO Failed to read keystore password on standard input
-      EXIT /B !ERRORLEVEL!
+    if defined KEYSTORE_PASSWORD (
+      ECHO Using value of KEYSTORE_PASSWORD from the environment
+    ) else (
+      SET /P KEYSTORE_PASSWORD=OpenSearch keystore password:
+      IF !ERRORLEVEL! NEQ 0 (
+        ECHO Failed to read keystore password on standard input
+        EXIT /B !ERRORLEVEL!
+      )
     )
   )
 )

diff --git a/modules/ingest-common/src/main/java/org/opensearch/ingest/common/ConvertProcessor.java b/modules/ingest-common/src/main/java/org/opensearch/ingest/common/ConvertProcessor.java
@@ -32,6 +32,7 @@
 
 package org.opensearch.ingest.common;
 
+import org.opensearch.common.network.InetAddresses;
 import org.opensearch.ingest.AbstractProcessor;
 import org.opensearch.ingest.ConfigurationUtils;
 import org.opensearch.ingest.IngestDocument;
@@ -118,6 +119,19 @@ public Object convert(Object value) {
                 return value.toString();
             }
         },
+        IP {
+            @Override
+            public Object convert(Object value) {
+                // If the value is a valid ipv4/ipv6 address, we return the original value directly because IpFieldType
+                // can accept string value, this is simpler than we return an InetAddress object which needs to do more
+                // work such as serialization
+                if (value instanceof String && InetAddresses.isInetAddress(value.toString())) {
+                    return value;
+                } else {
+                    throw new IllegalArgumentException("[" + value + "] is not a valid ipv4/ipv6 address");
+                }
+            }
+        },
         AUTO {
             @Override
             public Object convert(Object value) {

diff --git a/modules/ingest-common/src/test/java/org/opensearch/ingest/common/ConvertProcessorTests.java b/modules/ingest-common/src/test/java/org/opensearch/ingest/common/ConvertProcessorTests.java
@@ -550,4 +550,29 @@ public void testTargetField() throws Exception {
         assertThat(ingestDocument.getFieldValue(fieldName, String.class), equalTo(String.valueOf(randomInt)));
         assertThat(ingestDocument.getFieldValue(targetField, Integer.class), equalTo(randomInt));
     }
+
+    public void testConvertIP() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+        String validIPString;
+        if (randomBoolean()) {
+            validIPString = "1.2.3.4";
+        } else {
+            validIPString = "::1";
+        }
+        String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, validIPString);
+
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), null, fieldName, fieldName, Type.IP, false);
+        processor.execute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue(fieldName, String.class), equalTo(validIPString));
+
+        String invalidIPString = randomAlphaOfLength(10);
+        fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, invalidIPString);
+        Processor processorWithInvalidIP = new ConvertProcessor(randomAlphaOfLength(10), null, fieldName, fieldName, Type.IP, false);
+        try {
+            processorWithInvalidIP.execute(ingestDocument);
+            fail("processor execute should have failed");
+        } catch (IllegalArgumentException e) {
+            assertThat(e.getMessage(), equalTo("[" + invalidIPString + "] is not a valid ipv4/ipv6 address"));
+        }
+    }
 }
diff --git a/...est-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/330_convert_processor.yml b/...est-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/330_convert_processor.yml
@@ -0,0 +1,83 @@
+---
+teardown:
+  - do:
+      ingest.delete_pipeline:
+        id: "1"
+        ignore: 404
+
+---
+"Test convert processor with ip type":
+  - skip:
+      version: " - 2.13.99"
+      reason: "introduced in 2.14.0"
+  - do:
+      ingest.put_pipeline:
+        id: "1"
+        body:  >
+          {
+            "processors": [
+              {
+                "convert" : {
+                  "field" : "raw_ip",
+                  "type": "ip"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      catch: /\[1.1.1.\] is not a valid ipv4\/ipv6 address/
+      index:
+        index: test
+        id: 1
+        pipeline: "1"
+        body: {
+          raw_ip: "1.1.1."
+        }
+
+  - do:
+      ingest.put_pipeline:
+        id: "1"
+        body:  >
+          {
+            "processors": [
+              {
+                "convert" : {
+                  "field" : "raw_ip",
+                  "target_field" : "ip_field",
+                  "type" : "ip",
+                  "ignore_failure" : true
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      index:
+        index: test
+        id: 1
+        pipeline: "1"
+        body: {
+          raw_ip: "1.1.1."
+        }
+  - do:
+      get:
+        index: test
+        id: 1
+  - match: { _source: { raw_ip: "1.1.1."} }
+
+  - do:
+      index:
+        index: test
+        id: 1
+        pipeline: "1"
+        body: {
+          raw_ip: "1.1.1.1"
+        }
+  - do:
+      get:
+        index: test
+        id: 1
+  - match: { _source: { raw_ip: "1.1.1.1", ip_field: "1.1.1.1"} }
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.clone/10_basic.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.clone/10_basic.yml
@@ -113,12 +113,12 @@ setup:
             index.number_of_replicas: 0
             index.number_of_shards: 6
 
+# Related issue: https://github.com/opensearch-project/OpenSearch/issues/4845
 ---
 "Returns error if target index's metadata write is blocked":
-
   - skip:
-      version: " - 2.99.99"
-      reason: "only available in 3.0+"
+      version: " - 2.7.99"
+      reason: "the bug was fixed in 2.8.0"
 
   # block source index's write operations
   - do:

diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.shrink/10_basic.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.shrink/10_basic.yml
@@ -72,12 +72,12 @@ setup:
   - match: { _id:      "1"     }
   - match: { _source:  { foo: "hello world" } }
 
+# Related issue: https://github.com/opensearch-project/OpenSearch/issues/4845
 ---
 "Returns error if target index's metadata write is blocked":
-
   - skip:
-      version: " - 2.99.99"
-      reason: "only available in 3.0+"
+      version: " - 2.7.99"
+      reason: "the bug was fixed in 2.8.0"
 
   # block source index's write operations
   - do:

diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.split/10_basic.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.split/10_basic.yml
@@ -219,12 +219,12 @@ setup:
             index.number_of_replicas: 0
             index.number_of_shards: 6
 
+# Related issue: https://github.com/opensearch-project/OpenSearch/issues/4845
 ---
 "Returns error if target index's metadata write is blocked":
-
   - skip:
-      version: " - 2.99.99"
-      reason: "only available in 3.0+"
+      version: " - 2.7.99"
+      reason: "the bug was fixed in 2.8.0"
 
   # block source index's write operations
   - do:

diff --git a/...nternalClusterTest/java/org/opensearch/search/aggregations/AggregationsIntegrationIT.java b/...nternalClusterTest/java/org/opensearch/search/aggregations/AggregationsIntegrationIT.java
@@ -38,6 +38,8 @@
 import org.opensearch.action.index.IndexRequestBuilder;
 import org.opensearch.action.search.SearchPhaseExecutionException;
 import org.opensearch.action.search.SearchResponse;
+import org.opensearch.action.support.WriteRequest;
+import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.search.aggregations.bucket.terms.IncludeExclude;
@@ -56,6 +58,8 @@
 import java.util.List;
 
 import static org.opensearch.search.SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING;
+import static org.opensearch.search.aggregations.AggregationBuilders.global;
+import static org.opensearch.search.aggregations.AggregationBuilders.stats;
 import static org.opensearch.search.aggregations.AggregationBuilders.terms;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertSearchResponse;
@@ -164,4 +168,23 @@ private void runLargeStringAggregationTest(AggregationBuilder aggregation) {
         }
         assertTrue("Exception should have been thrown", exceptionThrown);
     }
+
+    public void testAggsOnEmptyShards() {
+        // Create index with 5 shards but only 1 doc
+        assertAcked(
+            prepareCreate(
+                "idx",
+                Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 5).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+            ).setMapping("score", "type=integer")
+        );
+        client().prepareIndex("idx").setId("1").setSource("score", "5").setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
+
+        // Validate global agg does not throw an exception
+        assertSearchResponse(
+            client().prepareSearch("idx").addAggregation(global("global").subAggregation(stats("value_stats").field("score"))).get()
+        );
+
+        // Validate non-global agg does not throw an exception
+        assertSearchResponse(client().prepareSearch("idx").addAggregation(stats("value_stats").field("score")).get());
+    }
 }
diff --git a/...rc/internalClusterTest/java/org/opensearch/search/aggregations/bucket/CompositeAggIT.java b/...rc/internalClusterTest/java/org/opensearch/search/aggregations/bucket/CompositeAggIT.java
@@ -26,6 +26,7 @@
 import java.util.Collection;
 import java.util.List;
 
+import static org.opensearch.indices.IndicesRequestCache.INDEX_CACHE_REQUEST_ENABLED_SETTING;
 import static org.opensearch.search.SearchService.CLUSTER_CONCURRENT_SEGMENT_SEARCH_SETTING;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertSearchResponse;
@@ -50,23 +51,25 @@ public void setupSuiteScopeCluster() throws Exception {
         assertAcked(
             prepareCreate(
                 "idx",
-                Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                Settings.builder()
+                    .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                    .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                    .put(INDEX_CACHE_REQUEST_ENABLED_SETTING.getKey(), false)
             ).setMapping("type", "type=keyword", "num", "type=integer", "score", "type=integer")
         );
         waitForRelocation(ClusterHealthStatus.GREEN);
 
-        client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "1", "score", "5").get();
-        client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "11", "score", "50").get();
-        refresh("idx");
-        client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "1", "score", "2").get();
-        client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "12", "score", "20").get();
-        refresh("idx");
-        client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "3", "score", "10").get();
-        client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "13", "score", "15").get();
-        refresh("idx");
-        client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "3", "score", "1").get();
-        client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "13", "score", "100").get();
-        refresh("idx");
+        indexRandom(
+            true,
+            client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "1", "score", "5"),
+            client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "11", "score", "50"),
+            client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "1", "score", "2"),
+            client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "12", "score", "20"),
+            client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "3", "score", "10"),
+            client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "13", "score", "15"),
+            client().prepareIndex("idx").setId("1").setSource("type", "type1", "num", "3", "score", "1"),
+            client().prepareIndex("idx").setId("1").setSource("type", "type2", "num", "13", "score", "100")
+        );
 
         waitForRelocation(ClusterHealthStatus.GREEN);
         refresh();

diff --git a/server/src/main/java/org/opensearch/search/aggregations/AggregationCollectorManager.java b/server/src/main/java/org/opensearch/search/aggregations/AggregationCollectorManager.java
@@ -15,9 +15,9 @@
 import org.opensearch.search.query.ReduceableSearchResult;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Objects;
 
 /**
  * Common {@link CollectorManager} used by both concurrent and non-concurrent aggregation path and also for global and non-global
@@ -56,17 +56,9 @@ public String getCollectorReason() {
 
     @Override
     public ReduceableSearchResult reduce(Collection<Collector> collectors) throws IOException {
-        final List<Aggregator> aggregators = context.bucketCollectorProcessor().toAggregators(collectors);
-        final List<InternalAggregation> internals = new ArrayList<>(aggregators.size());
+        final List<InternalAggregation> internals = context.bucketCollectorProcessor().toInternalAggregations(collectors);
+        assert internals.stream().noneMatch(Objects::isNull);
         context.aggregations().resetBucketMultiConsumer();
-        for (Aggregator aggregator : aggregators) {
-            try {
-                // post collection is called in ContextIndexSearcher after search on leaves are completed
-                internals.add(aggregator.buildTopLevel());
-            } catch (IOException e) {
-                throw new AggregationExecutionException("Failed to build aggregation [" + aggregator.name() + "]", e);
-            }
-        }
 
         final InternalAggregations internalAggregations = InternalAggregations.from(internals);
         return buildAggregationResult(internalAggregations);