feat(otlp-sink): Add support for metrics and logs with configurable SigV4 and headers (#6768)

* feat(otlp-sink): Add support for metrics and logs with configurable SigV4 and headers

Extend the OTLP sink plugin to support all three signal types (traces,
metrics, logs) with full OTLP protocol encoding, configurable SigV4
signing, and custom HTTP headers. Based on the work from @viquer in
#6488, with additional enhancements.

Multi-signal support:
- Extended OTelProtoCodec with convertToResourceMetrics() and
  convertToResourceLogs() for encoding all signal types
- Added generic OtlpSignalHandler<T> interface with type-safe
  implementations for traces, metrics, and logs
- Per-signal-type buffer architecture for optimal batching
- Signal type determined automatically at runtime from event type
- Generalized sink from Record<Span> to Record<Event>

Configurable SigV4 signing:
- Added service_name field to OtlpSinkConfig (default: xray)
- SigV4Signer uses configurable service name

Additional headers:
- Added additional_headers map config to OtlpSinkConfig
- Protected header blocklist prevents overriding signed headers
- Headers injected after SigV4 signing (not included in signature)

Per-signal metrics:
- Added per-signal counters (rejectedTracesCount, failedMetricsCount, etc.)
- Aggregate counters (rejectedRecordsCount, failedRecordsCount) retained

Region from AWS config:
- getAwsRegion() checks aws.region first, falls back to endpoint parsing

Breaking changes (plugin is @experimental):
- OtlpSink changed from AbstractSink<Record<Span>> to
  AbstractSink<Record<Event>>
- Metric counters renamed: rejectedSpansCount to rejectedRecordsCount,
  failedSpansCount to failedRecordsCount

Signed-off-by: Roberto Ramirez Vique <viquer@amazon.com>
Signed-off-by: Laszlo Kovacs <laszlokv@amazon.com>

* fix(otlp-sink): Address code review findings

Fix EventHandle leak when encodeEvent() returns null or throws in
OtlpSinkBuffer.runTyped() — release handle in both paths.

Fix SeverityNumber.forNumber() null safety in OTelProtoStandardCodec
to prevent NPE on unrecognized severity numbers.

Cache per-signal metric counters at construction time in
OtlpSinkMetrics to avoid string concatenation and counter lookup on
every increment call in the hot path.

Remove stale OTLP_PATH constant and fallback URI from SigV4Signer
since endpoint is a required config field.

Replace fully-qualified class names with imports in OTelProtoCodec
OTelProtoEncoder interface (ResourceMetrics, ResourceLogs, Log).

Update README: replace stale flush-on-signal-change batching
description with accurate per-signal buffer architecture, add
service_name to configuration options table.

Replace Thread.sleep(300) with Awaitility in OtlpHttpSenderTest for
deterministic async assertions.

Replace OutOfMemoryError with Error in OtlpSinkBufferMultiSignalTest
to avoid triggering JVM OOM handlers in CI.

Signed-off-by: Laszlo Kovacs <laszlokv@amazon.com>

* fix(otlp-sink): Keep rejectedSpansCount/failedSpansCount metric names

Rename TRACE metrics label from 'Traces' to 'Spans' to preserve
backwards compatibility with existing metric allowlists and dashboards.
Per-signal counters now emit rejectedSpansCount, rejectedMetricsCount,
rejectedLogsCount (and corresponding failed* counters).

Remove stale migration note from README since the original metric
names are preserved.

Signed-off-by: Laszlo Kovacs <laszlokv@amazon.com>

---------

Signed-off-by: Roberto Ramirez Vique <viquer@amazon.com>
Signed-off-by: Laszlo Kovacs <laszlokv@amazon.com>

Author: LaszloKovacs9001

.gitignore

@@ -27,6 +27,9 @@ data-prepper-main/src/test/resources/logstash-conf/logstash-filter.yaml
 *.iml
 .kiro
 .vscode
+.project
+.classpath
+.settings/
 
 # Manual testing files (should not be committed)
 MANUAL_TEST.md

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

@@ -5,24 +5,25 @@
 
 package org.opensearch.dataprepper.plugins.otel.codec;
 
-import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
+import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
-import io.opentelemetry.proto.trace.v1.ResourceSpans;
-import io.opentelemetry.proto.resource.v1.Resource;
+import io.opentelemetry.proto.logs.v1.ResourceLogs;
 import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
 import io.opentelemetry.proto.metrics.v1.ScopeMetrics;
+import io.opentelemetry.proto.resource.v1.Resource;
+import io.opentelemetry.proto.trace.v1.ResourceSpans;
 import org.apache.commons.codec.DecoderException;
-
+import org.opensearch.dataprepper.model.log.Log;
+import org.opensearch.dataprepper.model.log.OpenTelemetryLog;
 import org.opensearch.dataprepper.model.metric.Metric;
 import org.opensearch.dataprepper.model.record.Record;
-import org.opensearch.dataprepper.model.log.OpenTelemetryLog;
 import org.opensearch.dataprepper.model.trace.Span;
 
 import java.io.UnsupportedEncodingException;
 import java.time.Instant;
-import java.util.Collection;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -150,6 +151,10 @@ Collection<Record<? extends Metric>> parseExportMetricsServiceRequest(
 
     public interface OTelProtoEncoder {
         ResourceSpans convertToResourceSpans(final Span span) throws UnsupportedEncodingException, DecoderException;
+        
+        ResourceMetrics convertToResourceMetrics(final Metric metric) throws UnsupportedEncodingException, DecoderException;
+        
+        ResourceLogs convertToResourceLogs(final Log log) throws UnsupportedEncodingException, DecoderException;
     }
 
 }

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

@@ -8,8 +8,8 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.protobuf.ByteString;
-import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
+import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
 import io.opentelemetry.proto.common.v1.AnyValue;
 import io.opentelemetry.proto.common.v1.InstrumentationScope;
@@ -18,15 +18,15 @@
 import io.opentelemetry.proto.logs.v1.ResourceLogs;
 import io.opentelemetry.proto.metrics.v1.ExponentialHistogramDataPoint;
 import io.opentelemetry.proto.metrics.v1.NumberDataPoint;
-import io.opentelemetry.proto.metrics.v1.SummaryDataPoint;
 import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
 import io.opentelemetry.proto.metrics.v1.ScopeMetrics;
+import io.opentelemetry.proto.metrics.v1.SummaryDataPoint;
 import io.opentelemetry.proto.resource.v1.Resource;
 import io.opentelemetry.proto.trace.v1.ResourceSpans;
 import io.opentelemetry.proto.trace.v1.ScopeSpans;
 import io.opentelemetry.proto.trace.v1.Status;
-import org.apache.commons.codec.binary.Hex;
 import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.binary.Hex;
 import org.opensearch.dataprepper.model.log.JacksonOtelLog;
 import org.opensearch.dataprepper.model.log.OpenTelemetryLog;
 import org.opensearch.dataprepper.model.metric.Bucket;
@@ -53,9 +53,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import static org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCommonUtils.convertUnixNanosToISO8601;
-import static org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCommonUtils.convertISO8601ToNanos;
-import static org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCommonUtils.convertByteStringToString;
 import java.io.UnsupportedEncodingException;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -73,6 +70,10 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCommonUtils.convertByteStringToString;
+import static org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCommonUtils.convertISO8601ToNanos;
+import static org.opensearch.dataprepper.plugins.otel.codec.OTelProtoCommonUtils.convertUnixNanosToISO8601;
+
 /**
  * OTelProtoOpensearchCodec is for encoding/decoding between {@link org.opensearch.dataprepper.model.trace} and {@link io.opentelemetry.proto} in Opensearch friendly way.
  */
@@ -941,6 +942,20 @@ protected AnyValue objectToAnyValue(final Object obj) throws UnsupportedEncoding
 
             return anyValueBuilder.build();
         }
+
+        @Override
+        public io.opentelemetry.proto.metrics.v1.ResourceMetrics convertToResourceMetrics(final org.opensearch.dataprepper.model.metric.Metric metric) 
+                throws UnsupportedEncodingException, DecoderException {
+            throw new UnsupportedOperationException(
+                "OTelProtoOpensearchCodec does not support metric encoding. Use OTelProtoStandardCodec instead.");
+        }
+
+        @Override
+        public io.opentelemetry.proto.logs.v1.ResourceLogs convertToResourceLogs(final org.opensearch.dataprepper.model.log.Log log) 
+                throws UnsupportedEncodingException, DecoderException {
+            throw new UnsupportedOperationException(
+                "OTelProtoOpensearchCodec does not support log encoding. Use OTelProtoStandardCodec instead.");
+        }
     }
 
     /**