add support for OpenSearch formats & update readme

Signed-off-by: Shenoy Pratik <sgguruda@amazon.com>

Author: ps48

data-prepper-plugins/otlp-source/README.md

@@ -14,9 +14,9 @@ otel-telemetry-pipeline:
     otlp:
       ssl: false
   route:
-    - logs: 'getMetadata("eventType") == "LOG"'
-    - traces: 'getMetadata("eventType") == "TRACE"'
-    - metrics: 'getMetadata("eventType") == "METRIC"'
+    - logs: 'getEventType() == "LOG"'
+    - traces: 'getEventType() == "TRACE"'
+    - metrics: 'getEventType() == "METRIC"'
   sink:
     - pipeline:
         name: "logs-pipeline"
@@ -47,6 +47,15 @@ otel-telemetry-pipeline:
 - compression(Optional) => The compression type applied on the client request payload. Defaults to `none`. Supported values are:
   - `none`: no compression
   - `gzip`: apply GZip de-compression on the incoming request.
+- logs_output_format(Optional) => Specifies the decoded output format for logs. Supported values are:
+  - `otel`: OpenTelemetry format (default).
+  - `opensearch`: OpenSearch format.
+- metrics_output_format(Optional) => Specifies the decoded output format for metrics. Supported values are:
+  - `otel`: OpenTelemetry format (default).
+  - `opensearch`: OpenSearch format.
+- traces_output_format(Optional) => Specifies the decoded output format for traces. Supported values are:
+  - `otel`: OpenTelemetry format (default).
+  - `opensearch`: OpenSearch format.
 
 ### Retry Information
 

data-prepper-plugins/otlp-source/src/main/java/org/opensearch/dataprepper/plugins/source/otlp/OTLPSource.java

@@ -13,6 +13,8 @@
 
 import org.opensearch.dataprepper.model.source.Source;
 import org.opensearch.dataprepper.plugins.certificate.CertificateProvider;
+import org.opensearch.dataprepper.plugins.otel.codec.OTelOutputFormat;
+import org.opensearch.dataprepper.plugins.otel.codec.OTelProtoOpensearchCodec;
 import org.opensearch.dataprepper.plugins.otel.codec.OTelProtoStandardCodec;
 import org.opensearch.dataprepper.plugins.source.otellogs.OTelLogsGrpcService;
 import org.opensearch.dataprepper.plugins.source.otelmetrics.OTelMetricsGrpcService;
@@ -87,17 +89,17 @@ public void start(Buffer<Record<Object>> buffer) {
 
       final OTelLogsGrpcService oTelLogsGrpcService = new OTelLogsGrpcService(
           (int) (otlpSourceConfig.getRequestTimeoutInMillis() * 0.8),
-          new OTelProtoStandardCodec.OTelProtoDecoder(),
+          otlpSourceConfig.getLogsOutputFormat() == OTelOutputFormat.OPENSEARCH ? new OTelProtoOpensearchCodec.OTelProtoDecoder() : new OTelProtoStandardCodec.OTelProtoDecoder(),
           buffer, pluginMetrics);
 
       final OTelMetricsGrpcService oTelMetricsGrpcService = new OTelMetricsGrpcService(
           (int) (otlpSourceConfig.getRequestTimeoutInMillis() * 0.8),
-          new OTelProtoStandardCodec.OTelProtoDecoder(),
+          otlpSourceConfig.getMetricsOutputFormat() == OTelOutputFormat.OPENSEARCH ? new OTelProtoOpensearchCodec.OTelProtoDecoder() : new OTelProtoStandardCodec.OTelProtoDecoder(),
           metricBuffer, pluginMetrics);
 
       final OTelTraceGrpcService oTelTraceGrpcService = new OTelTraceGrpcService(
           (int) (otlpSourceConfig.getRequestTimeoutInMillis() * 0.8),
-          new OTelProtoStandardCodec.OTelProtoDecoder(),
+          otlpSourceConfig.getTracesOutputFormat() == OTelOutputFormat.OPENSEARCH ? new OTelProtoOpensearchCodec.OTelProtoDecoder() : new OTelProtoStandardCodec.OTelProtoDecoder(),
           buffer, pluginMetrics);
 
       ServerConfiguration serverConfiguration = ConvertConfiguration.convertConfiguration(otlpSourceConfig);

data-prepper-plugins/otlp-source/src/main/java/org/opensearch/dataprepper/plugins/source/otlp/OTLPSourceConfig.java

@@ -11,6 +11,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.opensearch.dataprepper.model.types.ByteCount;
 import org.opensearch.dataprepper.plugins.codec.CompressionOption;
+import org.opensearch.dataprepper.plugins.otel.codec.OTelOutputFormat;
 import org.opensearch.dataprepper.plugins.server.RetryInfoConfig;
 import org.opensearch.dataprepper.model.configuration.PluginModel;
 
@@ -21,6 +22,9 @@ public class OTLPSourceConfig {
   static final String METRICS_PATH = "metrics_path";
   static final String TRACES_PATH = "traces_path";
   static final String SSL = "ssl";
+  static final String LOGS_OUTPUT_FORMAT = "logs_output_format";
+  static final String METRICS_OUTPUT_FORMAT = "metrics_output_format";
+  static final String TRACES_OUTPUT_FORMAT = "traces_output_format";
   static final String USE_ACM_CERT_FOR_SSL = "useAcmCertForSSL";
   static final String ACM_CERT_ISSUE_TIME_OUT_MILLIS = "acmCertIssueTimeOutMillis";
   static final String HEALTH_CHECK_SERVICE = "health_check_service";
@@ -78,6 +82,15 @@ public class OTLPSourceConfig {
   @JsonProperty(SSL)
   private boolean ssl = DEFAULT_SSL;
 
+  @JsonProperty(LOGS_OUTPUT_FORMAT)
+  private OTelOutputFormat logsOutputFormat = OTelOutputFormat.OTEL;
+ 
+  @JsonProperty(METRICS_OUTPUT_FORMAT)
+  private OTelOutputFormat metricsOutputFormat = OTelOutputFormat.OTEL;
+
+  @JsonProperty(TRACES_OUTPUT_FORMAT)
+  private OTelOutputFormat tracesOutputFormat = OTelOutputFormat.OTEL;
+
   @JsonProperty(USE_ACM_CERT_FOR_SSL)
   private boolean useAcmCertForSSL = DEFAULT_USE_ACM_CERT_FOR_SSL;
 
@@ -181,6 +194,18 @@ public int getRequestTimeoutInMillis() {
     return requestTimeoutInMillis;
   }
 
+  public OTelOutputFormat getLogsOutputFormat() {
+    return logsOutputFormat;
+  }
+
+  public OTelOutputFormat getMetricsOutputFormat() {
+    return metricsOutputFormat;
+  }
+
+  public OTelOutputFormat getTracesOutputFormat() {
+    return tracesOutputFormat;
+  }
+
   public int getPort() {
     return port;
   }

data-prepper-plugins/otlp-source/src/test/java/org/opensearch/dataprepper/plugins/source/otlp/OTLPSourceConfigTests.java

@@ -13,6 +13,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.plugins.codec.CompressionOption;
+import org.opensearch.dataprepper.plugins.otel.codec.OTelOutputFormat;
 import org.opensearch.dataprepper.plugins.server.RetryInfoConfig;
 
 import java.time.Duration;
@@ -369,6 +370,30 @@ void testInvalidPathsDoNotStartWithSlash() {
     assertFalse(config.isTracesPathValid());
   }
 
+  @Test
+  void testDefaultOutputFormats() {
+    final OTLPSourceConfig config = new OTLPSourceConfig();
+
+    assertEquals(OTelOutputFormat.OTEL, config.getLogsOutputFormat());
+    assertEquals(OTelOutputFormat.OTEL, config.getMetricsOutputFormat());
+    assertEquals(OTelOutputFormat.OTEL, config.getTracesOutputFormat());
+  }
+
+  @Test
+  void testCustomOutputFormats() {
+    final Map<String, Object> settings = new HashMap<>();
+    settings.put(OTLPSourceConfig.LOGS_OUTPUT_FORMAT, OTelOutputFormat.OPENSEARCH.getFormatName());
+    settings.put(OTLPSourceConfig.METRICS_OUTPUT_FORMAT, OTelOutputFormat.OPENSEARCH.getFormatName());
+    settings.put(OTLPSourceConfig.TRACES_OUTPUT_FORMAT, OTelOutputFormat.OPENSEARCH.getFormatName());
+
+    final PluginSetting pluginSetting = new PluginSetting(PLUGIN_NAME, settings);
+    final OTLPSourceConfig config = OBJECT_MAPPER.convertValue(pluginSetting.getSettings(), OTLPSourceConfig.class);
+
+    assertEquals(OTelOutputFormat.OPENSEARCH, config.getLogsOutputFormat());
+    assertEquals(OTelOutputFormat.OPENSEARCH, config.getMetricsOutputFormat());
+    assertEquals(OTelOutputFormat.OPENSEARCH, config.getTracesOutputFormat());
+  }
+
   private PluginSetting completePluginSetting(final int requestTimeoutInMillis, final int port, final String path,
       final boolean healthCheck, final boolean protoReflectionService,
       final boolean enableUnframedRequests, final boolean isSSL,