New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

New OpenSearch API source implementation #4603

Closed

sb2k16 wants to merge 0 commits into opensearch-project:main from sb2k16:opensearch-api-async

Member

sb2k16 commented Jun 5, 2024 •

edited

Loading

Description

In order for DataPrepper to support all OpenSearch Document API(s), we need to build a new source similar to the existing http source. This pull request is intended to implement a new OpenSearch API source like opensearch_api similar to http source. This source should support the Document API Bulk.

This pull request includes the following:

Path and HTTP methods:
- POST _bulk
- POST <index>/_bulk
Optional URL parameters
- pipeline and routing. (TODO: pipeline parameter handling on Sink side)
Multi-line JSON Bulk Request payload

Example pipeline configuration with opensearch_api source looks like:

simple-sample-pipeline:
  source:
    opensearch_api:
      path: "/opensearch"
      port: 9202
  sink:
   ...

Issues Resolved

Contributes to #248

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
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.

sb2k16 requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, asifsmohammed, KarstenSchnitter, dlvenable and oeyh as code owners

June 5, 2024 16:39

dlvenable requested changes

View reviewed changes

...tp-source-common/src/main/java/org/opensearch/dataprepper/http/codec/MultiLineJsonCodec.java Outdated

+                      List<Map<String, Object>> jsonListData = new ArrayList<>();
+                      String requestBody = new String(httpData.toInputStream().readAllBytes(), StandardCharsets.UTF_8);
+                      List<String> jsonLines = Arrays.asList(requestBody.split(REGEX));

Member

dlvenable Jun 6, 2024

It may be better to create the Pattern in the constructor. Then you can call splitPattern.split(requestBody)). This should avoid compilation each time.

Member Author

sb2k16 Jun 12, 2024

Thanks @dlvenable . I am going to address this.

data-prepper-plugins/opensearch-api-source/build.gradle Outdated

+                  violationRules {
+                      rule { //in addition to core projects rule
+                          limit {
+                              minimum = 0.90

Member

dlvenable Jun 6, 2024

This is a new project. Can we aim for 100% coverage?

Member Author

sb2k16 Jun 12, 2024

Thanks @dlvenable . I am going to address this.

...c/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISource.java Outdated

+                          Thread.currentThread().interrupt();
+                          throw new RuntimeException(ex);
+                      }
+                      LOG.info("Started OpenSearch API source on port " + sourceConfig.getPort() + "...");

Member

dlvenable Jun 6, 2024

Suggested change

      
                    LOG.info("Started OpenSearch API source on port " + sourceConfig.getPort() + "...");
          
                    LOG.info("Started OpenSearch API source on port {}.", sourceConfig.getPort());

Let's avoid these ellipses at the end. They are unclear and make it seem that more is coming.

Also, please use SLF4J interpolation over string concatenation.

Member Author

sb2k16 Jun 12, 2024

Thanks @dlvenable . I am going to address this.

.../java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceConfig.java Outdated


		public class OpenSearchAPISourceConfig extends BaseHttpServerConfig {

		static final String DEFAULT_ENDPOINT_URI = "/opensearch";

Member

dlvenable Jun 6, 2024

I think the default should just be / since this matches existing OpenSearch domains.

Member Author

sb2k16 Jun 12, 2024

Thanks @dlvenable . I am going to address this.

.../main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java Outdated

+              import java.util.Arrays;
+              import java.util.ArrayList;
+              /*

Member

dlvenable Jun 6, 2024

Please run a code formatter over your changes. There is a lot of whitespace that is off.

Member Author

sb2k16 Jun 12, 2024

Thanks @dlvenable . I am going to address this.

...c/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISource.java Outdated

+                      if (buffer == null) {
+                          throw new IllegalStateException("Buffer provided is null");
+                      }
+                      if (server == null) {

Member

dlvenable Jun 6, 2024

There is a lot here that is similar to the http source. Can we extend your work from #4570 to have more of this shared in common?

Member Author

sb2k16 Jun 12, 2024

Thanks @dlvenable . I am going to address this.

.../main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java Outdated

+                  @Post("/_bulk")
+                  public HttpResponse doPostBulk(final ServiceRequestContext serviceRequestContext, final AggregatedHttpRequest aggregatedHttpRequest,
+                                                 @Param("pipeline") Optional<String> pipeline, @Param("routing") Optional<String> routing) throws Exception {

Member

dlvenable Jun 6, 2024

Passing Optional in as a parameter is an anti-pattern. You don't actually know if it is null or not. So you have tw checks you need to make pipeline != null && pipeline.isPresent(). Just take in a String and expect it to possibly be null.

Member Author

sb2k16 Jun 12, 2024

Thanks @dlvenable . I am going to address this.

.../main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java Outdated

+                  @Post("/{index}/_bulk")
+                  public HttpResponse doPostBulkIndex(final ServiceRequestContext serviceRequestContext, final AggregatedHttpRequest aggregatedHttpRequest, @Param("index") Optional<String> index,
+                                                  @Param("pipeline") Optional<String> pipeline, @Param("routing") Optional<String> routing) throws Exception {
+                      requestsReceivedCounter.increment();

Member

dlvenable Jun 6, 2024

Let's consolidate all of this logic in processBulkRequest. The only thing that each of these methods should do is create the BulkAPIRequestParams and then callprocessBulkRequest.

Member Author

sb2k16 Jun 12, 2024

Thanks @dlvenable . I am going to address this.

sb2k16 force-pushed the opensearch-api-async branch from f052e71 to 1ad4250 Compare

June 12, 2024 14:05

dlvenable previously approved these changes

View reviewed changes

Member

dlvenable left a comment

Thanks for the refactoring work!

.../main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java Outdated


		private static final Logger LOG = LoggerFactory.getLogger(OpenSearchAPIService.class);

		// TODO: support other data-types as request body, e.g. json_lines, msgpack

Member

dlvenable Jun 14, 2024

Is this TODO really true? I think this is the only codec we need for _bulk.

sb2k16 dismissed dlvenable’s stale review via

96025bc

July 22, 2024 22:00

sb2k16 force-pushed the opensearch-api-async branch 2 times, most recently from fe8659f to 77930fc Compare

July 25, 2024 16:53

sb2k16 force-pushed the opensearch-api-async branch from 918dfd9 to 0e7bf17 Compare

September 22, 2024 01:15

dlvenable requested changes

View reviewed changes

.../java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceConfig.java Outdated Show resolved Hide resolved

...st/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java Outdated

+                                      .scheme(SessionProtocol.HTTP)
+                                      .authority(AUTHORITY)
+                                      .method(HttpMethod.GET)
+                                      .path("/health")

Member

dlvenable Oct 3, 2024

Do we want a health check endpoint? This is not something that the OpenSearch API would normally have.

If anything, maybe we should support the root API?

GET /

We could add this in a follow-on PR. But, I think we want to remove the /health endpoint.

Do you want to create a follow-on PR for the root endpoint?

Member Author

sb2k16 Oct 4, 2024

Thanks @dlvenable. I will resolve this. The default health check endpoint is /health which is the default endpoint for Http source. I will modify the BastHttpSource implementation to ensure that the OpenSearch API source can control the endpoint / for health check.

...st/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java Outdated

+                  private static final ObjectMapper mapper = new ObjectMapper();
+                  private final String PLUGIN_NAME = "opensearch_api";
+                  private final String TEST_PIPELINE_NAME = "test_pipeline";
+                  private final String TEST_INDEX = "test-index";

Member

dlvenable Oct 3, 2024

We should have this be a variable generated in @BeforeEach.

private String testIndex;

...

@BeforeEach
void setUp() {
  testIndex = UUID.randomUUID().toString();
}

Member Author

sb2k16 Oct 4, 2024

Thanks @dlvenable. I will resolve this.

...st/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java Outdated

+              class OpenSearchAPISourceTest {
+                  private static final ObjectMapper mapper = new ObjectMapper();
+                  private final String PLUGIN_NAME = "opensearch_api";
+                  private final String TEST_PIPELINE_NAME = "test_pipeline";

Member

dlvenable Oct 3, 2024

Let's also make this a random value generated in the setup method.

Member Author

sb2k16 Oct 4, 2024

Thanks @dlvenable. I will resolve this.

...st/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java Outdated

+                      integerHashMap.put("batch_size", 1);
+                      final PluginSetting pluginSetting = new PluginSetting("blocking_buffer", integerHashMap);
+                      pluginSetting.setPipelineName(TEST_PIPELINE_NAME);
+                      return new BlockingBuffer<>(pluginSetting);

Member

dlvenable Oct 3, 2024

Can we use a mock buffer instead?

...st/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java Outdated

+                  @ParameterizedTest
+                  @ValueSource(booleans = {false, true})
+                  public void testBulkRequestAPIJsonResponse413(boolean includeIndexInPath) throws JsonProcessingException {
+                      testBulkRequestJsonResponse413(includeIndexInPath);

Member

dlvenable Oct 3, 2024

Let's inline this. We don't need an extra method here.

Member Author

sb2k16 Oct 4, 2024

Thanks @dlvenable. I will resolve this.

...st/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java Outdated

+                      // Fill in the buffer
+                      WebClient.of().execute(testRequestHeaders, testHttpData).aggregate()
+                              .whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.OK)).join();

Member

dlvenable Oct 3, 2024

Why is this expecting an OK? Shouldn't it be the 408?

...ource-common/src/test/java/org/opensearch/dataprepper/http/codec/MultiLineJsonCodecTest.java Outdated

+                  private final HttpData serializedRequestBadEmptyNewLines = HttpData.ofUtf8("\n\n\n\n\n\n\n     \n");
+                  private final HttpData serializedRequestBadInvalidJson = HttpData.ofUtf8("{\"text\":");
+                  private final MultiLineJsonCodec multiLineJsonCodec = new MultiLineJsonCodec();

Member

dlvenable Oct 3, 2024

Let's use the createObjectUnderTest() method pattern. It allows more flexibility as tests grow.

Member Author

sb2k16 Oct 4, 2024

Thanks @dlvenable. I will resolve this.

.../main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java Outdated

+                      try {
+                          if (buffer.isByteBuffer()) {
+                              buffer.writeBytes(content.array(), null, bufferWriteTimeoutInMillis);

Member

dlvenable Oct 3, 2024

We will probably want to support chunking. But, we can do this in a follow-on PR. Can you create a PR?

Member Author

sb2k16 Oct 4, 2024

Thanks @dlvenable. I will create a separate PR to follow up on adding the chunking functionality.

.../main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java Outdated

+               * Bulk API supports query parameters "pipeline", "routing" and "refresh"
+               */
+              @Blocking
+              public class OpenSearchAPIService implements BaseHttpService {

Member

dlvenable Oct 3, 2024

We will need to support end-to-end acknowledgements, but we can do this in a follow-on PR. Can you create a GitHub issue to track this?

Member Author

sb2k16 Oct 4, 2024

sb2k16 closed this

sb2k16 force-pushed the opensearch-api-async branch from e1fe2ba to 472c912 Compare

October 5, 2024 18:38

Member

dlvenable commented Oct 7, 2024

Replaced by #5024.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

dlvenable dlvenable requested changes

chenqi0805 Awaiting requested review from chenqi0805

engechas Awaiting requested review from engechas

graytaylor0 Awaiting requested review from graytaylor0

dinujoh Awaiting requested review from dinujoh

kkondaka Awaiting requested review from kkondaka

asifsmohammed Awaiting requested review from asifsmohammed

KarstenSchnitter Awaiting requested review from KarstenSchnitter

oeyh Awaiting requested review from oeyh

Labels

None yet