Support otel_logs codec in S3 source (#5028)

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

@@ -0,0 +1,40 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.otel.codec;
+
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public enum OTelLogsFormatOption {
+    JSON("json");
+
+    private static final Map<String, OTelLogsFormatOption> NAMES_MAP = Arrays.stream(OTelLogsFormatOption.values())
+            .collect(Collectors.toMap(
+                    value -> value.optionName,
+                    value -> value
+            ));
+
+    private final String optionName;
+
+    OTelLogsFormatOption(final String optionName) {
+        this.optionName = optionName;
+    }
+
+    @JsonValue
+    public String getFormatName() {
+        return optionName;
+    }
+
+    @JsonCreator
+    public static OTelLogsFormatOption fromFormatName(final String optionName) {
+        return NAMES_MAP.get(optionName);
+    }
+}

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

@@ -0,0 +1,34 @@
+/*
+ * 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.annotations.DataPrepperPluginConstructor;
+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.Objects;
+import java.util.function.Consumer;
+
+@DataPrepperPlugin(name = "otel_logs", pluginType = InputCodec.class, pluginConfigurationType = OTelLogsInputCodecConfig.class)
+public class OTelLogsInputCodec implements InputCodec {
+    private final OTelLogsInputCodecConfig config;
+
+    @DataPrepperPluginConstructor
+    public OTelLogsInputCodec(final OTelLogsInputCodecConfig config) {
+        Objects.requireNonNull(config);
+        this.config = config;
+    }    
+    public void parse(InputStream inputStream, Consumer<Record<Event>> eventConsumer) throws IOException {
+        if (config.getFormat() == OTelLogsFormatOption.JSON) {
+            OTelLogsJsonDecoder decoder = new OTelLogsJsonDecoder();
+            decoder.parse(inputStream, null, eventConsumer);
+        } 
+    }      
+}

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.otel.codec;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import jakarta.validation.constraints.AssertTrue;
+import jakarta.validation.constraints.NotNull;
+
+import com.fasterxml.jackson.annotation.JsonPropertyDescription;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.annotation.JsonClassDescription; 
+
+@JsonPropertyOrder
+@JsonClassDescription("The <code>otel_logs</code> codec parses log files in S3 that follow the OpenTelemetry Protocol Specification. " +
+        "It creates a Data Prepper log event for each log record along with the resource attributes in the file.")
+public class OTelLogsInputCodecConfig {
+    static final OTelLogsFormatOption DEFAULT_FORMAT = OTelLogsFormatOption.JSON;
+
+    @JsonProperty("format")
+    @JsonPropertyDescription("Specifies the format of the OTel logs. Default is <code>json</code>.") 
+    @NotNull
+    private OTelLogsFormatOption format = DEFAULT_FORMAT;
+
+    public OTelLogsFormatOption getFormat() {
+         return format;
+    }
+
+    @AssertTrue(message = "format must be json.")
+    boolean isValidFormat() {
+        return format != null;
+    }    
+}

data-prepper-plugins/otel-proto-common/src/main/java/org/opensearch/dataprepper/plugins/otel/codec/OTelLogsJsonDecoder.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 OTelLogsJsonDecoder implements ByteDecoder {
+    private final OTelProtoCodec.OTelProtoDecoder otelProtoDecoder;
+
+    public OTelLogsJsonDecoder() {
+        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/OTelLogsFormatOptionTest.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.otel.codec;
+
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.stream.Stream;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.emptyString;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+class OTelLogsFormatOptionTest {
+    @ParameterizedTest
+    @EnumSource(OTelLogsFormatOption.class)
+    void fromFormatName_returns_expected_value(final OTelLogsFormatOption formatOption) {
+        assertThat(OTelLogsFormatOption.fromFormatName(formatOption.getFormatName()), equalTo(formatOption));
+    }
+
+    @ParameterizedTest
+    @EnumSource(OTelLogsFormatOption.class)
+    void getFormatName_returns_non_empty_string_for_all_types(final OTelLogsFormatOption formatOption) {
+        assertThat(formatOption.getFormatName(), notNullValue());
+        assertThat(formatOption.getFormatName(), not(emptyString()));
+    }
+
+    @ParameterizedTest
+    @ArgumentsSource(OTelLogsFormatOptionToKnownName.class)
+    void getFormatName_returns_expected_name(final OTelLogsFormatOption formatOption, final String expectedString) {
+        assertThat(formatOption.getFormatName(), equalTo(expectedString));
+    }
+
+    static class OTelLogsFormatOptionToKnownName implements ArgumentsProvider {
+        @Override
+        public Stream<? extends Arguments> provideArguments(final ExtensionContext extensionContext) {
+            return Stream.of(
+                    arguments(OTelLogsFormatOption.JSON, "json")
+            );
+        }
+    }
+}

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

@@ -0,0 +1,61 @@
+/*
+ * 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.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.opensearch.dataprepper.model.log.OpenTelemetryLog;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class OTelLogsInputCodecTest {
+    private static final String TEST_REQUEST_LOGS_FILE = "test-request-multiple-logs.json";
+
+    @Mock
+    private OTelLogsInputCodecConfig config;
+    @Mock   
+    private OTelLogsInputCodec otelLogsCodec;
+
+    @BeforeEach
+    void setup() {
+        config = new OTelLogsInputCodecConfig();
+        otelLogsCodec = createObjectUnderTest();
+    }
+
+    public OTelLogsInputCodec createObjectUnderTest() {
+        return new OTelLogsInputCodec(config);
+    }    
+
+    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 = OTelLogsInputCodecTest.class.getClassLoader().getResourceAsStream(TEST_REQUEST_LOGS_FILE);
+        otelLogsCodec.parse(inputStream, (record) -> {
+            validateLog((OpenTelemetryLog)record.getData());
+        });
+
+    }    
+}

data-prepper-plugins/otel-proto-common/src/test/java/org/opensearch/dataprepper/plugins/otel/codec/OTelLogsJsonDecoderTest.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 OTelLogsJsonDecoderTest {
+    private static final String TEST_REQUEST_LOGS_FILE = "test-request-multiple-logs.json";
+
+    public OTelLogsJsonDecoder createObjectUnderTest() {
+        return new OTelLogsJsonDecoder();
+    }
+
+    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 = OTelLogsJsonDecoderTest.class.getClassLoader().getResourceAsStream(TEST_REQUEST_LOGS_FILE);
+        createObjectUnderTest().parse(inputStream, Instant.now(), (record) -> {
+            validateLog((OpenTelemetryLog)record.getData());
+        });
+
+    }
+}