diff --git a/.github/workflows/opensearch-sink-opendistro-integration-tests.yml b/.github/workflows/opensearch-sink-opendistro-integration-tests.yml
index cfaca4a330..45c477e12f 100644
--- a/.github/workflows/opensearch-sink-opendistro-integration-tests.yml
+++ b/.github/workflows/opensearch-sink-opendistro-integration-tests.yml
@@ -40,7 +40,7 @@ jobs:
- name: Run Open Distro tests
run: |
./gradlew :data-prepper-plugins:opensearch:integrationTest --tests "org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchIT" -Dtests.opensearch.host=localhost:9200 -Dtests.opensearch.user=admin -Dtests.opensearch.password=admin
- ./gradlew :data-prepper-plugins:opensearch:integrationTest -Dtests.opensearch.host=localhost:9200 -Dtests.opensearch.user=admin -Dtests.opensearch.password=admin -Dtests.opensearch.bundle=true
+ ./gradlew :data-prepper-plugins:opensearch:integrationTest -Dtests.opensearch.host=localhost:9200 -Dtests.opensearch.user=admin -Dtests.opensearch.password=admin -Dtests.opensearch.bundle=true -Dtests.opensearch.version=opendistro:${{ matrix.opendistro }}
- name: Upload Unit Test Results
if: always()
uses: actions/upload-artifact@v3
diff --git a/.github/workflows/opensearch-sink-opensearch-integration-tests.yml b/.github/workflows/opensearch-sink-opensearch-integration-tests.yml
index 925086127d..5e34350303 100644
--- a/.github/workflows/opensearch-sink-opensearch-integration-tests.yml
+++ b/.github/workflows/opensearch-sink-opensearch-integration-tests.yml
@@ -40,7 +40,7 @@ jobs:
- name: Run OpenSearch tests
run: |
./gradlew :data-prepper-plugins:opensearch:integrationTest --tests "org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchIT" -Dtests.opensearch.host=localhost:9200 -Dtests.opensearch.user=admin -Dtests.opensearch.password=admin
- ./gradlew :data-prepper-plugins:opensearch:integrationTest -Dtests.opensearch.host=localhost:9200 -Dtests.opensearch.user=admin -Dtests.opensearch.password=admin -Dtests.opensearch.bundle=true
+ ./gradlew :data-prepper-plugins:opensearch:integrationTest -Dtests.opensearch.host=localhost:9200 -Dtests.opensearch.user=admin -Dtests.opensearch.password=admin -Dtests.opensearch.bundle=true -Dtests.opensearch.version=opensearch:${{ matrix.opensearch }}
- name: Upload Unit Test Results
if: always()
uses: actions/upload-artifact@v3
diff --git a/data-prepper-plugins/opensearch/README.md b/data-prepper-plugins/opensearch/README.md
index 0ccfc39d33..edc297e3c0 100644
--- a/data-prepper-plugins/opensearch/README.md
+++ b/data-prepper-plugins/opensearch/README.md
@@ -129,6 +129,8 @@ Default is null.
* This index name can also be a plain string plus a date-time pattern as a suffix, such as `application-%{yyyy.MM.dd}`, `my-index-name-%{yyyy.MM.dd.HH}`. When OpenSearch Sink is sending data to OpenSearch, the date-time pattern will be replaced by actual UTC time. The pattern supports all the symbols that represent one hour or above and are listed in [Java DateTimeFormatter](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html). For example, with an index pattern like `my-index-name-%{yyyy.MM.dd}`, a new index is created for each day such as `my-index-name-2022.01.25`. For another example, with an index pattern like `my-index-name-%{yyyy.MM.dd.HH}`, a new index is created for each hour such as `my-index-name-2022.01.25.13`.
* This index name can also be a formatted string (with or without date-time pattern suffix), such as `my-${index}-name`. When OpenSearchSink is sending data to OpenSearch, the format portion "${index}" will be replaced by it's value in the event that is being processed. The format may also be like "${index1/index2/index3}" in which case the field "index1/index2/index3" is searched in the event and replaced by its value.
+- `template_type`(optional): Defines what type of OpenSearch template to use. The available options are `v1` and `index-template`. The default value is `v1`, which uses the original OpenSearch templates available at the `_template` API endpoints. Select `index-template` to use composable index templates which are available at OpenSearch's `_index_template` endpoint.
+
- `template_file`(optional): A json file path or AWS S3 URI to be read as index template for custom data ingestion. The json file content should be the json value of
`"template"` key in the json content of OpenSearch [Index templates API](https://opensearch.org/docs/latest/opensearch/index-templates/),
e.g. [otel-v1-apm-span-index-template.json](https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/opensearch/src/main/resources/otel-v1-apm-span-index-template.json)
diff --git a/data-prepper-plugins/opensearch/build.gradle b/data-prepper-plugins/opensearch/build.gradle
index e77cbc3815..40ecb6b2c7 100644
--- a/data-prepper-plugins/opensearch/build.gradle
+++ b/data-prepper-plugins/opensearch/build.gradle
@@ -64,9 +64,11 @@ task integrationTest(type: Test) {
systemProperty 'tests.opensearch.bundle', System.getProperty('tests.opensearch.bundle')
systemProperty 'tests.opensearch.user', System.getProperty('tests.opensearch.user')
systemProperty 'tests.opensearch.password', System.getProperty('tests.opensearch.password')
+ systemProperty 'tests.opensearch.version', System.getProperty('tests.opensearch.version')
filter {
includeTestsMatching '*IT'
+ includeTestsMatching '*Test'
}
}
diff --git a/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/DeclaredOpenSearchVersion.java b/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/DeclaredOpenSearchVersion.java
new file mode 100644
index 0000000000..ff79a0f555
--- /dev/null
+++ b/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/DeclaredOpenSearchVersion.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.sink.opensearch;
+
+import org.opensearch.Version;
+
+/**
+ * An interface for letting the test declare what OpenSearch version is being
+ * tested against.
+ */
+class DeclaredOpenSearchVersion implements Comparable {
+ private static final DeclaredOpenSearchVersion DEFAULT = new DeclaredOpenSearchVersion(Distribution.OPENSEARCH, "1.0.0");
+ public static final DeclaredOpenSearchVersion OPENDISTRO_1_9 = new DeclaredOpenSearchVersion(Distribution.OPENDISTRO, "1.9.0");
+
+ enum Distribution {
+ OPENDISTRO,
+ OPENSEARCH
+ }
+
+ private final Distribution distribution;
+ private final String version;
+ private final Version internalVersion;
+
+ private DeclaredOpenSearchVersion(final Distribution distribution, final String version) {
+ this.distribution = distribution;
+ this.version = version;
+
+ internalVersion = Version.fromString(version);
+ }
+
+ static DeclaredOpenSearchVersion parse(final String versionString) {
+ if(versionString == null) {
+ return DEFAULT;
+ }
+
+ final String[] parts = versionString.split(":");
+
+ if(parts.length != 2) {
+ throw new IllegalArgumentException("Invalid version string provided.");
+ }
+
+ final Distribution distribution = Distribution.valueOf(parts[0].toUpperCase());
+
+ return new DeclaredOpenSearchVersion(distribution, parts[1]);
+ }
+
+ @Override
+ public int compareTo(final DeclaredOpenSearchVersion other) {
+ final int distributionCompareTo = distribution.compareTo(other.distribution);
+
+ if(distributionCompareTo != 0) {
+ return distributionCompareTo;
+ }
+
+ return internalVersion.compareTo(other.internalVersion);
+ }
+
+ Distribution getDistribution() {
+ return distribution;
+ }
+
+ String getVersion() {
+ return version;
+ }
+}
diff --git a/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/DeclaredOpenSearchVersionTest.java b/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/DeclaredOpenSearchVersionTest.java
new file mode 100644
index 0000000000..820c39b820
--- /dev/null
+++ b/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/DeclaredOpenSearchVersionTest.java
@@ -0,0 +1,60 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.sink.opensearch;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class DeclaredOpenSearchVersionTest {
+ @ParameterizedTest
+ @CsvSource({
+ "opensearch:1.2.4,OPENSEARCH,1.2.4",
+ "opensearch:1.3.9,OPENSEARCH,1.3.9",
+ "opensearch:2.2.1,OPENSEARCH,2.2.1",
+ "opensearch:2.6.0,OPENSEARCH,2.6.0",
+ "opendistro:1.3.0,OPENDISTRO,1.3.0",
+ "opendistro:1.13.3,OPENDISTRO,1.13.3"
+ })
+ void parse_should_return_expected(final String versionString, final DeclaredOpenSearchVersion.Distribution expectedDistribution, final String expectedVersion) {
+ final DeclaredOpenSearchVersion version = DeclaredOpenSearchVersion.parse(versionString);
+
+ assertThat(version, notNullValue());
+ assertThat(version.getDistribution(), equalTo(expectedDistribution));
+ assertThat(version.getVersion(), equalTo(expectedVersion));
+ }
+
+ @Test
+ void parse_with_null_should_return_minimum_version() {
+ final DeclaredOpenSearchVersion version = DeclaredOpenSearchVersion.parse(null);
+
+ assertThat(version, notNullValue());
+ assertThat(version.getDistribution(), equalTo(DeclaredOpenSearchVersion.Distribution.OPENSEARCH));
+ assertThat(version.getVersion(), equalTo("1.0.0"));
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "opensearch:1.2.4,opensearch:1.2.4,0",
+ "opendistro:1.13.3,opendistro:1.13.3,0",
+ "opensearch:1.2.4,opensearch:1.2.3,1",
+ "opensearch:1.2.3,opensearch:1.2.4,-1",
+ "opensearch:2.6.0,opensearch:1.2.3,1",
+ "opensearch:1.2.3,opensearch:2.6.0,-1",
+ "opensearch:1.3.9,opendistro:1.13.3,1",
+ "opendistro:1.13.3,opensearch:1.3.9,-1"
+ })
+ void compareTo_returns_correct_value(final String versionStringToTest, final String otherVersionString, final int expectedCompareTo) {
+ final DeclaredOpenSearchVersion objectUnderTest = DeclaredOpenSearchVersion.parse(versionStringToTest);
+ final DeclaredOpenSearchVersion otherVersion = DeclaredOpenSearchVersion.parse(otherVersionString);
+
+ assertThat(objectUnderTest.compareTo(otherVersion), equalTo(expectedCompareTo));
+ }
+}
\ No newline at end of file
diff --git a/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchIntegrationHelper.java b/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchIntegrationHelper.java
index 43103636c5..b44562909d 100644
--- a/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchIntegrationHelper.java
+++ b/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchIntegrationHelper.java
@@ -92,6 +92,10 @@ static List getHosts() {
.map(ip -> String.format("https://%s", ip)).collect(Collectors.toList());
}
+ static DeclaredOpenSearchVersion getVersion() {
+ return DeclaredOpenSearchVersion.parse(System.getProperty("tests.opensearch.version"));
+ }
+
/**
* Copied from OpenSearch test framework
* TODO: Consolidate in OpenSearch
diff --git a/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkIT.java b/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkIT.java
index 57fe7320d0..f437c0c62c 100644
--- a/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkIT.java
+++ b/data-prepper-plugins/opensearch/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSinkIT.java
@@ -5,17 +5,6 @@
package org.opensearch.dataprepper.plugins.sink.opensearch;
-import org.mockito.Mock;
-import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
-import org.opensearch.dataprepper.metrics.MetricNames;
-import org.opensearch.dataprepper.metrics.MetricsTestUtil;
-import org.opensearch.dataprepper.model.configuration.PluginSetting;
-import org.opensearch.dataprepper.model.event.Event;
-import org.opensearch.dataprepper.model.event.EventType;
-import org.opensearch.dataprepper.model.event.EventHandle;
-import org.opensearch.dataprepper.model.event.JacksonEvent;
-import org.opensearch.dataprepper.model.plugin.PluginFactory;
-import org.opensearch.dataprepper.model.record.Record;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -28,19 +17,34 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtensionContext;
import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
+import org.junit.jupiter.params.provider.ArgumentsSource;
import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mock;
import org.opensearch.client.Request;
import org.opensearch.client.Response;
import org.opensearch.client.RestClient;
import org.opensearch.common.Strings;
import org.opensearch.common.xcontent.XContentFactory;
import org.opensearch.common.xcontent.XContentType;
+import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
+import org.opensearch.dataprepper.metrics.MetricNames;
+import org.opensearch.dataprepper.metrics.MetricsTestUtil;
+import org.opensearch.dataprepper.model.configuration.PluginSetting;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.event.EventHandle;
+import org.opensearch.dataprepper.model.event.EventType;
+import org.opensearch.dataprepper.model.event.JacksonEvent;
+import org.opensearch.dataprepper.model.plugin.PluginFactory;
+import org.opensearch.dataprepper.model.record.Record;
import org.opensearch.dataprepper.plugins.sink.opensearch.bulk.BulkAction;
+import org.opensearch.dataprepper.plugins.sink.opensearch.index.AbstractIndexManager;
import org.opensearch.dataprepper.plugins.sink.opensearch.index.IndexConfiguration;
import org.opensearch.dataprepper.plugins.sink.opensearch.index.IndexConstants;
-import org.opensearch.dataprepper.plugins.sink.opensearch.index.AbstractIndexManager;
import org.opensearch.dataprepper.plugins.sink.opensearch.index.IndexType;
import javax.ws.rs.HttpMethod;
@@ -53,6 +57,7 @@
import java.nio.file.Files;
import java.nio.file.Paths;
import java.text.SimpleDateFormat;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
@@ -64,43 +69,48 @@
import java.util.StringJoiner;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
+import java.util.function.BiFunction;
import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.http.HttpStatus.SC_OK;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.hasItems;
import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.CoreMatchers.hasItems;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.Matchers.closeTo;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
import static org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchIntegrationHelper.createContentParser;
import static org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchIntegrationHelper.createOpenSearchClient;
import static org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchIntegrationHelper.getHosts;
import static org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchIntegrationHelper.isOSBundle;
import static org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchIntegrationHelper.waitForClusterStateUpdatesToFinish;
import static org.opensearch.dataprepper.plugins.sink.opensearch.OpenSearchIntegrationHelper.wipeAllTemplates;
-import static org.mockito.Mockito.mock;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.lenient;
public class OpenSearchSinkIT {
- private static final String PLUGIN_NAME = "opensearch";
- private static final String PIPELINE_NAME = "integTestPipeline";
- private static final String TEST_CUSTOM_INDEX_POLICY_FILE = "test-custom-index-policy-file.json";
- private static final String TEST_TEMPLATE_V1_FILE = "test-index-template.json";
- private static final String TEST_TEMPLATE_V2_FILE = "test-index-template-v2.json";
- private static final String DEFAULT_RAW_SPAN_FILE_1 = "raw-span-1.json";
- private static final String DEFAULT_RAW_SPAN_FILE_2 = "raw-span-2.json";
- private static final String DEFAULT_SERVICE_MAP_FILE = "service-map-1.json";
-
- private RestClient client;
- private EventHandle eventHandle;
-
- @Mock
- private PluginFactory pluginFactory;
+ private static final String PLUGIN_NAME = "opensearch";
+ private static final String PIPELINE_NAME = "integTestPipeline";
+ private static final String TEST_CUSTOM_INDEX_POLICY_FILE = "test-custom-index-policy-file.json";
+ private static final String TEST_TEMPLATE_V1_FILE = "test-index-template.json";
+ private static final String TEST_TEMPLATE_V2_FILE = "test-index-template-v2.json";
+ private static final String TEST_INDEX_TEMPLATE_V1_FILE = "test-composable-index-template.json";
+ private static final String TEST_INDEX_TEMPLATE_V2_FILE = "test-composable-index-template-v2.json";
+ private static final String DEFAULT_RAW_SPAN_FILE_1 = "raw-span-1.json";
+ private static final String DEFAULT_RAW_SPAN_FILE_2 = "raw-span-2.json";
+ private static final String DEFAULT_SERVICE_MAP_FILE = "service-map-1.json";
+
+ private RestClient client;
+ private EventHandle eventHandle;
+
+ @Mock
+ private PluginFactory pluginFactory;
@Mock
private AwsCredentialsSupplier awsCredentialsSupplier;
@@ -113,866 +123,906 @@ public OpenSearchSink createObjectUnderTest(PluginSetting pluginSetting, boolean
return sink;
}
- @BeforeEach
- public void setup() {
- eventHandle = mock(EventHandle.class);
- lenient().doAnswer(a -> {
+ @BeforeEach
+ public void setup() {
+ eventHandle = mock(EventHandle.class);
+ lenient().doAnswer(a -> {
return null;
- }).when(eventHandle).release(any(Boolean.class));
- }
-
- @BeforeEach
- public void metricsInit() throws IOException {
- MetricsTestUtil.initMetrics();
-
- client = createOpenSearchClient();
- }
-
- @AfterEach
- public void cleanOpenSearch() throws Exception {
- wipeAllOpenSearchIndices();
- wipeAllTemplates();
- waitForClusterStateUpdatesToFinish();
- }
-
- @Test
- public void testInstantiateSinkRawSpanDefault() throws IOException {
- final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
- OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- final String indexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
- Request request = new Request(HttpMethod.HEAD, indexAlias);
- Response response = client.performRequest(request);
- MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
- final String index = String.format("%s-000001", indexAlias);
- final Map mappings = getIndexMappings(index);
- MatcherAssert.assertThat(mappings, notNullValue());
- MatcherAssert.assertThat((boolean) mappings.get("date_detection"), equalTo(false));
- sink.shutdown();
-
- if (isOSBundle()) {
- // Check managed index
- await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
- MatcherAssert.assertThat(getIndexPolicyId(index), equalTo(IndexConstants.RAW_ISM_POLICY));
- }
- );
- }
-
- // roll over initial index
- request = new Request(HttpMethod.POST, String.format("%s/_rollover", indexAlias));
- request.setJsonEntity("{ \"conditions\" : { } }\n");
- response = client.performRequest(request);
- MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
-
- // Instantiate sink again
- sink = createObjectUnderTest(pluginSetting, true);
- // Make sure no new write index *-000001 is created under alias
- final String rolloverIndexName = String.format("%s-000002", indexAlias);
- request = new Request(HttpMethod.GET, rolloverIndexName + "/_alias");
- response = client.performRequest(request);
- MatcherAssert.assertThat(checkIsWriteIndex(EntityUtils.toString(response.getEntity()), indexAlias, rolloverIndexName), equalTo(true));
- sink.shutdown();
-
- if (isOSBundle()) {
- // Check managed index
- MatcherAssert.assertThat(getIndexPolicyId(rolloverIndexName), equalTo(IndexConstants.RAW_ISM_POLICY));
+ }).when(eventHandle).release(any(Boolean.class));
}
- }
-
- @Test
- public void testInstantiateSinkRawSpanReservedAliasAlreadyUsedAsIndex() throws IOException {
- final String reservedIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
- final Request request = new Request(HttpMethod.PUT, reservedIndexAlias);
- client.performRequest(request);
- final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
- OpenSearchSink sink = createObjectUnderTest(pluginSetting, false);
- Assert.assertThrows(String.format(AbstractIndexManager.INDEX_ALIAS_USED_AS_INDEX_ERROR, reservedIndexAlias),
- RuntimeException.class, () -> sink.doInitialize());
- }
-
- @Test
- public void testOutputRawSpanDefault() throws IOException, InterruptedException {
- final String testDoc1 = readDocFromFile(DEFAULT_RAW_SPAN_FILE_1);
- final String testDoc2 = readDocFromFile(DEFAULT_RAW_SPAN_FILE_2);
- final ObjectMapper mapper = new ObjectMapper();
- @SuppressWarnings("unchecked") final Map expData1 = mapper.readValue(testDoc1, Map.class);
- @SuppressWarnings("unchecked") final Map expData2 = mapper.readValue(testDoc2, Map.class);
-
- final List> testRecords = Arrays.asList(jsonStringToRecord(testDoc1), jsonStringToRecord(testDoc2));
- final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
-
- final String expIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
- final List> retSources = getSearchResponseDocSources(expIndexAlias);
- MatcherAssert.assertThat(retSources.size(), equalTo(2));
- MatcherAssert.assertThat(retSources, hasItems(expData1, expData2));
- MatcherAssert.assertThat(getDocumentCount(expIndexAlias, "_id", (String) expData1.get("spanId")), equalTo(Integer.valueOf(1)));
- sink.shutdown();
-
- // Verify metrics
- final List bulkRequestErrors = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_ERRORS).toString());
- MatcherAssert.assertThat(bulkRequestErrors.size(), equalTo(1));
- Assert.assertEquals(0.0, bulkRequestErrors.get(0).getValue(), 0);
- final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
- MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
- // COUNT
- Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
- // TOTAL_TIME
- Assert.assertTrue(bulkRequestLatencies.get(1).getValue() > 0.0);
- // MAX
- Assert.assertTrue(bulkRequestLatencies.get(2).getValue() > 0.0);
- final List documentsSuccessMeasurements = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(BulkRetryStrategy.DOCUMENTS_SUCCESS).toString());
- MatcherAssert.assertThat(documentsSuccessMeasurements.size(), equalTo(1));
- MatcherAssert.assertThat(documentsSuccessMeasurements.get(0).getValue(), closeTo(2.0, 0));
- final List documentsSuccessFirstAttemptMeasurements = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(BulkRetryStrategy.DOCUMENTS_SUCCESS_FIRST_ATTEMPT).toString());
- MatcherAssert.assertThat(documentsSuccessFirstAttemptMeasurements.size(), equalTo(1));
- MatcherAssert.assertThat(documentsSuccessFirstAttemptMeasurements.get(0).getValue(), closeTo(2.0, 0));
- final List documentErrorsMeasurements = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(BulkRetryStrategy.DOCUMENT_ERRORS).toString());
- MatcherAssert.assertThat(documentErrorsMeasurements.size(), equalTo(1));
- MatcherAssert.assertThat(documentErrorsMeasurements.get(0).getValue(), closeTo(0.0, 0));
-
- /**
- * Metrics: Bulk Request Size in Bytes
- */
- final List bulkRequestSizeBytesMetrics = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_SIZE_BYTES).toString());
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.size(), equalTo(3));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(0).getValue(), closeTo(1.0, 0));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(1).getValue(), closeTo(2058.0, 0));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(2).getValue(), closeTo(2058.0, 0));
- }
-
- @Test
- public void testOutputRawSpanWithDLQ() throws IOException, InterruptedException {
- // TODO: write test case
- final String testDoc1 = readDocFromFile("raw-span-error.json");
- final String testDoc2 = readDocFromFile(DEFAULT_RAW_SPAN_FILE_1);
- final ObjectMapper mapper = new ObjectMapper();
- @SuppressWarnings("unchecked") final Map expData = mapper.readValue(testDoc2, Map.class);
-
- final List> testRecords = Arrays.asList(jsonStringToRecord(testDoc1), jsonStringToRecord(testDoc2));
- final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
- // generate temporary directory for dlq file
- final File tempDirectory = Files.createTempDirectory("").toFile();
- // add dlq file path into setting
- final String expDLQFile = tempDirectory.getAbsolutePath() + "/test-dlq.txt";
- pluginSetting.getSettings().put(RetryConfiguration.DLQ_FILE, expDLQFile);
-
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
- sink.shutdown();
-
- final StringBuilder dlqContent = new StringBuilder();
- Files.lines(Paths.get(expDLQFile)).forEach(dlqContent::append);
- final String nonPrettyJsonString = mapper.writeValueAsString(mapper.readValue(testDoc1, JsonNode.class));
- MatcherAssert.assertThat(dlqContent.toString(), containsString(nonPrettyJsonString));
- final String expIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
- final List> retSources = getSearchResponseDocSources(expIndexAlias);
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(retSources.get(0), equalTo(expData));
-
- // clean up temporary directory
- FileUtils.deleteQuietly(tempDirectory);
-
- // verify metrics
- final List documentsSuccessMeasurements = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(BulkRetryStrategy.DOCUMENTS_SUCCESS).toString());
- MatcherAssert.assertThat(documentsSuccessMeasurements.size(), equalTo(1));
- MatcherAssert.assertThat(documentsSuccessMeasurements.get(0).getValue(), closeTo(1.0, 0));
- final List documentErrorsMeasurements = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(BulkRetryStrategy.DOCUMENT_ERRORS).toString());
- MatcherAssert.assertThat(documentErrorsMeasurements.size(), equalTo(1));
- MatcherAssert.assertThat(documentErrorsMeasurements.get(0).getValue(), closeTo(1.0, 0));
-
- /**
- * Metrics: Bulk Request Size in Bytes
- */
- final List bulkRequestSizeBytesMetrics = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_SIZE_BYTES).toString());
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.size(), equalTo(3));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(0).getValue(), closeTo(1.0, 0));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(1).getValue(), closeTo(2072.0, 0));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(2).getValue(), closeTo(2072.0, 0));
- }
+ @BeforeEach
+ public void metricsInit() throws IOException {
+ MetricsTestUtil.initMetrics();
- @Test
- public void testInstantiateSinkServiceMapDefault() throws IOException {
- final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_SERVICE_MAP.getValue(), null, null);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- final String indexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_SERVICE_MAP);
- final Request request = new Request(HttpMethod.HEAD, indexAlias);
- final Response response = client.performRequest(request);
- MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
- final Map mappings = getIndexMappings(indexAlias);
- MatcherAssert.assertThat(mappings, notNullValue());
- MatcherAssert.assertThat((boolean) mappings.get("date_detection"), equalTo(false));
- sink.shutdown();
-
- if (isOSBundle()) {
- // Check managed index
- MatcherAssert.assertThat(getIndexPolicyId(indexAlias), nullValue());
+ client = createOpenSearchClient();
}
- }
-
- @Test
- public void testOutputServiceMapDefault() throws IOException, InterruptedException {
- final String testDoc = readDocFromFile(DEFAULT_SERVICE_MAP_FILE);
- final ObjectMapper mapper = new ObjectMapper();
- @SuppressWarnings("unchecked") final Map expData = mapper.readValue(testDoc, Map.class);
-
- final List> testRecords = Collections.singletonList(jsonStringToRecord(testDoc));
- final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_SERVICE_MAP.getValue(), null, null);
- OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
- final String expIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_SERVICE_MAP);
- final List> retSources = getSearchResponseDocSources(expIndexAlias);
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(retSources.get(0), equalTo(expData));
- MatcherAssert.assertThat(getDocumentCount(expIndexAlias, "_id", (String) expData.get("hashId")), equalTo(Integer.valueOf(1)));
- sink.shutdown();
-
- // verify metrics
- final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
- MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
- // COUNT
- MatcherAssert.assertThat(bulkRequestLatencies.get(0).getValue(), closeTo(1.0, 0));
-
- /**
- * Metrics: Bulk Request Size in Bytes
- */
- final List bulkRequestSizeBytesMetrics = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_SIZE_BYTES).toString());
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.size(), equalTo(3));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(0).getValue(), closeTo(1.0, 0));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(1).getValue(), closeTo(265.0, 0));
- MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(2).getValue(), closeTo(265.0, 0));
-
- // Check restart for index already exists
- sink = createObjectUnderTest(pluginSetting, true);
- sink.shutdown();
- }
- @Test
- public void testInstantiateSinkCustomIndex_NoRollOver() throws IOException {
- final String testIndexAlias = "test-alias";
- final String testTemplateFile = Objects.requireNonNull(
- getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE)).getFile();
- final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFile);
- OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- final Request request = new Request(HttpMethod.HEAD, testIndexAlias);
- final Response response = client.performRequest(request);
- MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
- sink.shutdown();
-
- // Check restart for index already exists
- sink = createObjectUnderTest(pluginSetting, true);
- sink.shutdown();
- }
-
- @Test
- public void testInstantiateSinkCustomIndex_WithIsmPolicy() throws IOException {
- final String indexAlias = "sink-custom-index-ism-test-alias";
- final String testTemplateFile = Objects.requireNonNull(
- getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE)).getFile();
- final Map metadata = initializeConfigurationMetadata(null, indexAlias, testTemplateFile);
- metadata.put(IndexConfiguration.ISM_POLICY_FILE, TEST_CUSTOM_INDEX_POLICY_FILE);
- final PluginSetting pluginSetting = generatePluginSettingByMetadata(metadata);
- OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- Request request = new Request(HttpMethod.HEAD, indexAlias);
- Response response = client.performRequest(request);
- MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
- final String index = String.format("%s-000001", indexAlias);
- final Map mappings = getIndexMappings(index);
- MatcherAssert.assertThat(mappings, notNullValue());
- MatcherAssert.assertThat((boolean) mappings.get("date_detection"), equalTo(false));
- sink.shutdown();
-
- final String expectedIndexPolicyName = indexAlias + "-policy";
- if (isOSBundle()) {
- // Check managed index
- await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
- MatcherAssert.assertThat(getIndexPolicyId(index), equalTo(expectedIndexPolicyName)); }
- );
- }
-
- // roll over initial index
- request = new Request(HttpMethod.POST, String.format("%s/_rollover", indexAlias));
- request.setJsonEntity("{ \"conditions\" : { } }\n");
- response = client.performRequest(request);
- MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
-
- // Instantiate sink again
- sink = createObjectUnderTest(pluginSetting, true);
- // Make sure no new write index *-000001 is created under alias
- final String rolloverIndexName = String.format("%s-000002", indexAlias);
- request = new Request(HttpMethod.GET, rolloverIndexName + "/_alias");
- response = client.performRequest(request);
- MatcherAssert.assertThat(checkIsWriteIndex(EntityUtils.toString(response.getEntity()), indexAlias, rolloverIndexName), equalTo(true));
- sink.shutdown();
-
- if (isOSBundle()) {
- // Check managed index
- MatcherAssert.assertThat(getIndexPolicyId(rolloverIndexName), equalTo(expectedIndexPolicyName));
+ @AfterEach
+ public void cleanOpenSearch() throws Exception {
+ wipeAllOpenSearchIndices();
+ wipeAllTemplates();
+ waitForClusterStateUpdatesToFinish();
}
- }
-
- @Test
- public void testInstantiateSinkDoesNotOverwriteNewerIndexTemplates() throws IOException {
- final String testIndexAlias = "test-alias";
- final String expectedIndexTemplateName = testIndexAlias + "-index-template";
- final String testTemplateFileV1 = getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE).getFile();
- final String testTemplateFileV2 = getClass().getClassLoader().getResource(TEST_TEMPLATE_V2_FILE).getFile();
- // Create sink with template version 1
- PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFileV1);
- OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ @Test
+ public void testInstantiateSinkRawSpanDefault() throws IOException {
+ final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
+ OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ final String indexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
+ Request request = new Request(HttpMethod.HEAD, indexAlias);
+ Response response = client.performRequest(request);
+ MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
+ final String index = String.format("%s-000001", indexAlias);
+ final Map mappings = getIndexMappings(index);
+ MatcherAssert.assertThat(mappings, notNullValue());
+ MatcherAssert.assertThat((boolean) mappings.get("date_detection"), equalTo(false));
+ sink.shutdown();
+
+ if (isOSBundle()) {
+ // Check managed index
+ await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
+ MatcherAssert.assertThat(getIndexPolicyId(index), equalTo(IndexConstants.RAW_ISM_POLICY));
+ }
+ );
+ }
+
+ // roll over initial index
+ request = new Request(HttpMethod.POST, String.format("%s/_rollover", indexAlias));
+ request.setJsonEntity("{ \"conditions\" : { } }\n");
+ response = client.performRequest(request);
+ MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
+
+ // Instantiate sink again
+ sink = createObjectUnderTest(pluginSetting, true);
+ // Make sure no new write index *-000001 is created under alias
+ final String rolloverIndexName = String.format("%s-000002", indexAlias);
+ request = new Request(HttpMethod.GET, rolloverIndexName + "/_alias");
+ response = client.performRequest(request);
+ MatcherAssert.assertThat(checkIsWriteIndex(EntityUtils.toString(response.getEntity()), indexAlias, rolloverIndexName), equalTo(true));
+ sink.shutdown();
+
+ if (isOSBundle()) {
+ // Check managed index
+ MatcherAssert.assertThat(getIndexPolicyId(rolloverIndexName), equalTo(IndexConstants.RAW_ISM_POLICY));
+ }
+ }
- Request getTemplateRequest = new Request(HttpMethod.GET, "/_template/" + expectedIndexTemplateName);
- Response getTemplateResponse = client.performRequest(getTemplateRequest);
- MatcherAssert.assertThat(getTemplateResponse.getStatusLine().getStatusCode(), equalTo(SC_OK));
+ @Test
+ public void testInstantiateSinkRawSpanReservedAliasAlreadyUsedAsIndex() throws IOException {
+ final String reservedIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
+ final Request request = new Request(HttpMethod.PUT, reservedIndexAlias);
+ client.performRequest(request);
+ final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
+ OpenSearchSink sink = createObjectUnderTest(pluginSetting, false);
+ Assert.assertThrows(String.format(AbstractIndexManager.INDEX_ALIAS_USED_AS_INDEX_ERROR, reservedIndexAlias),
+ RuntimeException.class, () -> sink.doInitialize());
+ }
- String responseBody = EntityUtils.toString(getTemplateResponse.getEntity());
- @SuppressWarnings("unchecked") final Integer firstResponseVersion =
- (Integer) ((Map) createContentParser(XContentType.JSON.xContent(),
- responseBody).map().get(expectedIndexTemplateName)).get("version");
+ @Test
+ public void testOutputRawSpanDefault() throws IOException, InterruptedException {
+ final String testDoc1 = readDocFromFile(DEFAULT_RAW_SPAN_FILE_1);
+ final String testDoc2 = readDocFromFile(DEFAULT_RAW_SPAN_FILE_2);
+ final ObjectMapper mapper = new ObjectMapper();
+ @SuppressWarnings("unchecked") final Map expData1 = mapper.readValue(testDoc1, Map.class);
+ @SuppressWarnings("unchecked") final Map expData2 = mapper.readValue(testDoc2, Map.class);
+
+ final List> testRecords = Arrays.asList(jsonStringToRecord(testDoc1), jsonStringToRecord(testDoc2));
+ final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+
+ final String expIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
+ final List> retSources = getSearchResponseDocSources(expIndexAlias);
+ MatcherAssert.assertThat(retSources.size(), equalTo(2));
+ MatcherAssert.assertThat(retSources, hasItems(expData1, expData2));
+ MatcherAssert.assertThat(getDocumentCount(expIndexAlias, "_id", (String) expData1.get("spanId")), equalTo(Integer.valueOf(1)));
+ sink.shutdown();
+
+ // Verify metrics
+ final List bulkRequestErrors = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_ERRORS).toString());
+ MatcherAssert.assertThat(bulkRequestErrors.size(), equalTo(1));
+ Assert.assertEquals(0.0, bulkRequestErrors.get(0).getValue(), 0);
+ final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
+ MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
+ // COUNT
+ Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
+ // TOTAL_TIME
+ Assert.assertTrue(bulkRequestLatencies.get(1).getValue() > 0.0);
+ // MAX
+ Assert.assertTrue(bulkRequestLatencies.get(2).getValue() > 0.0);
+ final List documentsSuccessMeasurements = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(BulkRetryStrategy.DOCUMENTS_SUCCESS).toString());
+ MatcherAssert.assertThat(documentsSuccessMeasurements.size(), equalTo(1));
+ MatcherAssert.assertThat(documentsSuccessMeasurements.get(0).getValue(), closeTo(2.0, 0));
+ final List documentsSuccessFirstAttemptMeasurements = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(BulkRetryStrategy.DOCUMENTS_SUCCESS_FIRST_ATTEMPT).toString());
+ MatcherAssert.assertThat(documentsSuccessFirstAttemptMeasurements.size(), equalTo(1));
+ MatcherAssert.assertThat(documentsSuccessFirstAttemptMeasurements.get(0).getValue(), closeTo(2.0, 0));
+ final List documentErrorsMeasurements = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(BulkRetryStrategy.DOCUMENT_ERRORS).toString());
+ MatcherAssert.assertThat(documentErrorsMeasurements.size(), equalTo(1));
+ MatcherAssert.assertThat(documentErrorsMeasurements.get(0).getValue(), closeTo(0.0, 0));
+
+ /**
+ * Metrics: Bulk Request Size in Bytes
+ */
+ final List bulkRequestSizeBytesMetrics = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_SIZE_BYTES).toString());
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.size(), equalTo(3));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(0).getValue(), closeTo(1.0, 0));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(1).getValue(), closeTo(2058.0, 0));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(2).getValue(), closeTo(2058.0, 0));
+ }
- MatcherAssert.assertThat(firstResponseVersion, equalTo(Integer.valueOf(1)));
- sink.shutdown();
+ @Test
+ public void testOutputRawSpanWithDLQ() throws IOException, InterruptedException {
+ // TODO: write test case
+ final String testDoc1 = readDocFromFile("raw-span-error.json");
+ final String testDoc2 = readDocFromFile(DEFAULT_RAW_SPAN_FILE_1);
+ final ObjectMapper mapper = new ObjectMapper();
+ @SuppressWarnings("unchecked") final Map expData = mapper.readValue(testDoc2, Map.class);
+
+ final List> testRecords = Arrays.asList(jsonStringToRecord(testDoc1), jsonStringToRecord(testDoc2));
+ final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
+ // generate temporary directory for dlq file
+ final File tempDirectory = Files.createTempDirectory("").toFile();
+ // add dlq file path into setting
+ final String expDLQFile = tempDirectory.getAbsolutePath() + "/test-dlq.txt";
+ pluginSetting.getSettings().put(RetryConfiguration.DLQ_FILE, expDLQFile);
+
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+ sink.shutdown();
+
+ final StringBuilder dlqContent = new StringBuilder();
+ Files.lines(Paths.get(expDLQFile)).forEach(dlqContent::append);
+ final String nonPrettyJsonString = mapper.writeValueAsString(mapper.readValue(testDoc1, JsonNode.class));
+ MatcherAssert.assertThat(dlqContent.toString(), containsString(nonPrettyJsonString));
+ final String expIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
+ final List> retSources = getSearchResponseDocSources(expIndexAlias);
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(retSources.get(0), equalTo(expData));
+
+ // clean up temporary directory
+ FileUtils.deleteQuietly(tempDirectory);
+
+ // verify metrics
+ final List documentsSuccessMeasurements = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(BulkRetryStrategy.DOCUMENTS_SUCCESS).toString());
+ MatcherAssert.assertThat(documentsSuccessMeasurements.size(), equalTo(1));
+ MatcherAssert.assertThat(documentsSuccessMeasurements.get(0).getValue(), closeTo(1.0, 0));
+ final List documentErrorsMeasurements = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(BulkRetryStrategy.DOCUMENT_ERRORS).toString());
+ MatcherAssert.assertThat(documentErrorsMeasurements.size(), equalTo(1));
+ MatcherAssert.assertThat(documentErrorsMeasurements.get(0).getValue(), closeTo(1.0, 0));
+
+ /**
+ * Metrics: Bulk Request Size in Bytes
+ */
+ final List bulkRequestSizeBytesMetrics = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_SIZE_BYTES).toString());
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.size(), equalTo(3));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(0).getValue(), closeTo(1.0, 0));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(1).getValue(), closeTo(2072.0, 0));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(2).getValue(), closeTo(2072.0, 0));
- // Create sink with template version 2
- pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFileV2);
- sink = createObjectUnderTest(pluginSetting, true);
+ }
- getTemplateRequest = new Request(HttpMethod.GET, "/_template/" + expectedIndexTemplateName);
- getTemplateResponse = client.performRequest(getTemplateRequest);
- MatcherAssert.assertThat(getTemplateResponse.getStatusLine().getStatusCode(), equalTo(SC_OK));
+ @Test
+ public void testInstantiateSinkServiceMapDefault() throws IOException {
+ final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_SERVICE_MAP.getValue(), null, null);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ final String indexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_SERVICE_MAP);
+ final Request request = new Request(HttpMethod.HEAD, indexAlias);
+ final Response response = client.performRequest(request);
+ MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
+ final Map mappings = getIndexMappings(indexAlias);
+ MatcherAssert.assertThat(mappings, notNullValue());
+ MatcherAssert.assertThat((boolean) mappings.get("date_detection"), equalTo(false));
+ sink.shutdown();
+
+ if (isOSBundle()) {
+ // Check managed index
+ MatcherAssert.assertThat(getIndexPolicyId(indexAlias), nullValue());
+ }
+ }
- responseBody = EntityUtils.toString(getTemplateResponse.getEntity());
- @SuppressWarnings("unchecked") final Integer secondResponseVersion =
- (Integer) ((Map) createContentParser(XContentType.JSON.xContent(),
- responseBody).map().get(expectedIndexTemplateName)).get("version");
+ @Test
+ public void testOutputServiceMapDefault() throws IOException, InterruptedException {
+ final String testDoc = readDocFromFile(DEFAULT_SERVICE_MAP_FILE);
+ final ObjectMapper mapper = new ObjectMapper();
+ @SuppressWarnings("unchecked") final Map expData = mapper.readValue(testDoc, Map.class);
+
+ final List> testRecords = Collections.singletonList(jsonStringToRecord(testDoc));
+ final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_SERVICE_MAP.getValue(), null, null);
+ OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+ final String expIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_SERVICE_MAP);
+ final List> retSources = getSearchResponseDocSources(expIndexAlias);
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(retSources.get(0), equalTo(expData));
+ MatcherAssert.assertThat(getDocumentCount(expIndexAlias, "_id", (String) expData.get("hashId")), equalTo(Integer.valueOf(1)));
+ sink.shutdown();
+
+ // verify metrics
+ final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
+ MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
+ // COUNT
+ MatcherAssert.assertThat(bulkRequestLatencies.get(0).getValue(), closeTo(1.0, 0));
+
+ /**
+ * Metrics: Bulk Request Size in Bytes
+ */
+ final List bulkRequestSizeBytesMetrics = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_SIZE_BYTES).toString());
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.size(), equalTo(3));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(0).getValue(), closeTo(1.0, 0));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(1).getValue(), closeTo(265.0, 0));
+ MatcherAssert.assertThat(bulkRequestSizeBytesMetrics.get(2).getValue(), closeTo(265.0, 0));
+
+ // Check restart for index already exists
+ sink = createObjectUnderTest(pluginSetting, true);
+ sink.shutdown();
+ }
- MatcherAssert.assertThat(secondResponseVersion, equalTo(Integer.valueOf(2)));
- sink.shutdown();
+ @Test
+ public void testInstantiateSinkCustomIndex_NoRollOver() throws IOException {
+ final String testIndexAlias = "test-alias";
+ final String testTemplateFile = Objects.requireNonNull(
+ getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE)).getFile();
+ final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFile);
+ OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ final Request request = new Request(HttpMethod.HEAD, testIndexAlias);
+ final Response response = client.performRequest(request);
+ MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
+ sink.shutdown();
+
+ // Check restart for index already exists
+ sink = createObjectUnderTest(pluginSetting, true);
+ sink.shutdown();
+ }
- // Create sink with template version 1 again
- pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFileV1);
- sink = createObjectUnderTest(pluginSetting, true);
+ @Test
+ public void testInstantiateSinkCustomIndex_WithIsmPolicy() throws IOException {
+ final String indexAlias = "sink-custom-index-ism-test-alias";
+ final String testTemplateFile = Objects.requireNonNull(
+ getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE)).getFile();
+ final Map metadata = initializeConfigurationMetadata(null, indexAlias, testTemplateFile);
+ metadata.put(IndexConfiguration.ISM_POLICY_FILE, TEST_CUSTOM_INDEX_POLICY_FILE);
+ final PluginSetting pluginSetting = generatePluginSettingByMetadata(metadata);
+ OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ Request request = new Request(HttpMethod.HEAD, indexAlias);
+ Response response = client.performRequest(request);
+ MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
+ final String index = String.format("%s-000001", indexAlias);
+ final Map mappings = getIndexMappings(index);
+ MatcherAssert.assertThat(mappings, notNullValue());
+ MatcherAssert.assertThat((boolean) mappings.get("date_detection"), equalTo(false));
+ sink.shutdown();
+
+ final String expectedIndexPolicyName = indexAlias + "-policy";
+ if (isOSBundle()) {
+ // Check managed index
+ await().atMost(1, TimeUnit.SECONDS).untilAsserted(() -> {
+ MatcherAssert.assertThat(getIndexPolicyId(index), equalTo(expectedIndexPolicyName));
+ }
+ );
+ }
+
+ // roll over initial index
+ request = new Request(HttpMethod.POST, String.format("%s/_rollover", indexAlias));
+ request.setJsonEntity("{ \"conditions\" : { } }\n");
+ response = client.performRequest(request);
+ MatcherAssert.assertThat(response.getStatusLine().getStatusCode(), equalTo(SC_OK));
+
+ // Instantiate sink again
+ sink = createObjectUnderTest(pluginSetting, true);
+ // Make sure no new write index *-000001 is created under alias
+ final String rolloverIndexName = String.format("%s-000002", indexAlias);
+ request = new Request(HttpMethod.GET, rolloverIndexName + "/_alias");
+ response = client.performRequest(request);
+ MatcherAssert.assertThat(checkIsWriteIndex(EntityUtils.toString(response.getEntity()), indexAlias, rolloverIndexName), equalTo(true));
+ sink.shutdown();
+
+ if (isOSBundle()) {
+ // Check managed index
+ MatcherAssert.assertThat(getIndexPolicyId(rolloverIndexName), equalTo(expectedIndexPolicyName));
+ }
+ }
- getTemplateRequest = new Request(HttpMethod.GET, "/_template/" + expectedIndexTemplateName);
- getTemplateResponse = client.performRequest(getTemplateRequest);
- MatcherAssert.assertThat(getTemplateResponse.getStatusLine().getStatusCode(), equalTo(SC_OK));
+ @ParameterizedTest
+ @ArgumentsSource(CreateWithTemplatesArgumentsProvider.class)
+ public void testInstantiateSinkDoesNotOverwriteNewerIndexTemplates(
+ final String templateType,
+ final String templatePath,
+ final String v1File,
+ final String v2File,
+ final BiFunction, String, Integer> extractVersionFunction) throws IOException {
+ final String testIndexAlias = "test-alias";
+ final String expectedIndexTemplateName = testIndexAlias + "-index-template";
+ final String testTemplateFileV1 = getClass().getClassLoader().getResource(v1File).getFile();
+ final String testTemplateFileV2 = getClass().getClassLoader().getResource(v2File).getFile();
+
+ // Create sink with template version 1
+ PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, templateType, testTemplateFileV1);
+ OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+
+ Request getTemplateRequest = new Request(HttpMethod.GET, "/" + templatePath + "/" + expectedIndexTemplateName);
+ Response getTemplateResponse = client.performRequest(getTemplateRequest);
+ MatcherAssert.assertThat(getTemplateResponse.getStatusLine().getStatusCode(), equalTo(SC_OK));
+
+ String responseBody = EntityUtils.toString(getTemplateResponse.getEntity());
+ @SuppressWarnings("unchecked") final Integer firstResponseVersion =
+ extractVersionFunction.apply(createContentParser(XContentType.JSON.xContent(),
+ responseBody).map(), expectedIndexTemplateName);
+
+ MatcherAssert.assertThat(firstResponseVersion, equalTo(Integer.valueOf(1)));
+ sink.shutdown();
+
+ // Create sink with template version 2
+ pluginSetting = generatePluginSetting(null, testIndexAlias, templateType, testTemplateFileV2);
+ sink = createObjectUnderTest(pluginSetting, true);
+
+ getTemplateRequest = new Request(HttpMethod.GET, "/" + templatePath + "/" + expectedIndexTemplateName);
+ getTemplateResponse = client.performRequest(getTemplateRequest);
+ MatcherAssert.assertThat(getTemplateResponse.getStatusLine().getStatusCode(), equalTo(SC_OK));
+
+ responseBody = EntityUtils.toString(getTemplateResponse.getEntity());
+ @SuppressWarnings("unchecked") final Integer secondResponseVersion =
+ extractVersionFunction.apply(createContentParser(XContentType.JSON.xContent(),
+ responseBody).map(), expectedIndexTemplateName);
+
+ MatcherAssert.assertThat(secondResponseVersion, equalTo(Integer.valueOf(2)));
+ sink.shutdown();
+
+ // Create sink with template version 1 again
+ pluginSetting = generatePluginSetting(null, testIndexAlias, templateType, testTemplateFileV1);
+ sink = createObjectUnderTest(pluginSetting, true);
+
+ getTemplateRequest = new Request(HttpMethod.GET, "/" + templatePath + "/" + expectedIndexTemplateName);
+ getTemplateResponse = client.performRequest(getTemplateRequest);
+ MatcherAssert.assertThat(getTemplateResponse.getStatusLine().getStatusCode(), equalTo(SC_OK));
+
+ responseBody = EntityUtils.toString(getTemplateResponse.getEntity());
+ @SuppressWarnings("unchecked") final Integer thirdResponseVersion =
+ extractVersionFunction.apply(createContentParser(XContentType.JSON.xContent(),
+ responseBody).map(), expectedIndexTemplateName);
+
+ // Assert version 2 was not overwritten by version 1
+ MatcherAssert.assertThat(thirdResponseVersion, equalTo(Integer.valueOf(2)));
+ sink.shutdown();
- responseBody = EntityUtils.toString(getTemplateResponse.getEntity());
- @SuppressWarnings("unchecked") final Integer thirdResponseVersion =
- (Integer) ((Map) createContentParser(XContentType.JSON.xContent(),
- responseBody).map().get(expectedIndexTemplateName)).get("version");
+ }
- // Assert version 2 was not overwritten by version 1
- MatcherAssert.assertThat(thirdResponseVersion, equalTo(Integer.valueOf(2)));
- sink.shutdown();
+ static class CreateWithTemplatesArgumentsProvider implements ArgumentsProvider {
+ @Override
+ public Stream provideArguments(ExtensionContext context) {
+ final List arguments = new ArrayList<>();
+ arguments.add(
+ arguments("v1", "_template",
+ TEST_TEMPLATE_V1_FILE, TEST_TEMPLATE_V2_FILE,
+ (BiFunction, String, Integer>) (map, templateName) ->
+ (Integer) ((Map) map.get(templateName)).get("version")
+ )
+ );
+
+ if(OpenSearchIntegrationHelper.getVersion().compareTo(DeclaredOpenSearchVersion.OPENDISTRO_1_9) >= 0) {
+ arguments.add(
+ arguments("index-template", "_index_template",
+ TEST_INDEX_TEMPLATE_V1_FILE, TEST_INDEX_TEMPLATE_V2_FILE,
+ (BiFunction, String, Integer>) (map, unused) ->
+ (Integer) ((List>>) map.get("index_templates")).get(0).get("index_template").get("version")
+ )
+ );
+ }
+ return arguments.stream();
+ }
+ }
- }
+ @Test
+ public void testOutputCustomIndex() throws IOException, InterruptedException {
+ final String testIndexAlias = "test-alias";
+ final String testTemplateFile = Objects.requireNonNull(
+ getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE)).getFile();
+ final String testIdField = "someId";
+ final String testId = "foo";
+ final List> testRecords = Collections.singletonList(jsonStringToRecord(generateCustomRecordJson(testIdField, testId)));
+ final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFile);
+ pluginSetting.getSettings().put(IndexConfiguration.DOCUMENT_ID_FIELD, testIdField);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+ final List> retSources = getSearchResponseDocSources(testIndexAlias);
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(getDocumentCount(testIndexAlias, "_id", testId), equalTo(Integer.valueOf(1)));
+ sink.shutdown();
+
+ // verify metrics
+ final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
+ MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
+ // COUNT
+ Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
+ }
- @Test
- public void testOutputCustomIndex() throws IOException, InterruptedException {
- final String testIndexAlias = "test-alias";
- final String testTemplateFile = Objects.requireNonNull(
- getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE)).getFile();
- final String testIdField = "someId";
- final String testId = "foo";
- final List> testRecords = Collections.singletonList(jsonStringToRecord(generateCustomRecordJson(testIdField, testId)));
- final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFile);
- pluginSetting.getSettings().put(IndexConfiguration.DOCUMENT_ID_FIELD, testIdField);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
- final List> retSources = getSearchResponseDocSources(testIndexAlias);
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(getDocumentCount(testIndexAlias, "_id", testId), equalTo(Integer.valueOf(1)));
- sink.shutdown();
-
- // verify metrics
- final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
- MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
- // COUNT
- Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
- }
+ @Test
+ public void testBulkActionCreate() throws IOException, InterruptedException {
+ final String testIndexAlias = "test-alias";
+ final String testTemplateFile = Objects.requireNonNull(
+ getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE)).getFile();
+ final String testIdField = "someId";
+ final String testId = "foo";
+ final List> testRecords = Collections.singletonList(jsonStringToRecord(generateCustomRecordJson(testIdField, testId)));
+ final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFile);
+ pluginSetting.getSettings().put(IndexConfiguration.DOCUMENT_ID_FIELD, testIdField);
+ pluginSetting.getSettings().put(IndexConfiguration.ACTION, BulkAction.CREATE.toString());
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+ final List> retSources = getSearchResponseDocSources(testIndexAlias);
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(getDocumentCount(testIndexAlias, "_id", testId), equalTo(Integer.valueOf(1)));
+ sink.shutdown();
+
+ // verify metrics
+ final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
+ MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
+ // COUNT
+ Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
+ }
- @Test
- public void testBulkActionCreate() throws IOException, InterruptedException {
- final String testIndexAlias = "test-alias";
- final String testTemplateFile = Objects.requireNonNull(
- getClass().getClassLoader().getResource(TEST_TEMPLATE_V1_FILE)).getFile();
- final String testIdField = "someId";
- final String testId = "foo";
- final List> testRecords = Collections.singletonList(jsonStringToRecord(generateCustomRecordJson(testIdField, testId)));
- final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, testTemplateFile);
- pluginSetting.getSettings().put(IndexConfiguration.DOCUMENT_ID_FIELD, testIdField);
- pluginSetting.getSettings().put(IndexConfiguration.ACTION, BulkAction.CREATE.toString());
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
- final List> retSources = getSearchResponseDocSources(testIndexAlias);
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(getDocumentCount(testIndexAlias, "_id", testId), equalTo(Integer.valueOf(1)));
- sink.shutdown();
-
- // verify metrics
- final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
- MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
- // COUNT
- Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
- }
+ @Test
+ public void testEventOutput() throws IOException, InterruptedException {
- @Test
- public void testEventOutput() throws IOException, InterruptedException {
+ final Event testEvent = JacksonEvent.builder()
+ .withData("{\"log\": \"foobar\"}")
+ .withEventType("event")
+ .build();
+ ((JacksonEvent) testEvent).setEventHandle(eventHandle);
- final Event testEvent = JacksonEvent.builder()
- .withData("{\"log\": \"foobar\"}")
- .withEventType("event")
- .build();
- ((JacksonEvent)testEvent).setEventHandle(eventHandle);
+ final List> testRecords = Collections.singletonList(new Record<>(testEvent));
- final List> testRecords = Collections.singletonList(new Record<>(testEvent));
+ final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
- final PluginSetting pluginSetting = generatePluginSetting(IndexType.TRACE_ANALYTICS_RAW.getValue(), null, null);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
+ final String expIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
+ final List> retSources = getSearchResponseDocSources(expIndexAlias);
+ final Map expectedContent = new HashMap<>();
+ expectedContent.put("log", "foobar");
- final String expIndexAlias = IndexConstants.TYPE_TO_DEFAULT_ALIAS.get(IndexType.TRACE_ANALYTICS_RAW);
- final List> retSources = getSearchResponseDocSources(expIndexAlias);
- final Map expectedContent = new HashMap<>();
- expectedContent.put("log", "foobar");
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(retSources.containsAll(Arrays.asList(expectedContent)), equalTo(true));
+ MatcherAssert.assertThat(getDocumentCount(expIndexAlias, "log", "foobar"), equalTo(Integer.valueOf(1)));
+ sink.shutdown();
+ }
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(retSources.containsAll(Arrays.asList(expectedContent)), equalTo(true));
- MatcherAssert.assertThat(getDocumentCount(expIndexAlias, "log", "foobar"), equalTo(Integer.valueOf(1)));
- sink.shutdown();
- }
+ @ParameterizedTest
+ @ValueSource(strings = {"info/ids/id", "id"})
+ public void testOpenSearchDocumentId(final String testDocumentIdField) throws IOException, InterruptedException {
+ final String expectedId = UUID.randomUUID().toString();
+ final String testIndexAlias = "test_index";
+ final Event testEvent = JacksonEvent.builder()
+ .withData(Map.of("arbitrary_data", UUID.randomUUID().toString()))
+ .withEventType("event")
+ .build();
+ ((JacksonEvent) testEvent).setEventHandle(eventHandle);
+ testEvent.put(testDocumentIdField, expectedId);
+
+ final List> testRecords = Collections.singletonList(new Record<>(testEvent));
+
+ final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, null);
+ pluginSetting.getSettings().put(IndexConfiguration.DOCUMENT_ID_FIELD, testDocumentIdField);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+
+ final List docIds = getSearchResponseDocIds(testIndexAlias);
+ for (String docId : docIds) {
+ MatcherAssert.assertThat(docId, equalTo(expectedId));
+ }
+ sink.shutdown();
+ }
- @ParameterizedTest
- @ValueSource(strings = {"info/ids/id", "id"})
- public void testOpenSearchDocumentId(final String testDocumentIdField) throws IOException, InterruptedException {
- final String expectedId = UUID.randomUUID().toString();
- final String testIndexAlias = "test_index";
- final Event testEvent = JacksonEvent.builder()
- .withData(Map.of("arbitrary_data", UUID.randomUUID().toString()))
- .withEventType("event")
- .build();
- ((JacksonEvent)testEvent).setEventHandle(eventHandle);
- testEvent.put(testDocumentIdField, expectedId);
-
- final List> testRecords = Collections.singletonList(new Record<>(testEvent));
-
- final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, null);
- pluginSetting.getSettings().put(IndexConfiguration.DOCUMENT_ID_FIELD, testDocumentIdField);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
-
- final List docIds = getSearchResponseDocIds(testIndexAlias);
- for (String docId: docIds) {
- MatcherAssert.assertThat(docId, equalTo(expectedId));
- }
- sink.shutdown();
- }
+ @ParameterizedTest
+ @ValueSource(strings = {"info/ids/rid", "rid"})
+ public void testOpenSearchRoutingField(final String testRoutingField) throws IOException, InterruptedException {
+ final String expectedRoutingField = UUID.randomUUID().toString();
+ final String testIndexAlias = "test_index";
+ final Event testEvent = JacksonEvent.builder()
+ .withData(Map.of("arbitrary_data", UUID.randomUUID().toString()))
+ .withEventType("event")
+ .build();
+ ((JacksonEvent) testEvent).setEventHandle(eventHandle);
+ testEvent.put(testRoutingField, expectedRoutingField);
+
+ final List> testRecords = Collections.singletonList(new Record<>(testEvent));
+
+ final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, null);
+ pluginSetting.getSettings().put(IndexConfiguration.ROUTING_FIELD, testRoutingField);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+
+ final List routingFields = getSearchResponseRoutingFields(testIndexAlias);
+ for (String routingField : routingFields) {
+ MatcherAssert.assertThat(routingField, equalTo(expectedRoutingField));
+ }
+ sink.shutdown();
+ }
- @ParameterizedTest
- @ValueSource(strings = {"info/ids/rid", "rid"})
- public void testOpenSearchRoutingField(final String testRoutingField) throws IOException, InterruptedException {
- final String expectedRoutingField = UUID.randomUUID().toString();
- final String testIndexAlias = "test_index";
- final Event testEvent = JacksonEvent.builder()
- .withData(Map.of("arbitrary_data", UUID.randomUUID().toString()))
- .withEventType("event")
- .build();
- ((JacksonEvent)testEvent).setEventHandle(eventHandle);
- testEvent.put(testRoutingField, expectedRoutingField);
-
- final List> testRecords = Collections.singletonList(new Record<>(testEvent));
-
- final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, null);
- pluginSetting.getSettings().put(IndexConfiguration.ROUTING_FIELD, testRoutingField);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
-
- final List routingFields = getSearchResponseRoutingFields(testIndexAlias);
- for (String routingField: routingFields) {
- MatcherAssert.assertThat(routingField, equalTo(expectedRoutingField));
- }
- sink.shutdown();
- }
+ @ParameterizedTest
+ @ValueSource(strings = {"info/ids/id", "id"})
+ public void testOpenSearchDynamicIndex(final String testIndex) throws IOException, InterruptedException {
+ final String dynamicTestIndexAlias = "test-${" + testIndex + "}-index";
+ final String testIndexName = "idx1";
+ final String testIndexAlias = "test-" + testIndexName + "-index";
+ final String data = UUID.randomUUID().toString();
+ final Map dataMap = Map.of("data", data);
+ final Event testEvent = JacksonEvent.builder()
+ .withData(dataMap)
+ .withEventType("event")
+ .build();
+ ((JacksonEvent) testEvent).setEventHandle(eventHandle);
+ testEvent.put(testIndex, testIndexName);
+
+ Map expectedMap = testEvent.toMap();
+
+ final List> testRecords = Collections.singletonList(new Record<>(testEvent));
+
+ final PluginSetting pluginSetting = generatePluginSetting(null, dynamicTestIndexAlias, null);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+ final List> retSources = getSearchResponseDocSources(testIndexAlias);
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(retSources, hasItem(expectedMap));
+ sink.shutdown();
+ }
- @ParameterizedTest
- @ValueSource(strings = {"info/ids/id", "id"})
- public void testOpenSearchDynamicIndex(final String testIndex) throws IOException, InterruptedException {
- final String dynamicTestIndexAlias = "test-${"+testIndex+"}-index";
- final String testIndexName = "idx1";
- final String testIndexAlias = "test-"+testIndexName+"-index";
- final String data = UUID.randomUUID().toString();
- final Map dataMap = Map.of("data", data);
- final Event testEvent = JacksonEvent.builder()
- .withData(dataMap)
- .withEventType("event")
- .build();
- ((JacksonEvent)testEvent).setEventHandle(eventHandle);
- testEvent.put(testIndex, testIndexName);
-
- Map expectedMap = testEvent.toMap();
-
- final List> testRecords = Collections.singletonList(new Record<>(testEvent));
-
- final PluginSetting pluginSetting = generatePluginSetting(null, dynamicTestIndexAlias, null);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
- final List> retSources = getSearchResponseDocSources(testIndexAlias);
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(retSources, hasItem(expectedMap));
- sink.shutdown();
- }
+ @ParameterizedTest
+ @CsvSource({
+ "info/ids/id, yyyy-MM",
+ "id, yyyy-MM-dd",
+ })
+ public void testOpenSearchDynamicIndexWithDate(final String testIndex, final String testDatePattern) throws IOException, InterruptedException {
+ final String dynamicTestIndexAlias = "test-${" + testIndex + "}-index-%{" + testDatePattern + "}";
+ final String testIndexName = "idx1";
+ SimpleDateFormat formatter = new SimpleDateFormat(testDatePattern);
+ Date date = new Date();
+ String expectedDate = formatter.format(date);
+ final String expectedIndexAlias = "test-" + testIndexName + "-index-" + expectedDate;
+ final String data = UUID.randomUUID().toString();
+ final Map dataMap = Map.of("data", data);
+ final Event testEvent = JacksonEvent.builder()
+ .withData(dataMap)
+ .withEventType("event")
+ .build();
+ ((JacksonEvent) testEvent).setEventHandle(eventHandle);
+ testEvent.put(testIndex, testIndexName);
+
+ Map expectedMap = testEvent.toMap();
+
+ final List> testRecords = Collections.singletonList(new Record<>(testEvent));
+
+ final PluginSetting pluginSetting = generatePluginSetting(null, dynamicTestIndexAlias, null);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+ final List> retSources = getSearchResponseDocSources(expectedIndexAlias);
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(retSources, hasItem(expectedMap));
+ sink.shutdown();
+ }
- @ParameterizedTest
- @CsvSource({
- "info/ids/id, yyyy-MM",
- "id, yyyy-MM-dd",
- })
- public void testOpenSearchDynamicIndexWithDate(final String testIndex, final String testDatePattern) throws IOException, InterruptedException {
- final String dynamicTestIndexAlias = "test-${"+testIndex+"}-index-%{"+testDatePattern+"}";
- final String testIndexName = "idx1";
- SimpleDateFormat formatter = new SimpleDateFormat(testDatePattern);
- Date date = new Date();
- String expectedDate = formatter.format(date);
- final String expectedIndexAlias = "test-"+testIndexName+"-index-"+expectedDate;
- final String data = UUID.randomUUID().toString();
- final Map dataMap = Map.of("data", data);
- final Event testEvent = JacksonEvent.builder()
- .withData(dataMap)
- .withEventType("event")
- .build();
- ((JacksonEvent)testEvent).setEventHandle(eventHandle);
- testEvent.put(testIndex, testIndexName);
-
- Map expectedMap = testEvent.toMap();
-
- final List> testRecords = Collections.singletonList(new Record<>(testEvent));
-
- final PluginSetting pluginSetting = generatePluginSetting(null, dynamicTestIndexAlias, null);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
- final List> retSources = getSearchResponseDocSources(expectedIndexAlias);
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(retSources, hasItem(expectedMap));
- sink.shutdown();
- }
+ @ParameterizedTest
+ @ValueSource(strings = {"yyyy-MM", "yyyy-MM-dd", "dd-MM-yyyy"})
+ public void testOpenSearchIndexWithDate(final String testDatePattern) throws IOException, InterruptedException {
+ SimpleDateFormat formatter = new SimpleDateFormat(testDatePattern);
+ Date date = new Date();
+ String expectedIndexName = "test-index-" + formatter.format(date);
+ final String testIndexName = "idx1";
+ final String testIndexAlias = "test-index-%{" + testDatePattern + "}";
+ final String data = UUID.randomUUID().toString();
+ final Map dataMap = Map.of("data", data);
+ final Event testEvent = JacksonEvent.builder()
+ .withData(dataMap)
+ .withEventType("event")
+ .build();
+ ((JacksonEvent) testEvent).setEventHandle(eventHandle);
+
+ Map expectedMap = testEvent.toMap();
+
+ final List> testRecords = Collections.singletonList(new Record<>(testEvent));
+
+ final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, null);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+ sink.output(testRecords);
+ final List> retSources = getSearchResponseDocSources(expectedIndexName);
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(retSources, hasItem(expectedMap));
+ sink.shutdown();
+ }
- @ParameterizedTest
- @ValueSource(strings = {"yyyy-MM", "yyyy-MM-dd", "dd-MM-yyyy"})
- public void testOpenSearchIndexWithDate(final String testDatePattern) throws IOException, InterruptedException {
- SimpleDateFormat formatter = new SimpleDateFormat(testDatePattern);
- Date date = new Date();
- String expectedIndexName = "test-index-"+formatter.format(date);
- final String testIndexName = "idx1";
- final String testIndexAlias = "test-index-%{"+testDatePattern+"}";
- final String data = UUID.randomUUID().toString();
- final Map dataMap = Map.of("data", data);
- final Event testEvent = JacksonEvent.builder()
- .withData(dataMap)
- .withEventType("event")
- .build();
- ((JacksonEvent)testEvent).setEventHandle(eventHandle);
-
- Map expectedMap = testEvent.toMap();
-
- final List> testRecords = Collections.singletonList(new Record<>(testEvent));
-
- final PluginSetting pluginSetting = generatePluginSetting(null, testIndexAlias, null);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
- sink.output(testRecords);
- final List> retSources = getSearchResponseDocSources(expectedIndexName);
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(retSources, hasItem(expectedMap));
- sink.shutdown();
- }
+ @Test
+ public void testOpenSearchIndexWithInvalidDate() throws IOException, InterruptedException {
+ String invalidDatePattern = "yyyy-MM-dd HH:ss:mm";
+ final String invalidTestIndexAlias = "test-index-%{" + invalidDatePattern + "}";
+ final PluginSetting pluginSetting = generatePluginSetting(null, invalidTestIndexAlias, null);
+ OpenSearchSink sink = createObjectUnderTest(pluginSetting, false);
+ Assert.assertThrows(IllegalArgumentException.class, () -> sink.doInitialize());
+ }
- @Test
- public void testOpenSearchIndexWithInvalidDate() throws IOException, InterruptedException {
- String invalidDatePattern = "yyyy-MM-dd HH:ss:mm";
- final String invalidTestIndexAlias = "test-index-%{"+invalidDatePattern+"}";
- final PluginSetting pluginSetting = generatePluginSetting(null, invalidTestIndexAlias, null);
- OpenSearchSink sink = createObjectUnderTest(pluginSetting, false);
- Assert.assertThrows(IllegalArgumentException.class, () -> sink.doInitialize());
- }
+ @Test
+ public void testOpenSearchIndexWithInvalidChars() throws IOException, InterruptedException {
+ final String invalidTestIndexAlias = "test#-index";
+ final PluginSetting pluginSetting = generatePluginSetting(null, invalidTestIndexAlias, null);
+ OpenSearchSink sink = createObjectUnderTest(pluginSetting, false);
+ Assert.assertThrows(RuntimeException.class, () -> sink.doInitialize());
+ }
- @Test
- public void testOpenSearchIndexWithInvalidChars() throws IOException, InterruptedException {
- final String invalidTestIndexAlias = "test#-index";
- final PluginSetting pluginSetting = generatePluginSetting(null, invalidTestIndexAlias, null);
- OpenSearchSink sink = createObjectUnderTest(pluginSetting, false);
- Assert.assertThrows(RuntimeException.class, () -> sink.doInitialize());
- }
+ @Test
+ @Timeout(value = 1, unit = TimeUnit.MINUTES)
+ public void testOutputManagementDisabled() throws IOException, InterruptedException {
+ final String testIndexAlias = "test-" + UUID.randomUUID();
+ final String roleName = UUID.randomUUID().toString();
+ final String username = UUID.randomUUID().toString();
+ final String password = UUID.randomUUID().toString();
+ final OpenSearchSecurityAccessor securityAccessor = new OpenSearchSecurityAccessor(client);
+ securityAccessor.createBulkWritingRole(roleName, testIndexAlias + "*");
+ securityAccessor.createUser(username, password, roleName);
+
+ final String testIdField = "someId";
+ final String testId = "foo";
+
+ final List> testRecords = Collections.singletonList(jsonStringToRecord(generateCustomRecordJson(testIdField, testId)));
+
+ final Map metadata = initializeConfigurationMetadata(null, testIndexAlias, null);
+ metadata.put(IndexConfiguration.INDEX_TYPE, IndexType.MANAGEMENT_DISABLED.getValue());
+ metadata.put(ConnectionConfiguration.USERNAME, username);
+ metadata.put(ConnectionConfiguration.PASSWORD, password);
+ metadata.put(IndexConfiguration.DOCUMENT_ID_FIELD, testIdField);
+ final PluginSetting pluginSetting = generatePluginSettingByMetadata(metadata);
+ final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
+
+ final String testTemplateFile = Objects.requireNonNull(
+ getClass().getClassLoader().getResource("management-disabled-index-template.json")).getFile();
+ createV1IndexTemplate(testIndexAlias, testIndexAlias + "*", testTemplateFile);
+ createIndex(testIndexAlias);
+
+ sink.output(testRecords);
+ final List> retSources = getSearchResponseDocSources(testIndexAlias);
+ MatcherAssert.assertThat(retSources.size(), equalTo(1));
+ MatcherAssert.assertThat(getDocumentCount(testIndexAlias, "_id", testId), equalTo(Integer.valueOf(1)));
+ sink.shutdown();
+
+ // verify metrics
+ final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
+ new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
+ .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
+ MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
+ // COUNT
+ Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
+ }
- @Test
- @Timeout(value = 1, unit = TimeUnit.MINUTES)
- public void testOutputManagementDisabled() throws IOException, InterruptedException {
- final String testIndexAlias = "test-" + UUID.randomUUID();
- final String roleName = UUID.randomUUID().toString();
- final String username = UUID.randomUUID().toString();
- final String password = UUID.randomUUID().toString();
- final OpenSearchSecurityAccessor securityAccessor = new OpenSearchSecurityAccessor(client);
- securityAccessor.createBulkWritingRole(roleName, testIndexAlias + "*");
- securityAccessor.createUser(username, password, roleName);
-
- final String testIdField = "someId";
- final String testId = "foo";
-
- final List> testRecords = Collections.singletonList(jsonStringToRecord(generateCustomRecordJson(testIdField, testId)));
-
- final Map metadata = initializeConfigurationMetadata(null, testIndexAlias, null);
- metadata.put(IndexConfiguration.INDEX_TYPE, IndexType.MANAGEMENT_DISABLED.getValue());
- metadata.put(ConnectionConfiguration.USERNAME, username);
- metadata.put(ConnectionConfiguration.PASSWORD, password);
- metadata.put(IndexConfiguration.DOCUMENT_ID_FIELD, testIdField);
- final PluginSetting pluginSetting = generatePluginSettingByMetadata(metadata);
- final OpenSearchSink sink = createObjectUnderTest(pluginSetting, true);
-
- final String testTemplateFile = Objects.requireNonNull(
- getClass().getClassLoader().getResource("management-disabled-index-template.json")).getFile();
- createIndexTemplate(testIndexAlias, testIndexAlias + "*", testTemplateFile);
- createIndex(testIndexAlias);
-
- sink.output(testRecords);
- final List> retSources = getSearchResponseDocSources(testIndexAlias);
- MatcherAssert.assertThat(retSources.size(), equalTo(1));
- MatcherAssert.assertThat(getDocumentCount(testIndexAlias, "_id", testId), equalTo(Integer.valueOf(1)));
- sink.shutdown();
-
- // verify metrics
- final List bulkRequestLatencies = MetricsTestUtil.getMeasurementList(
- new StringJoiner(MetricNames.DELIMITER).add(PIPELINE_NAME).add(PLUGIN_NAME)
- .add(OpenSearchSink.BULKREQUEST_LATENCY).toString());
- MatcherAssert.assertThat(bulkRequestLatencies.size(), equalTo(3));
- // COUNT
- Assert.assertEquals(1.0, bulkRequestLatencies.get(0).getValue(), 0);
- }
+ private Map initializeConfigurationMetadata(final String indexType, final String indexAlias,
+ final String templateFilePath) {
+ final Map metadata = new HashMap<>();
+ metadata.put(IndexConfiguration.INDEX_TYPE, indexType);
+ metadata.put(ConnectionConfiguration.HOSTS, getHosts());
+ metadata.put(IndexConfiguration.INDEX_ALIAS, indexAlias);
+ metadata.put(IndexConfiguration.TEMPLATE_FILE, templateFilePath);
+ final String user = System.getProperty("tests.opensearch.user");
+ final String password = System.getProperty("tests.opensearch.password");
+ if (user != null) {
+ metadata.put(ConnectionConfiguration.USERNAME, user);
+ metadata.put(ConnectionConfiguration.PASSWORD, password);
+ }
+ return metadata;
+ }
- private Map initializeConfigurationMetadata (final String indexType, final String indexAlias,
- final String templateFilePath) {
- final Map metadata = new HashMap<>();
- metadata.put(IndexConfiguration.INDEX_TYPE, indexType);
- metadata.put(ConnectionConfiguration.HOSTS, getHosts());
- metadata.put(IndexConfiguration.INDEX_ALIAS, indexAlias);
- metadata.put(IndexConfiguration.TEMPLATE_FILE, templateFilePath);
- final String user = System.getProperty("tests.opensearch.user");
- final String password = System.getProperty("tests.opensearch.password");
- if (user != null) {
- metadata.put(ConnectionConfiguration.USERNAME, user);
- metadata.put(ConnectionConfiguration.PASSWORD, password);
- }
- return metadata;
- }
+ private PluginSetting generatePluginSetting(final String indexType, final String indexAlias,
+ final String templateFilePath) {
+ final Map metadata = initializeConfigurationMetadata(indexType, indexAlias, templateFilePath);
+ return generatePluginSettingByMetadata(metadata);
+ }
- private PluginSetting generatePluginSetting(final String indexType, final String indexAlias,
- final String templateFilePath) {
- final Map metadata = initializeConfigurationMetadata(indexType, indexAlias, templateFilePath);
- return generatePluginSettingByMetadata(metadata);
- }
+ private PluginSetting generatePluginSetting(final String indexType, final String indexAlias,
+ final String templateType,
+ final String templateFilePath) {
+ final Map metadata = initializeConfigurationMetadata(indexType, indexAlias, templateFilePath);
+ metadata.put(IndexConfiguration.TEMPLATE_TYPE, templateType);
+ return generatePluginSettingByMetadata(metadata);
+ }
- private PluginSetting generatePluginSettingByMetadata(final Map configurationMetadata) {
- final PluginSetting pluginSetting = new PluginSetting(PLUGIN_NAME, configurationMetadata);
- pluginSetting.setPipelineName(PIPELINE_NAME);
- return pluginSetting;
- }
+ private PluginSetting generatePluginSettingByMetadata(final Map configurationMetadata) {
+ final PluginSetting pluginSetting = new PluginSetting(PLUGIN_NAME, configurationMetadata);
+ pluginSetting.setPipelineName(PIPELINE_NAME);
+ return pluginSetting;
+ }
- private String generateCustomRecordJson(final String idField, final String documentId) throws IOException {
- return Strings.toString(
- XContentFactory.jsonBuilder()
- .startObject()
- .field(idField, documentId)
- .endObject()
- );
- }
+ private String generateCustomRecordJson(final String idField, final String documentId) throws IOException {
+ return Strings.toString(
+ XContentFactory.jsonBuilder()
+ .startObject()
+ .field(idField, documentId)
+ .endObject()
+ );
+ }
- private String readDocFromFile(final String filename) throws IOException {
- final StringBuilder jsonBuilder = new StringBuilder();
- try (final InputStream inputStream = Objects.requireNonNull(
- getClass().getClassLoader().getResourceAsStream(filename))) {
- final BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
- bufferedReader.lines().forEach(jsonBuilder::append);
+ private String readDocFromFile(final String filename) throws IOException {
+ final StringBuilder jsonBuilder = new StringBuilder();
+ try (final InputStream inputStream = Objects.requireNonNull(
+ getClass().getClassLoader().getResourceAsStream(filename))) {
+ final BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
+ bufferedReader.lines().forEach(jsonBuilder::append);
+ }
+ return jsonBuilder.toString();
}
- return jsonBuilder.toString();
- }
- private Boolean checkIsWriteIndex(final String responseBody, final String aliasName, final String indexName) throws IOException {
- @SuppressWarnings("unchecked") final Map indexBlob = (Map) createContentParser(XContentType.JSON.xContent(),
- responseBody).map().get(indexName);
- @SuppressWarnings("unchecked") final Map aliasesBlob = (Map) indexBlob.get("aliases");
- @SuppressWarnings("unchecked") final Map aliasBlob = (Map) aliasesBlob.get(aliasName);
- return (Boolean) aliasBlob.get("is_write_index");
- }
+ private Boolean checkIsWriteIndex(final String responseBody, final String aliasName, final String indexName) throws IOException {
+ @SuppressWarnings("unchecked") final Map indexBlob = (Map) createContentParser(XContentType.JSON.xContent(),
+ responseBody).map().get(indexName);
+ @SuppressWarnings("unchecked") final Map aliasesBlob = (Map) indexBlob.get("aliases");
+ @SuppressWarnings("unchecked") final Map aliasBlob = (Map) aliasesBlob.get(aliasName);
+ return (Boolean) aliasBlob.get("is_write_index");
+ }
- private Integer getDocumentCount(final String index, final String field, final String value) throws IOException, InterruptedException {
- final Request request = new Request(HttpMethod.GET, index + "/_count");
- if (field != null && value != null) {
- final String jsonEntity = Strings.toString(
- XContentFactory.jsonBuilder().startObject()
- .startObject("query")
- .startObject("match")
- .field(field, value)
- .endObject()
- .endObject()
- .endObject()
- );
- request.setJsonEntity(jsonEntity);
- }
- final Response response = client.performRequest(request);
- final String responseBody = EntityUtils.toString(response.getEntity());
- return (Integer) createContentParser(XContentType.JSON.xContent(), responseBody).map().get("count");
- }
+ private Integer getDocumentCount(final String index, final String field, final String value) throws IOException, InterruptedException {
+ final Request request = new Request(HttpMethod.GET, index + "/_count");
+ if (field != null && value != null) {
+ final String jsonEntity = Strings.toString(
+ XContentFactory.jsonBuilder().startObject()
+ .startObject("query")
+ .startObject("match")
+ .field(field, value)
+ .endObject()
+ .endObject()
+ .endObject()
+ );
+ request.setJsonEntity(jsonEntity);
+ }
+ final Response response = client.performRequest(request);
+ final String responseBody = EntityUtils.toString(response.getEntity());
+ return (Integer) createContentParser(XContentType.JSON.xContent(), responseBody).map().get("count");
+ }
- private List getSearchResponseDocIds(final String index) throws IOException {
- final Request refresh = new Request(HttpMethod.POST, index + "/_refresh");
- client.performRequest(refresh);
- final Request request = new Request(HttpMethod.GET, index + "/_search");
- final Response response = client.performRequest(request);
- final String responseBody = EntityUtils.toString(response.getEntity());
-
- @SuppressWarnings("unchecked") final List