Create OpenSearch source client with auth and lookup version to detect search strategy #2806

graytaylor0 · 2023-06-02T06:56:31Z

Description

This change mainly focused on implementing the OpenSearch client with auth and detecting the search strategy through a version lookup in SearchAccessorStrategy. The starting implementation here will only allow opensearch distributions for now.

Much of the code for the client construction and ssl was taken from the OpenSearch sink.

Tested with both a self managed cluster and Amazon OpenSearch domain successfully. I have not manually tested the cert and insecure: false

Issues Resolved

[List any issues this PR will resolve]

Check List

New functionality includes testing.
New functionality has been documented.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

…t search strategy Signed-off-by: Taylor Gray <[email protected]>

graytaylor0 · 2023-06-02T06:58:11Z

.../dataprepper/plugins/source/opensearch/worker/client/AwsRequestSigningApacheInterceptor.java

+ * An {@link HttpRequestInterceptor} that signs requests using any AWS {@link Signer}
+ * and {@link AwsCredentialsProvider}. This is a copy from the opensearch sink
+ */
+final class AwsRequestSigningApacheInterceptor implements HttpRequestInterceptor {


This entire class is a direct copy from the opensearch sink (https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/AwsRequestSigningApacheInterceptor.java). There don't appear to be any unit tests for it...

Can you create an issue to eliminate the code duplication between these two plugins and add unit tests

Will do. I started with trying to use the Extension Points for this class but it wasn't working and started taking up a lot of time. Will create the issue

graytaylor0 · 2023-06-02T07:01:54Z

.../org/opensearch/dataprepper/plugins/source/opensearch/worker/client/X509TrustAllManager.java

+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+
+public class X509TrustAllManager implements X509TrustManager {


This is a direct copy from the opensearch sink (https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/X509TrustAllManager.java)

graytaylor0 · 2023-06-02T07:02:50Z

...g/opensearch/dataprepper/plugins/source/opensearch/worker/client/SearchAccessorStrategy.java

+        return new OpenSearchAccessor(openSearchClient, searchContextType);
+    }
+
+    private RestClient createOpenSearchRestClient() {


The majority of these private methods were taken/based off of the opensearch sink (https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/ConnectionConfiguration.java)

cmanning09 · 2023-06-02T15:07:25Z

...ce/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/OpenSearchService.java

+        switch(searchAccessor.getSearchContextType()) {
+            case POINT_IN_TIME:
+                searchWorkerThread = new Thread(new PitWorker(searchAccessor, openSearchSourceConfiguration, sourceCoordinator, buffer));
+                break;
+            case SCROLL:
+                searchWorkerThread = new Thread(new ScrollWorker(searchAccessor, openSearchSourceConfiguration, sourceCoordinator, buffer));
+                break;
+        }


For a future CR, let's pull this logic into a strategy class that is deleted in this PR and do something like this here. It will be easier to test this control logic separate from thread creation:

final SearchWorker worker = SearchWorkerStrategy.get(searchAccessor, ...); searchWorkerThread = new Thread(worker); searchWorkerThread.start()

cmanning09 · 2023-06-02T15:26:10Z

...arch/dataprepper/plugins/source/opensearch/configuration/AwsAuthenticationConfiguration.java

-        }
-
-        return awsCredentialsProvider;
+    public Map<String, String> getAwsStsHeaderOverrides() {


Nice, thanks for pulling this work in too.

cmanning09 · 2023-06-02T15:27:46Z

...arch/dataprepper/plugins/source/opensearch/configuration/AwsAuthenticationConfiguration.java

-            throw new IllegalArgumentException("sts_role_arn must be an IAM Role");
-        }
-    }
-
    private Arn getArn() {


You can delete this too. I don't see it used anywhere now.

cmanning09 · 2023-06-02T15:28:28Z

...rce/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/PitWorker.java

-        //todo: to implement
+


Still a todo ;)

Haha it doesn't look done?

cmanning09 · 2023-06-02T15:29:18Z

.../src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/worker/SearchWorker.java

+
+package org.opensearch.dataprepper.plugins.source.opensearch.worker;
+
+public interface SearchWorker extends Runnable {


I like this change. Thanks for making this more extendable.

cmanning09 · 2023-06-02T15:31:10Z

.../dataprepper/plugins/source/opensearch/worker/client/AwsRequestSigningApacheInterceptor.java

+ * An {@link HttpRequestInterceptor} that signs requests using any AWS {@link Signer}
+ * and {@link AwsCredentialsProvider}. This is a copy from the opensearch sink
+ */
+final class AwsRequestSigningApacheInterceptor implements HttpRequestInterceptor {


Can you create an issue to eliminate the code duplication between these two plugins and add unit tests

cmanning09 · 2023-06-02T15:34:02Z

...g/opensearch/dataprepper/plugins/source/opensearch/worker/client/SearchAccessorStrategy.java

@@ -13,14 +59,242 @@
 */
 public class SearchAccessorStrategy {

+    private static final Logger LOG = LoggerFactory.getLogger(SearchAccessorStrategy.class);
+
+    private static final String AOS_SERVICE_NAME = "es";


For a future PR. We should support AOSS as well.

cmanning09 · 2023-06-02T15:37:56Z

...ce/src/main/java/org/opensearch/dataprepper/plugins/source/opensearch/OpenSearchService.java

    public void start() {
-        // todo: to implement
-        // Leverages a runnable (SearchWorker) to perform the querying on the source cluster
+        switch(searchAccessor.getSearchContextType()) {


NONE is a supported type. We should throw an exception if NONE is returned or we will encounter an NPE on line 52.

Will do until we end up adding a search worker for this case

cmanning09 · 2023-06-02T15:38:46Z

.../dataprepper/plugins/source/opensearch/configuration/AwsAuthenticationConfigurationTest.java

-    @Test
-    void getAwsRegion_returns_Region_of() throws NoSuchFieldException, IllegalAccessException {
-        final String regionString = UUID.randomUUID().toString();
-        final Region expectedRegionObject = mock(Region.class);
-        reflectivelySetField(awsAuthenticationOptions, "awsRegion", regionString);
-        final Region actualRegion;
-        try(final MockedStatic<Region> regionMockedStatic = mockStatic(Region.class)) {
-            regionMockedStatic.when(() -> Region.of(regionString)).thenReturn(expectedRegionObject);
-            actualRegion = awsAuthenticationOptions.getAwsRegion();
-        }
-        assertThat(actualRegion, equalTo(expectedRegionObject));
-    }
-
-    @Test
-    void getAwsRegion_returns_null_when_region_is_null() throws NoSuchFieldException, IllegalAccessException {
-        reflectivelySetField(awsAuthenticationOptions, "awsRegion", null);
-        assertThat(awsAuthenticationOptions.getAwsRegion(), nullValue());
-    }


We should keep these tests around and add tests for the getter methods in future CRs.

Yeah will add these basic getters back

Signed-off-by: Taylor Gray <[email protected]>

kkondaka · 2023-06-02T22:23:55Z

...est/java/org/opensearch/dataprepper/expression/GenericExpressionEvaluator_ConditionalIT.java

@@ -94,7 +94,7 @@ void testConditionalExpressionEvaluator(final String expression, final Event eve
    void testGenericExpressionEvaluatorWithMultipleThreads(final String expression, final Event event, final Boolean expected) {
        final GenericExpressionEvaluator evaluator = applicationContext.getBean(GenericExpressionEvaluator.class);

-        final int numberOfThreads = 50;
+        final int numberOfThreads = 10;


Why is this file being changed here? Why are the number of threads reduced?

That test runs over 50 times, and it was spawning 50 threads for each test. My build failed every time due to that with an error about being out of threads. It didn’t fail after I lowered it. Is this test really necessary for every conditional expression?

Create OpenSearch source client with auth and lookup version to detec…

91fabf4

…t search strategy Signed-off-by: Taylor Gray <[email protected]>

graytaylor0 requested review from chenqi0805, engechas, dinujoh, kkondaka, cmanning09, asifsmohammed, dlvenable and oeyh as code owners June 2, 2023 06:56

graytaylor0 commented Jun 2, 2023

View reviewed changes

cmanning09 reviewed Jun 2, 2023

View reviewed changes

cmanning09 mentioned this pull request Jun 2, 2023

OpenSearch source plugin #1985

Closed

Address PR comments

b5ff864

Signed-off-by: Taylor Gray <[email protected]>

graytaylor0 requested a review from cmanning09 June 2, 2023 17:15

cmanning09 approved these changes Jun 2, 2023

View reviewed changes

kkondaka reviewed Jun 2, 2023

View reviewed changes

kkondaka approved these changes Jun 2, 2023

View reviewed changes

graytaylor0 merged commit c2d7767 into opensearch-project:main Jun 3, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Create OpenSearch source client with auth and lookup version to detect search strategy #2806

Create OpenSearch source client with auth and lookup version to detect search strategy #2806

graytaylor0 commented Jun 2, 2023 •

edited

Loading

graytaylor0 Jun 2, 2023

cmanning09 Jun 2, 2023

graytaylor0 Jun 2, 2023 •

edited

Loading

graytaylor0 Jun 2, 2023

graytaylor0 Jun 2, 2023

graytaylor0 Jun 2, 2023

cmanning09 Jun 2, 2023

cmanning09 Jun 2, 2023

cmanning09 Jun 2, 2023

cmanning09 Jun 2, 2023

graytaylor0 Jun 2, 2023

cmanning09 Jun 2, 2023

cmanning09 Jun 2, 2023

cmanning09 Jun 2, 2023

cmanning09 Jun 2, 2023

graytaylor0 Jun 2, 2023

cmanning09 Jun 2, 2023

graytaylor0 Jun 2, 2023

kkondaka Jun 2, 2023

graytaylor0 Jun 3, 2023 •

edited

Loading


		package org.opensearch.dataprepper.plugins.source.opensearch.worker;

		public interface SearchWorker extends Runnable {

Create OpenSearch source client with auth and lookup version to detect search strategy #2806

Create OpenSearch source client with auth and lookup version to detect search strategy #2806

Conversation

graytaylor0 commented Jun 2, 2023 • edited Loading

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 Jun 2, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 Jun 3, 2023 • edited Loading

Choose a reason for hiding this comment

graytaylor0 commented Jun 2, 2023 •

edited

Loading

graytaylor0 Jun 2, 2023 •

edited

Loading

graytaylor0 Jun 3, 2023 •

edited

Loading