Support otel_logs codec in S3 source (#5028)

data-prepper-plugins/otel-proto-common/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OTLPJsonLogsCodec.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.otel.codec;
+
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.codec.InputCodec;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.function.Consumer;
+
+@DataPrepperPlugin(name = "opentelemetry_logs", pluginType = InputCodec.class)
+public class OTLPJsonLogsCodec extends OTLPJsonLogsDecoder implements InputCodec {
+    public void parse(InputStream inputStream, Consumer<Record<Event>> eventConsumer) throws IOException {
+        parse(inputStream, null, eventConsumer);
+    }      
+}

data-prepper-plugins/otel-proto-common/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OTLPJsonLogsDecoder.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.dataprepper.plugins.otel.codec;
+
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
+import org.opensearch.dataprepper.model.codec.ByteDecoder;
+import org.opensearch.dataprepper.model.record.Record;
+
+import com.google.protobuf.util.JsonFormat;
+
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.log.OpenTelemetryLog;
+
+import java.util.List;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.util.function.Consumer;
+import java.time.Instant;
+
+
+public class OTLPJsonLogsDecoder implements ByteDecoder {
+    private final OTelProtoCodec.OTelProtoDecoder otelProtoDecoder;
+    public OTLPJsonLogsDecoder() {
+        otelProtoDecoder = new OTelProtoCodec.OTelProtoDecoder();
+    }
+    public void parse(InputStream inputStream, Instant timeReceivedMs, Consumer<Record<Event>> eventConsumer) throws IOException {
+        Reader reader = new InputStreamReader(inputStream);
+        ExportLogsServiceRequest.Builder builder = ExportLogsServiceRequest.newBuilder();
+        JsonFormat.parser().merge(reader, builder);
+        ExportLogsServiceRequest request = builder.build(); 
+        List<OpenTelemetryLog> logs = otelProtoDecoder.parseExportLogsServiceRequest(request, timeReceivedMs);
+        for (OpenTelemetryLog log: logs) {
+            eventConsumer.accept(new Record<>(log));
+        }
+    }
+
+}

data-prepper-plugins/otel-proto-common/src/test/java/org/opensearch/dataprepper/plugins/otel/codec/OTLPJsonLogsCodecTest.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.otel.codec;
+
+import java.io.InputStream;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.model.log.OpenTelemetryLog;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class OTLPJsonLogsCodecTest {
+    private static final String TEST_REQUEST_LOGS_FILE = "test-request-multiple-logs.json";
+
+    public OTLPJsonLogsCodec createObjectUnderTest() {
+        return new OTLPJsonLogsCodec();
+    }    
+
+    private void validateLog(OpenTelemetryLog logRecord) {
+        assertThat(logRecord.getServiceName(), is("service"));
+        assertThat(logRecord.getTime(), is("2020-05-24T14:00:00Z"));
+        assertThat(logRecord.getObservedTime(), is("2020-05-24T14:00:02Z"));
+        assertThat(logRecord.getBody(), is("Log value"));
+        assertThat(logRecord.getDroppedAttributesCount(), is(3));
+        assertThat(logRecord.getSchemaUrl(), is("schemaurl"));
+        assertThat(logRecord.getSeverityNumber(), is(5));
+        assertThat(logRecord.getSeverityText(), is("Severity value"));
+        assertThat(logRecord.getTraceId(), is("ba1a1c23b4093b63"));
+        assertThat(logRecord.getSpanId(), is("2cc83ac90ebc469c"));
+        Map<String, Object> mergedAttributes = logRecord.getAttributes();
+        assertThat(mergedAttributes.keySet().size(), is(2));
+        assertThat(mergedAttributes.get("log.attributes.statement@params"), is("us-east-1"));
+        assertThat(mergedAttributes.get("resource.attributes.service@name"), is("service"));
+    }
+
+    @Test
+    public void testParse() throws Exception {
+        InputStream inputStream = OTLPJsonLogsCodecTest.class.getClassLoader().getResourceAsStream(TEST_REQUEST_LOGS_FILE);
+        createObjectUnderTest().parse(inputStream, (record) -> {
+            validateLog((OpenTelemetryLog)record.getData());
+        });
+
+    }    
+}

data-prepper-plugins/otel-proto-common/src/test/java/org/opensearch/dataprepper/plugins/otel/codec/OTLPJsonLogsDecoderTest.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.otel.codec;
+
+import java.io.InputStream;
+import java.time.Instant;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.model.log.OpenTelemetryLog;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class OTLPJsonLogsDecoderTest {
+    private static final String TEST_REQUEST_LOGS_FILE = "test-request-multiple-logs.json";
+
+    public OTLPJsonLogsDecoder createObjectUnderTest() {
+        return new OTLPJsonLogsDecoder();
+    }
+
+    private void validateLog(OpenTelemetryLog logRecord) {
+        assertThat(logRecord.getServiceName(), is("service"));
+        assertThat(logRecord.getTime(), is("2020-05-24T14:00:00Z"));
+        assertThat(logRecord.getObservedTime(), is("2020-05-24T14:00:02Z"));
+        assertThat(logRecord.getBody(), is("Log value"));
+        assertThat(logRecord.getDroppedAttributesCount(), is(3));
+        assertThat(logRecord.getSchemaUrl(), is("schemaurl"));
+        assertThat(logRecord.getSeverityNumber(), is(5));
+        assertThat(logRecord.getSeverityText(), is("Severity value"));
+        assertThat(logRecord.getTraceId(), is("ba1a1c23b4093b63"));
+        assertThat(logRecord.getSpanId(), is("2cc83ac90ebc469c"));
+        Map<String, Object> mergedAttributes = logRecord.getAttributes();
+        assertThat(mergedAttributes.keySet().size(), is(2));
+        assertThat(mergedAttributes.get("log.attributes.statement@params"), is("us-east-1"));
+        assertThat(mergedAttributes.get("resource.attributes.service@name"), is("service"));
+    }
+
+    @Test
+    public void testParse() throws Exception {
+        InputStream inputStream = OTLPJsonLogsDecoderTest.class.getClassLoader().getResourceAsStream(TEST_REQUEST_LOGS_FILE);
+        createObjectUnderTest().parse(inputStream, Instant.now(), (record) -> {
+            validateLog((OpenTelemetryLog)record.getData());
+        });
+
+    }
+}