Add e2e test for protobuf requests

Signed-off-by: Tomas Longo <tlongo@sternad.de>

Author: TomasLongo

data-prepper-plugins/otel-logs-source/src/main/java/org/opensearch/dataprepper/plugins/source/otellogs/http/ArmeriaHttpService.java

@@ -17,7 +17,8 @@
 import org.slf4j.LoggerFactory;
 
 import com.linecorp.armeria.server.ServiceRequestContext;
-import com.linecorp.armeria.server.annotation.Consumes;
+import com.linecorp.armeria.server.annotation.ConsumesJson;
+import com.linecorp.armeria.server.annotation.ConsumesProtobuf;
 import com.linecorp.armeria.server.annotation.Post;
 
 import io.micrometer.core.instrument.Counter;
@@ -57,7 +58,8 @@ public ArmeriaHttpService(Buffer<Record<Object>> buffer, final PluginMetrics plu
 
     // no path provided. Will be set by config.
     @Post("")
-    @Consumes(value = "application/json")
+    @ConsumesJson
+    @ConsumesProtobuf
     public ExportLogsServiceResponse exportLog(ExportLogsServiceRequest request) {
         requestsReceivedCounter.increment();
         payloadSizeSummary.record(request.getSerializedSize());

e2e-test/log/build.gradle

@@ -92,6 +92,14 @@ List<LogTestConfiguration> logTestConfigurations = [
                 'otel-logs-source-unframed-requests-pipeline.yml',
                 'data_prepper.yml'
         ),
+        new LogTestConfiguration(
+                'basicLogOtelLogsSourceProtobufPayloadEndToEndTest',
+                'Runs the basic end-to-end test using the otel logs source grpc service with unframed requests enabled.',
+                'org.opensearch.dataprepper.integration.log.EndToEndOtelLogsSourceTest.testOtelLogsSourcePipelineWithProtobufPayloadEndToEnd*',
+                'data-prepper-otel-logs-source',
+                'otel-logs-source-pipeline.yml',
+                'data_prepper.yml'
+        ),
 ]
 
 

e2e-test/log/src/integrationTest/java/org/opensearch/dataprepper/integration/log/EndToEndOtelLogsSourceTest.java

@@ -56,20 +56,27 @@ public class EndToEndOtelLogsSourceTest {
 
     @Test
     public void testOtelLogsSourcePipelineEndToEnd() throws InvalidProtocolBufferException {
-        ingestLogs("/otel-logs-pipeline/logs");
+        ingestLogs("/otel-logs-pipeline/logs", createOtelLogsJsonRequest());
 
         searchForLogsAndAssert();
     }
 
     @Test
     public void testOtelLogsSourceWithUnframedRequestsPipelineEndToEnd() throws InvalidProtocolBufferException {
-        ingestLogs("/opentelemetry.proto.collector.logs.v1.LogsService/Export");
+        ingestLogs("/opentelemetry.proto.collector.logs.v1.LogsService/Export", createOtelLogsJsonRequest());
+
+        searchForLogsAndAssert();
+    }
+
+    @Test
+    public void testOtelLogsSourcePipelineWithProtobufPayloadEndToEnd() throws InvalidProtocolBufferException {
+        ingestLogs("/otel-logs-pipeline/logs", createOtelLogsProtobufRequest());
 
         searchForLogsAndAssert();
     }
 
 
-    private HttpData createOtelLogsHttpRequest() throws InvalidProtocolBufferException {
+    private HttpData createOtelLogsJsonRequest() throws InvalidProtocolBufferException {
         ExportLogsServiceRequest exportLogsServiceRequest = ExportLogsServiceRequest.newBuilder().addResourceLogs(
                 ResourceLogs.newBuilder()
                         .addScopeLogs(ScopeLogs.newBuilder()
@@ -86,6 +93,23 @@ private HttpData createOtelLogsHttpRequest() throws InvalidProtocolBufferExcepti
         return HttpData.copyOf(JsonFormat.printer().print(exportLogsServiceRequest).getBytes());
     }
 
+    private HttpData createOtelLogsProtobufRequest() throws InvalidProtocolBufferException {
+        ExportLogsServiceRequest exportLogsServiceRequest = ExportLogsServiceRequest.newBuilder().addResourceLogs(
+                ResourceLogs.newBuilder()
+                        .addScopeLogs(ScopeLogs.newBuilder()
+                                .addLogRecords(LogRecord.newBuilder()
+                                        .setBody(AnyValue.newBuilder().setStringValue(apacheLogFaker.generateRandomCommonApacheLog()).build())
+                                        .setSeverityNumberValue(1)
+                                        .setTimeUnixNano(System.currentTimeMillis() * 1_000_000)
+                                        .setTraceId(ByteString.copyFromUtf8("trace-id"))
+                                        .setSpanId(ByteString.copyFromUtf8("span-id")
+                                        )).build()
+                        )
+        ).build();
+
+        return HttpData.copyOf(exportLogsServiceRequest.toByteArray());
+    }
+
     private void searchForLogsAndAssert() {
         await().atMost(10, TimeUnit.SECONDS).untilAsserted(
                 () -> {
@@ -109,7 +133,7 @@ private RestHighLevelClient createOpenSearchClient() {
                 .createClient(null);
     }
 
-    private void ingestLogs(String path) throws InvalidProtocolBufferException {
+    private void ingestLogs(String path, HttpData payload) throws InvalidProtocolBufferException {
         RequestHeaders headers = RequestHeaders.builder()
                 .scheme(SessionProtocol.HTTP)
                 .authority(String.format("127.0.0.1:%d", SOURCE_PORT))
@@ -118,7 +142,7 @@ private void ingestLogs(String path) throws InvalidProtocolBufferException {
                 .contentType(MediaType.JSON_UTF_8)
                 .build();
 
-        WebClient.of().execute(headers, createOtelLogsHttpRequest())
+        WebClient.of().execute(headers, payload)
                 .aggregate()
                 .whenComplete((i, ex) -> assertThat(i.status(), is(HttpStatus.OK)))
                 .join();