diff --git a/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java b/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java index 90cb77a21f..5b25a20c39 100644 --- a/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java +++ b/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIService.java @@ -17,28 +17,19 @@ import io.micrometer.core.instrument.Counter; import io.micrometer.core.instrument.DistributionSummary; import io.micrometer.core.instrument.Timer; -import org.apache.commons.lang3.StringUtils; import org.opensearch.dataprepper.http.BaseHttpService; -import org.opensearch.dataprepper.http.codec.MultiLineJsonCodec; import org.opensearch.dataprepper.metrics.PluginMetrics; import org.opensearch.dataprepper.model.buffer.Buffer; import org.opensearch.dataprepper.model.event.Event; -import org.opensearch.dataprepper.model.event.EventType; -import org.opensearch.dataprepper.model.event.JacksonEvent; -import org.opensearch.dataprepper.model.opensearch.OpenSearchBulkActions; import org.opensearch.dataprepper.model.record.Record; -import org.opensearch.dataprepper.plugins.source.opensearchapi.model.BulkAPIEventMetadataKeyAttributes; import org.opensearch.dataprepper.plugins.source.opensearchapi.model.BulkAPIRequestParams; -import org.opensearch.dataprepper.plugins.source.opensearchapi.model.BulkActionAndMetadataObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; +import java.io.ByteArrayInputStream; +import java.time.Instant; import java.util.ArrayList; -import java.util.Arrays; -import java.util.Iterator; import java.util.List; -import java.util.Map; /* * OpenSearch API Service class is responsible for handling bulk API requests. @@ -56,8 +47,7 @@ public class OpenSearchAPIService implements BaseHttpService { private static final Logger LOG = LoggerFactory.getLogger(OpenSearchAPIService.class); - // TODO: support other data-types as request body, e.g. json_lines, msgpack - private final MultiLineJsonCodec jsonCodec = new MultiLineJsonCodec(); + private final OpenSearchBulkByteDecoder bulkByteDecoder = new OpenSearchBulkByteDecoder(); private final Buffer> buffer; private final int bufferWriteTimeoutInMillis; private final Counter requestsReceivedCounter; @@ -110,21 +100,13 @@ private HttpResponse processBulkRequest(final ServiceRequestContext serviceReque } final HttpData content = aggregatedHttpRequest.content(); - List> bulkRequestPayloadList; - - // parse the request payload - try { - bulkRequestPayloadList = jsonCodec.parse(content); - } catch (IOException e) { - LOG.error("Failed to parse the request of size {} due to: {}", content.length(), e.getMessage()); - throw new IOException("Bad request data format.", e.getCause()); - } try { if (buffer.isByteBuffer()) { buffer.writeBytes(content.array(), null, bufferWriteTimeoutInMillis); } else { - List> records = generateEventsFromBulkRequest(bulkRequestPayloadList, bulkAPIRequestParams); + List> records = new ArrayList<>(); + bulkByteDecoder.parse(new ByteArrayInputStream(content.array()), Instant.now(), bulkAPIRequestParams, records::add); buffer.writeAll(records, bufferWriteTimeoutInMillis); } } catch (Exception e) { @@ -134,72 +116,4 @@ private HttpResponse processBulkRequest(final ServiceRequestContext serviceReque successRequestsCounter.increment(); return HttpResponse.of(HttpStatus.OK); } - - private boolean isValidBulkAction(Map actionMap) { - return Arrays.stream(OpenSearchBulkActions.values()) - .anyMatch(bulkAction -> actionMap.containsKey(bulkAction.toString())); - } - - private List> generateEventsFromBulkRequest(final List> bulkRequestPayloadList, final BulkAPIRequestParams bulkAPIRequestParams) throws Exception { - List> records = new ArrayList<>(); - Iterator> bulkRequestPayloadListIterator = bulkRequestPayloadList.iterator(); - - while (bulkRequestPayloadListIterator.hasNext()) { - Map actionMetadataRow = bulkRequestPayloadListIterator.next(); - if (!isValidBulkAction(actionMetadataRow)) { - throw new IOException("Invalid request data."); - } - - BulkActionAndMetadataObject bulkActionAndMetadataObject = new BulkActionAndMetadataObject(actionMetadataRow); - final boolean isDeleteAction = bulkActionAndMetadataObject.getAction().equals(OpenSearchBulkActions.DELETE.toString()); - Map documentDataObject = null; - if (!isDeleteAction) { - if (!bulkRequestPayloadListIterator.hasNext()) { - throw new IOException("Invalid request data."); - } - documentDataObject = bulkRequestPayloadListIterator.next(); - // Performing another validation check to make sure that the doc row is not a valid action row - if (isValidBulkAction(documentDataObject)) { - throw new IOException("Invalid request data."); - } - } - final JacksonEvent event = createBulkRequestActionEvent(bulkActionAndMetadataObject, bulkAPIRequestParams, documentDataObject); - records.add(new Record<>(event)); - } - - return records; - } - - private JacksonEvent createBulkRequestActionEvent( - final BulkActionAndMetadataObject bulkActionAndMetadataObject, - final BulkAPIRequestParams bulkAPIRequestParams, Map optionalDocumentData) { - - final JacksonEvent.Builder eventBuilder = JacksonEvent.builder().withEventType(EventType.DOCUMENT.toString()); - if (optionalDocumentData != null) { - eventBuilder.withData(optionalDocumentData); - } - final JacksonEvent event = eventBuilder.build(); - - final String index = !StringUtils.isEmpty(bulkAPIRequestParams.getIndex()) ? bulkAPIRequestParams.getIndex() : bulkActionAndMetadataObject.getIndex(); - - event.getMetadata().setAttribute(BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION, bulkActionAndMetadataObject.getAction()); - event.getMetadata().setAttribute(BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_INDEX, index); - - String docId = bulkActionAndMetadataObject.getDocId(); - if (!StringUtils.isBlank(docId) && !StringUtils.isEmpty(docId)) { - event.getMetadata().setAttribute(BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ID, docId); - } - - String pipeline = bulkAPIRequestParams.getPipeline(); - if (!StringUtils.isBlank(pipeline) && !StringUtils.isEmpty(pipeline)) { - event.getMetadata().setAttribute(BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_PIPELINE, pipeline); - } - - String routing = bulkAPIRequestParams.getRouting(); - if (!StringUtils.isBlank(routing) && !StringUtils.isEmpty(routing)) { - event.getMetadata().setAttribute(BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ROUTING, routing); - } - - return event; - } } diff --git a/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISource.java b/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISource.java index 9bfa1a9c53..4f95c8e108 100644 --- a/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISource.java +++ b/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISource.java @@ -11,6 +11,7 @@ import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin; import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor; import org.opensearch.dataprepper.model.buffer.Buffer; +import org.opensearch.dataprepper.model.codec.ByteDecoder; import org.opensearch.dataprepper.model.configuration.PipelineDescription; import org.opensearch.dataprepper.model.event.Event; import org.opensearch.dataprepper.model.plugin.PluginFactory; @@ -38,4 +39,9 @@ public BaseHttpService getHttpService(final int bufferWriteTimeoutInMillis, fina public String getHttpHealthCheckPath() { return HTTP_HEALTH_CHECK_PATH; } + + @Override + public ByteDecoder getDecoder() { + return new OpenSearchBulkByteDecoder(); + } } diff --git a/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchBulkByteDecoder.java b/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchBulkByteDecoder.java new file mode 100644 index 0000000000..672e6ef5f0 --- /dev/null +++ b/data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchBulkByteDecoder.java @@ -0,0 +1,137 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + * + */ + +package org.opensearch.dataprepper.plugins.source.opensearchapi; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.opensearch.dataprepper.model.codec.ByteDecoder; +import org.opensearch.dataprepper.model.event.Event; +import org.opensearch.dataprepper.model.event.EventType; +import org.opensearch.dataprepper.model.event.JacksonEvent; +import org.opensearch.dataprepper.model.opensearch.OpenSearchBulkActions; +import org.opensearch.dataprepper.model.record.Record; +import org.opensearch.dataprepper.plugins.source.opensearchapi.model.BulkAPIEventMetadataKeyAttributes; +import org.opensearch.dataprepper.plugins.source.opensearchapi.model.BulkAPIRequestParams; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.Arrays; +import java.util.Map; +import java.util.function.Consumer; + +public class OpenSearchBulkByteDecoder implements ByteDecoder { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final TypeReference> MAP_TYPE = new TypeReference<>() {}; + + @Override + public void parse(InputStream inputStream, Instant timeReceived, Consumer> eventConsumer) throws IOException { + parse(inputStream, timeReceived, null, eventConsumer); + } + + public void parse(InputStream inputStream, Instant timeReceived, BulkAPIRequestParams requestParams, Consumer> eventConsumer) throws IOException { + final BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); + String line; + while ((line = reader.readLine()) != null) { + if (line.isBlank()) continue; + + Map actionLine = OBJECT_MAPPER.readValue(line, MAP_TYPE); + String action = extractAction(actionLine); + if (action == null) continue; + + @SuppressWarnings("unchecked") + Map metadata = (Map) actionLine.get(action); + boolean isDelete = OpenSearchBulkActions.DELETE.toString().equals(action); + + Map documentData = null; + if (!isDelete) { + String docLine = reader.readLine(); + if (docLine == null || docLine.isBlank()) continue; + documentData = OBJECT_MAPPER.readValue(docLine, MAP_TYPE); + } + + JacksonEvent.Builder builder = JacksonEvent.builder() + .withEventType(EventType.DOCUMENT.toString()); + if (documentData != null) { + builder.withData(documentData); + } + if (timeReceived != null) { + builder.withTimeReceived(timeReceived); + } + JacksonEvent event = builder.build(); + + event.getMetadata().setAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION, action); + + // Set index: action line takes priority, then URL-level fallback + String index = getMetadataValue(metadata, "_index"); + if (isBlank(index) && requestParams != null && !isBlank(requestParams.getIndex())) { + index = requestParams.getIndex(); + } + if (!isBlank(index)) { + event.getMetadata().setAttribute(BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_INDEX, index); + } + + // Set id + setIfPresent(event, metadata, "_id", BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ID); + + // Set routing: action line takes priority, then URL-level fallback + String routing = getMetadataValue(metadata, "routing"); + if (isBlank(routing) && requestParams != null && !isBlank(requestParams.getRouting())) { + routing = requestParams.getRouting(); + } + if (!isBlank(routing)) { + event.getMetadata().setAttribute(BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ROUTING, routing); + } + + // Set pipeline: action line takes priority, then URL-level fallback + String pipeline = getMetadataValue(metadata, "pipeline"); + if (isBlank(pipeline) && requestParams != null && !isBlank(requestParams.getPipeline())) { + pipeline = requestParams.getPipeline(); + } + if (!isBlank(pipeline)) { + event.getMetadata().setAttribute(BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_PIPELINE, pipeline); + } + + eventConsumer.accept(new Record<>(event)); + } + } + + private String extractAction(Map actionLine) { + return Arrays.stream(OpenSearchBulkActions.values()) + .map(OpenSearchBulkActions::toString) + .filter(actionLine::containsKey) + .findFirst() + .orElse(null); + } + + private String getMetadataValue(Map metadata, String key) { + if (metadata == null) return null; + Object value = metadata.get(key); + return value != null ? value.toString() : null; + } + + private void setIfPresent(JacksonEvent event, Map metadata, String key, String attribute) { + if (metadata == null) return; + Object value = metadata.get(key); + if (value != null) { + event.getMetadata().setAttribute(attribute, value.toString()); + } + } + + private boolean isBlank(String value) { + return value == null || value.isBlank(); + } +} diff --git a/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIServiceTest.java b/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIServiceTest.java index 28a567c4f4..c629e4ed0b 100644 --- a/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIServiceTest.java +++ b/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPIServiceTest.java @@ -85,10 +85,10 @@ class OpenSearchAPIServiceTest { @BeforeEach public void setUp() throws Exception { - lenient().when(pluginMetrics.counter(openSearchAPIService.REQUESTS_RECEIVED)).thenReturn(requestsReceivedCounter); - lenient().when(pluginMetrics.counter(openSearchAPIService.SUCCESS_REQUESTS)).thenReturn(successRequestsCounter); - lenient().when(pluginMetrics.summary(openSearchAPIService.PAYLOAD_SIZE)).thenReturn(payloadSizeSummary); - lenient().when(pluginMetrics.timer(openSearchAPIService.REQUEST_PROCESS_DURATION)).thenReturn(requestProcessDuration); + lenient().when(pluginMetrics.counter(OpenSearchAPIService.REQUESTS_RECEIVED)).thenReturn(requestsReceivedCounter); + lenient().when(pluginMetrics.counter(OpenSearchAPIService.SUCCESS_REQUESTS)).thenReturn(successRequestsCounter); + lenient().when(pluginMetrics.summary(OpenSearchAPIService.PAYLOAD_SIZE)).thenReturn(payloadSizeSummary); + lenient().when(pluginMetrics.timer(OpenSearchAPIService.REQUEST_PROCESS_DURATION)).thenReturn(requestProcessDuration); lenient().when(serviceRequestContext.isTimedOut()).thenReturn(false); lenient().when(requestProcessDuration.recordCallable(ArgumentMatchers.>any())).thenAnswer( (Answer) invocation -> { @@ -219,8 +219,10 @@ public void testBulkRequestAPIEmptyIndex() throws Exception { null, null).aggregate().get(); } + // Decoder is lenient: treats the second action line as document data for the first action, + // matching OpenSearch's own tolerant bulk parsing behavior. @Test - public void testBulkRequestAPIInvalidRequestMissingDocRow() throws Exception { + public void testBulkRequestAPIWithMissingDocRowSucceeds() throws Exception { RequestHeaders requestHeaders = RequestHeaders.builder() .contentType(MediaType.JSON) .method(HttpMethod.POST) @@ -234,12 +236,14 @@ public void testBulkRequestAPIInvalidRequestMissingDocRow() throws Exception { Buffer> blockingBuffer = mock(BlockingBuffer.class); OpenSearchAPIService openSearchAPIService = new OpenSearchAPIService(TEST_TIMEOUT_IN_MILLIS, blockingBuffer, pluginMetrics); - assertThrows(IOException.class, () -> openSearchAPIService.doPostBulk(serviceRequestContext, testRequest, - null, null).aggregate().get()); + AggregatedHttpResponse response = openSearchAPIService.doPostBulk(serviceRequestContext, testRequest, + null, null).aggregate().get(); + assertEquals(HttpStatus.OK, response.status()); } + // Decoder skips an action line that has no following document line (EOF reached), producing zero events. @Test - public void testBulkRequestAPIInvalidRequestEmptyDocRow() throws Exception { + public void testBulkRequestAPIWithSingleActionLineSucceeds() throws Exception { RequestHeaders requestHeaders = RequestHeaders.builder() .contentType(MediaType.JSON) .method(HttpMethod.POST) @@ -252,55 +256,118 @@ public void testBulkRequestAPIInvalidRequestEmptyDocRow() throws Exception { Buffer> blockingBuffer = mock(BlockingBuffer.class); OpenSearchAPIService openSearchAPIService = new OpenSearchAPIService(TEST_TIMEOUT_IN_MILLIS, blockingBuffer, pluginMetrics); - assertThrows(IOException.class, () -> openSearchAPIService.doPostBulk(serviceRequestContext, testRequest, - null, null).aggregate().get()); + AggregatedHttpResponse response = openSearchAPIService.doPostBulk(serviceRequestContext, testRequest, + null, null).aggregate().get(); + assertEquals(HttpStatus.OK, response.status()); } + // Empty payload produces zero events and returns 200. @ParameterizedTest @ValueSource(booleans = {false, true}) - public void testBulkRequestAPIBadRequestWithEmpty(boolean testBulkRequestAPIWithIndexInPath) throws Exception { - testBadRequestWithPayload(testBulkRequestAPIWithIndexInPath, ""); - } + public void testBulkRequestAPIWithEmptyPayloadSucceeds(boolean testBulkRequestAPIWithIndexInPath) throws Exception { + RequestHeaders requestHeaders = RequestHeaders.builder() + .contentType(MediaType.JSON) + .method(HttpMethod.POST) + .path("/") + .build(); - @ParameterizedTest - @ValueSource(booleans = {false, true}) - public void testBulkRequestAPIBadRequestWithInvalidPayload(boolean testBulkRequestAPIWithIndexInPath) throws Exception { - List jsonList = new ArrayList<>(); - for (int i = 0; i < 2; i++) { - jsonList.add(mapper.writeValueAsString(Collections.singletonMap("index", Collections.singletonMap("_index", "test-index")))); + HttpData httpData = HttpData.ofUtf8(""); + AggregatedHttpRequest testRequest = HttpRequest.of(requestHeaders, httpData).aggregate().get(); + + AggregatedHttpResponse response; + if (testBulkRequestAPIWithIndexInPath) { + response = openSearchAPIService.doPostBulkIndex(serviceRequestContext, testRequest, null, + null, null).aggregate().get(); + } else { + response = openSearchAPIService.doPostBulk(serviceRequestContext, testRequest, + null, null).aggregate().get(); } - testBadRequestWithPayload(testBulkRequestAPIWithIndexInPath, String.join("\n", jsonList)); + assertEquals(HttpStatus.OK, response.status()); + verify(requestsReceivedCounter, times(1)).increment(); + verify(successRequestsCounter, times(1)).increment(); } + // Decoder skips JSON lines with no valid bulk action key, producing zero events. @ParameterizedTest @ValueSource(booleans = {false, true}) - public void testBulkRequestAPIBadRequestWithEmptyMap(boolean testBulkRequestAPIWithIndexInPath) throws Exception { + public void testBulkRequestAPIWithEmptyMapPayloadSucceeds(boolean testBulkRequestAPIWithIndexInPath) throws Exception { + RequestHeaders requestHeaders = RequestHeaders.builder() + .contentType(MediaType.JSON) + .method(HttpMethod.POST) + .path("/") + .build(); List jsonList = new ArrayList<>(); for (int i = 0; i < 2; i++) { jsonList.add(mapper.writeValueAsString(Collections.emptyMap())); } - testBadRequestWithPayload(testBulkRequestAPIWithIndexInPath, String.join("\n", jsonList)); + HttpData httpData = HttpData.ofUtf8(String.join("\n", jsonList)); + AggregatedHttpRequest testRequest = HttpRequest.of(requestHeaders, httpData).aggregate().get(); + + AggregatedHttpResponse response; + if (testBulkRequestAPIWithIndexInPath) { + response = openSearchAPIService.doPostBulkIndex(serviceRequestContext, testRequest, null, + null, null).aggregate().get(); + } else { + response = openSearchAPIService.doPostBulk(serviceRequestContext, testRequest, + null, null).aggregate().get(); + } + assertEquals(HttpStatus.OK, response.status()); + verify(requestsReceivedCounter, times(1)).increment(); + verify(successRequestsCounter, times(1)).increment(); } + // Decoder skips JSON lines with no valid bulk action key, producing zero events. @ParameterizedTest @ValueSource(booleans = {false, true}) - public void testBulkRequestAPIBadRequestWithInvalidPayload2(boolean testBulkRequestAPIWithIndexInPath) throws Exception { + public void testBulkRequestAPIWithNonActionPayloadSucceeds(boolean testBulkRequestAPIWithIndexInPath) throws Exception { + RequestHeaders requestHeaders = RequestHeaders.builder() + .contentType(MediaType.JSON) + .method(HttpMethod.POST) + .path("/") + .build(); List jsonList = new ArrayList<>(); for (int i = 0; i < 2; i++) { jsonList.add(mapper.writeValueAsString(Collections.singletonMap("text", Collections.singletonMap("x", "test")))); } - testBadRequestWithPayload(testBulkRequestAPIWithIndexInPath, String.join("\n", jsonList)); + HttpData httpData = HttpData.ofUtf8(String.join("\n", jsonList)); + AggregatedHttpRequest testRequest = HttpRequest.of(requestHeaders, httpData).aggregate().get(); + + AggregatedHttpResponse response; + if (testBulkRequestAPIWithIndexInPath) { + response = openSearchAPIService.doPostBulkIndex(serviceRequestContext, testRequest, null, + null, null).aggregate().get(); + } else { + response = openSearchAPIService.doPostBulk(serviceRequestContext, testRequest, + null, null).aggregate().get(); + } + // Decoder skips lines with no valid bulk action + assertEquals(HttpStatus.OK, response.status()); + verify(requestsReceivedCounter, times(1)).increment(); + verify(successRequestsCounter, times(1)).increment(); } @ParameterizedTest @ValueSource(booleans = {false, true}) - public void testBulkRequestAPIBadRequestWithInvalidPayload3(boolean testBulkRequestAPIWithIndexInPath) throws Exception { - List jsonList = new ArrayList<>(); - jsonList.add(mapper.writeValueAsString(Collections.singletonMap("index", Collections.singletonMap("_index", "test-index")))); - jsonList.add(mapper.writeValueAsString(Collections.singletonMap("log", UUID.randomUUID().toString()))); - jsonList.add(mapper.writeValueAsString(Collections.singletonMap("index", Collections.singletonMap("_index", "test-index")))); + public void testBulkRequestAPIBadRequestWithInvalidJson(boolean testBulkRequestAPIWithIndexInPath) throws Exception { + RequestHeaders requestHeaders = RequestHeaders.builder() + .contentType(MediaType.JSON) + .method(HttpMethod.POST) + .path("/") + .build(); + + HttpData httpData = HttpData.ofUtf8("this is not valid json\n"); + AggregatedHttpRequest testBadRequest = HttpRequest.of(requestHeaders, httpData).aggregate().get(); - testBadRequestWithPayload(testBulkRequestAPIWithIndexInPath, String.join("\n", jsonList)); + if (testBulkRequestAPIWithIndexInPath) { + assertThrows(IOException.class, () -> openSearchAPIService.doPostBulkIndex(serviceRequestContext, testBadRequest, null, + null, null).aggregate().get()); + } else { + assertThrows(IOException.class, () -> openSearchAPIService.doPostBulk(serviceRequestContext, testBadRequest, null, + null).aggregate().get()); + } + + verify(requestsReceivedCounter, times(1)).increment(); + verify(successRequestsCounter, never()).increment(); } @ParameterizedTest @@ -356,34 +423,6 @@ public void testBulkRequestAPIRequestTimeout() throws Exception { verify(requestsReceivedCounter, times(1)).increment(); } - private void testBadRequestWithPayload(boolean testBulkRequestAPIWithIndexInPath, String requestBody) throws Exception { - // Prepare - RequestHeaders requestHeaders = RequestHeaders.builder() - .contentType(MediaType.JSON) - .method(HttpMethod.POST) - .path("/") - .build(); - - HttpData httpData = HttpData.ofUtf8(requestBody); - AggregatedHttpRequest testBadRequest = HttpRequest.of(requestHeaders, httpData).aggregate().get(); - // When - if (testBulkRequestAPIWithIndexInPath) { - assertThrows(IOException.class, () -> openSearchAPIService.doPostBulkIndex(serviceRequestContext, testBadRequest, null, - null, null).aggregate().get()); - } else { - assertThrows(IOException.class, () -> openSearchAPIService.doPostBulk(serviceRequestContext, testBadRequest, null, - null).aggregate().get()); - } - - // Then - verify(requestsReceivedCounter, times(1)).increment(); - verify(successRequestsCounter, never()).increment(); - final ArgumentCaptor payloadLengthCaptor = ArgumentCaptor.forClass(Double.class); - verify(payloadSizeSummary, times(1)).record(payloadLengthCaptor.capture()); - assertEquals(testBadRequest.content().length(), Math.round(payloadLengthCaptor.getValue())); - verify(requestProcessDuration, times(1)).recordCallable(ArgumentMatchers.>any()); - } - private AggregatedHttpRequest generateRandomValidBulkRequest(int numJson) throws JsonProcessingException, ExecutionException, InterruptedException { RequestHeaders requestHeaders = RequestHeaders.builder() diff --git a/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java b/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java index c9d77b77fa..377b146058 100644 --- a/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java +++ b/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISourceTest.java @@ -293,11 +293,12 @@ public void testHealthCheckUnauthenticatedDisabled() { .whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.UNAUTHORIZED)).join(); } + // Empty payload produces zero events, returns 200. @ParameterizedTest @ValueSource(booleans = {false, true}) - public void testBulkRequestJsonResponse400WithEmptyPayload(boolean includeIndexInPath) { + public void testBulkRequestWithEmptyPayloadReturns200(boolean includeIndexInPath) { // Prepare - final String testBadData = ""; //Empty body + final String testEmptyData = ""; openSearchAPISource.start(testBuffer); refreshMeasurements(); @@ -309,9 +310,9 @@ public void testBulkRequestJsonResponse400WithEmptyPayload(boolean includeIndexI .path(includeIndexInPath ? "/" + testIndex + "/_bulk" + testQueryParams : "/_bulk" + testQueryParams) .contentType(MediaType.JSON_UTF_8) .build(), - HttpData.ofUtf8(testBadData)) + HttpData.ofUtf8(testEmptyData)) .aggregate() - .whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.BAD_REQUEST)).join(); + .whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.OK)).join(); // Then Assertions.assertTrue(testBuffer.isEmpty()); @@ -319,20 +320,21 @@ public void testBulkRequestJsonResponse400WithEmptyPayload(boolean includeIndexI final Measurement requestReceivedCount = MetricsTestUtil.getMeasurementFromList( requestsReceivedMeasurements, Statistic.COUNT); Assertions.assertEquals(1.0, requestReceivedCount.getValue()); - final Measurement badRequestsCount = MetricsTestUtil.getMeasurementFromList( - badRequestsMeasurements, Statistic.COUNT); - Assertions.assertEquals(1.0, badRequestsCount.getValue()); + final Measurement successRequestsCount = MetricsTestUtil.getMeasurementFromList( + successRequestsMeasurements, Statistic.COUNT); + Assertions.assertEquals(1.0, successRequestsCount.getValue()); } + // Decoder treats the second action line as document data for the first. Returns 200. @ParameterizedTest @ValueSource(booleans = {false, true}) - public void testBulkRequestJsonResponse400WithInvalidPayload(boolean includeIndexInPath) throws Exception { - // Prepare + public void testBulkRequestWithInvalidPayloadReturns200(boolean includeIndexInPath) throws Exception { + // Prepare - two action lines without doc lines: decoder treats second as doc for first List jsonList = new ArrayList<>(); for (int i = 0; i < 2; i++) { jsonList.add(mapper.writeValueAsString(Collections.singletonMap("index", Collections.singletonMap("_index", "test-index")))); } - final String testBadData = String.join("\n", jsonList); + final String testData = String.join("\n", jsonList); openSearchAPISource.start(testBuffer); refreshMeasurements(); @@ -344,19 +346,17 @@ public void testBulkRequestJsonResponse400WithInvalidPayload(boolean includeInde .path(includeIndexInPath ? "/" + testIndex + "/_bulk" + testQueryParams : "/_bulk" + testQueryParams) .contentType(MediaType.JSON_UTF_8) .build(), - HttpData.ofUtf8(testBadData)) + HttpData.ofUtf8(testData)) .aggregate() - .whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.BAD_REQUEST)).join(); + .whenComplete((i, ex) -> assertSecureResponseWithStatusCode(i, HttpStatus.OK)).join(); // Then - Assertions.assertTrue(testBuffer.isEmpty()); - // Verify metrics final Measurement requestReceivedCount = MetricsTestUtil.getMeasurementFromList( requestsReceivedMeasurements, Statistic.COUNT); Assertions.assertEquals(1.0, requestReceivedCount.getValue()); - final Measurement badRequestsCount = MetricsTestUtil.getMeasurementFromList( - badRequestsMeasurements, Statistic.COUNT); - Assertions.assertEquals(1.0, badRequestsCount.getValue()); + final Measurement successRequestsCount = MetricsTestUtil.getMeasurementFromList( + successRequestsMeasurements, Statistic.COUNT); + Assertions.assertEquals(1.0, successRequestsCount.getValue()); } @ParameterizedTest diff --git a/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchBulkByteDecoderTest.java b/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchBulkByteDecoderTest.java new file mode 100644 index 0000000000..7fd4d70268 --- /dev/null +++ b/data-prepper-plugins/opensearch-api-source/src/test/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchBulkByteDecoderTest.java @@ -0,0 +1,138 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + * + */ + +package org.opensearch.dataprepper.plugins.source.opensearchapi; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.opensearch.dataprepper.model.event.Event; +import org.opensearch.dataprepper.model.record.Record; +import org.opensearch.dataprepper.plugins.source.opensearchapi.model.BulkAPIEventMetadataKeyAttributes; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class OpenSearchBulkByteDecoderTest { + + private OpenSearchBulkByteDecoder decoder; + + @BeforeEach + void setUp() { + decoder = new OpenSearchBulkByteDecoder(); + } + + @Test + void parse_withIndexAction_createsEventWithCorrectMetadata() throws IOException { + String bulk = "{\"index\":{\"_index\":\"my-index\",\"_id\":\"1\"}}\n" + + "{\"field1\":\"value1\",\"field2\":42}\n"; + + List> records = parseAll(bulk); + + assertEquals(1, records.size()); + Event event = records.get(0).getData(); + assertEquals("value1", event.get("field1", String.class)); + assertEquals(42, event.get("field2", Integer.class)); + assertEquals("index", event.getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION)); + assertEquals("my-index", event.getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_INDEX)); + assertEquals("1", event.getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ID)); + } + + @Test + void parse_withDeleteAction_createsEventWithNoBody() throws IOException { + String bulk = "{\"delete\":{\"_index\":\"my-index\",\"_id\":\"2\"}}\n"; + + List> records = parseAll(bulk); + + assertEquals(1, records.size()); + Event event = records.get(0).getData(); + assertEquals("delete", event.getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION)); + assertEquals("my-index", event.getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_INDEX)); + } + + @Test + void parse_withMultipleActions_createsMultipleEvents() throws IOException { + String bulk = "{\"index\":{\"_index\":\"idx\",\"_id\":\"1\"}}\n" + + "{\"name\":\"doc1\"}\n" + + "{\"create\":{\"_index\":\"idx\",\"_id\":\"2\"}}\n" + + "{\"name\":\"doc2\"}\n" + + "{\"delete\":{\"_index\":\"idx\",\"_id\":\"3\"}}\n"; + + List> records = parseAll(bulk); + + assertEquals(3, records.size()); + assertEquals("index", records.get(0).getData().getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION)); + assertEquals("create", records.get(1).getData().getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION)); + assertEquals("delete", records.get(2).getData().getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION)); + } + + @Test + void parse_withRoutingAndPipeline_setsMetadata() throws IOException { + String bulk = "{\"index\":{\"_index\":\"idx\",\"routing\":\"r1\",\"pipeline\":\"p1\"}}\n" + + "{\"data\":true}\n"; + + List> records = parseAll(bulk); + + assertEquals(1, records.size()); + Event event = records.get(0).getData(); + assertEquals("r1", event.getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ROUTING)); + assertEquals("p1", event.getMetadata().getAttribute( + BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_PIPELINE)); + } + + @Test + void parse_withBlankLines_skipsBlankLines() throws IOException { + String bulk = "\n\n{\"index\":{\"_index\":\"idx\"}}\n{\"x\":1}\n\n"; + + List> records = parseAll(bulk); + + assertEquals(1, records.size()); + } + + @Test + void parse_withEmptyInput_returnsNoEvents() throws IOException { + List> records = parseAll(""); + assertTrue(records.isEmpty()); + } + + @Test + void parse_withTimeReceived_setsTimeOnEvent() throws IOException { + String bulk = "{\"index\":{\"_index\":\"idx\"}}\n{\"x\":1}\n"; + Instant now = Instant.now(); + + List> records = new ArrayList<>(); + decoder.parse(new ByteArrayInputStream(bulk.getBytes(StandardCharsets.UTF_8)), now, records::add); + + assertEquals(1, records.size()); + assertNotNull(records.get(0).getData().getMetadata().getTimeReceived()); + } + + private List> parseAll(String input) throws IOException { + List> records = new ArrayList<>(); + decoder.parse(new ByteArrayInputStream(input.getBytes(StandardCharsets.UTF_8)), Instant.now(), records::add); + return records; + } +}