Introduce otlp/http support

data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/GrpcRequestExceptionHandler.java

@@ -39,14 +39,14 @@ public class GrpcRequestExceptionHandler implements GoogleGrpcExceptionHandlerFu
     private final Counter badRequestsCounter;
     private final Counter requestsTooLargeCounter;
     private final Counter internalServerErrorCounter;
-    private final GrpcRetryInfoCalculator retryInfoCalculator;
+    private final RetryInfoCalculator retryInfoCalculator;
 
     public GrpcRequestExceptionHandler(final PluginMetrics pluginMetrics, Duration retryInfoMinDelay, Duration retryInfoMaxDelay) {
         requestTimeoutsCounter = pluginMetrics.counter(REQUEST_TIMEOUTS);
         badRequestsCounter = pluginMetrics.counter(BAD_REQUESTS);
         requestsTooLargeCounter = pluginMetrics.counter(REQUESTS_TOO_LARGE);
         internalServerErrorCounter = pluginMetrics.counter(INTERNAL_SERVER_ERROR);
-        retryInfoCalculator = new GrpcRetryInfoCalculator(retryInfoMinDelay, retryInfoMaxDelay);
+        retryInfoCalculator = new RetryInfoCalculator(retryInfoMinDelay, retryInfoMaxDelay);
     }
 
     @Override

data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/GrpcRetryInfoCalculator.java → data-prepper-plugins/armeria-common/src/main/java/org/opensearch/dataprepper/RetryInfoCalculator.java

@@ -6,15 +6,15 @@
 import java.time.Instant;
 import java.util.concurrent.atomic.AtomicReference;
 
-class GrpcRetryInfoCalculator {
+public class RetryInfoCalculator {
 
     private final Duration minimumDelay;
     private final Duration maximumDelay;
 
     private final AtomicReference<Instant> lastTimeCalled;
     private final AtomicReference<Duration> nextDelay;
 
-    GrpcRetryInfoCalculator(Duration minimumDelay, Duration maximumDelay) {
+    public RetryInfoCalculator(Duration minimumDelay, Duration maximumDelay) {
         this.minimumDelay = minimumDelay;
         this.maximumDelay = maximumDelay;
         // Create a cushion so that the calculator treats a first quick exception (after prepper startup) as normal request (e.g. does not calculate a backoff)
@@ -34,7 +34,7 @@ private static com.google.protobuf.Duration.Builder mapDuration(Duration duratio
         return com.google.protobuf.Duration.newBuilder().setSeconds(duration.getSeconds()).setNanos(duration.getNano());
     }
 
-    RetryInfo createRetryInfo() {
+    public RetryInfo createRetryInfo() {
         Instant now = Instant.now();
         // Is the last time we got called longer ago than the next delay?
         if (lastTimeCalled.getAndSet(now).isBefore(now.minus(nextDelay.get()))) {

data-prepper-plugins/armeria-common/src/test/java/org/opensearch/dataprepper/GrpcRetryInfoCalculatorTest.java

@@ -12,16 +12,16 @@ public class GrpcRetryInfoCalculatorTest {
 
     @Test
     public void testMinimumDelayOnFirstCall() {
-        RetryInfo retryInfo = new GrpcRetryInfoCalculator(Duration.ofMillis(100), Duration.ofSeconds(1)).createRetryInfo();
+        RetryInfo retryInfo = new RetryInfoCalculator(Duration.ofMillis(100), Duration.ofSeconds(1)).createRetryInfo();
 
         assertThat(retryInfo.getRetryDelay().getNanos(), equalTo(100_000_000));
         assertThat(retryInfo.getRetryDelay().getSeconds(), equalTo(0L));
     }
 
     @Test
     public void testExponentialBackoff() {
-        GrpcRetryInfoCalculator calculator =
-                new GrpcRetryInfoCalculator(Duration.ofSeconds(1), Duration.ofSeconds(10));
+        RetryInfoCalculator calculator =
+                new RetryInfoCalculator(Duration.ofSeconds(1), Duration.ofSeconds(10));
         RetryInfo retryInfo1 = calculator.createRetryInfo();
         RetryInfo retryInfo2 = calculator.createRetryInfo();
         RetryInfo retryInfo3 = calculator.createRetryInfo();
@@ -35,8 +35,8 @@ public void testExponentialBackoff() {
 
     @Test
     public void testUsesMaximumAsLongestDelay() {
-        GrpcRetryInfoCalculator calculator =
-                new GrpcRetryInfoCalculator(Duration.ofSeconds(1), Duration.ofSeconds(2));
+        RetryInfoCalculator calculator =
+                new RetryInfoCalculator(Duration.ofSeconds(1), Duration.ofSeconds(2));
         RetryInfo retryInfo1 = calculator.createRetryInfo();
         RetryInfo retryInfo2 = calculator.createRetryInfo();
         RetryInfo retryInfo3 = calculator.createRetryInfo();
@@ -49,8 +49,8 @@ public void testUsesMaximumAsLongestDelay() {
     @Test
     public void testResetAfterDelayWearsOff() throws InterruptedException {
         int minDelayNanos = 1_000_000;
-        GrpcRetryInfoCalculator calculator =
-                new GrpcRetryInfoCalculator(Duration.ofNanos(minDelayNanos), Duration.ofSeconds(1));
+        RetryInfoCalculator calculator =
+                new RetryInfoCalculator(Duration.ofNanos(minDelayNanos), Duration.ofSeconds(1));
 
         RetryInfo retryInfo1 = calculator.createRetryInfo();
         RetryInfo retryInfo2 = calculator.createRetryInfo();
@@ -66,8 +66,8 @@ public void testResetAfterDelayWearsOff() throws InterruptedException {
 
     @Test
     public void testQuickFirstExceptionDoesNotTriggerBackoffCalculationEvenWithLongMinDelay() throws InterruptedException {
-        GrpcRetryInfoCalculator calculator =
-                new GrpcRetryInfoCalculator(Duration.ofSeconds(10), Duration.ofSeconds(20));
+        RetryInfoCalculator calculator =
+                new RetryInfoCalculator(Duration.ofSeconds(10), Duration.ofSeconds(20));
 
         RetryInfo retryInfo1 = calculator.createRetryInfo();
         RetryInfo retryInfo2 = calculator.createRetryInfo();

data-prepper-plugins/kafka-plugins/src/integrationTest/java/org/opensearch/dataprepper/plugins/kafka/buffer/KafkaBufferOTelIT.java

@@ -45,8 +45,6 @@
 import io.opentelemetry.proto.resource.v1.Resource;
 import io.opentelemetry.proto.trace.v1.ScopeSpans;
 import io.opentelemetry.proto.trace.v1.ResourceSpans;
-import io.opentelemetry.proto.trace.v1.InstrumentationLibrarySpans;
-import io.opentelemetry.proto.common.v1.InstrumentationLibrary;
 import io.opentelemetry.proto.common.v1.InstrumentationScope;
 import io.opentelemetry.proto.metrics.v1.Gauge;
 import io.opentelemetry.proto.metrics.v1.Sum;
@@ -407,24 +405,9 @@ private ExportTraceServiceRequest createExportTraceRequest() {
                             .build())
                 .build();
 
-        final InstrumentationLibrarySpans ilSpans = InstrumentationLibrarySpans.newBuilder()
-                .setInstrumentationLibrary(InstrumentationLibrary.newBuilder()
-                        .setName(ilName)
-                        .setVersion(ilVersion)
-                        .build())
-                .addSpans(io.opentelemetry.proto.trace.v1.Span.newBuilder()
-                            .setTraceId(ByteString.copyFrom(TraceId2.getBytes()))
-                            .setSpanId(ByteString.copyFrom(SpanId2.getBytes()))
-                            .setKind(io.opentelemetry.proto.trace.v1.Span.SpanKind.SPAN_KIND_INTERNAL)
-                            .setName(ilSpanName)
-                            .setStartTimeUnixNano(currentUnixTimeNano)
-                            .setEndTimeUnixNano(currentUnixTimeNano+TIME_DELTA*1000_000_000)
-                            .build())
-                .build();
         ResourceSpans resourceSpans = ResourceSpans.newBuilder()
                 .setResource(resource)
                 .addScopeSpans(scopeSpans)
-                .addInstrumentationLibrarySpans(ilSpans)
                 .build();
 
         return ExportTraceServiceRequest.newBuilder()

data-prepper-plugins/otel-logs-source/src/test/java/org/opensearch/dataprepper/plugins/source/otellogs/OTelLogsGrpcServiceTest.java

@@ -12,9 +12,10 @@
 import io.micrometer.core.instrument.Timer;
 import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
 import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceResponse;
-import io.opentelemetry.proto.logs.v1.InstrumentationLibraryLogs;
 import io.opentelemetry.proto.logs.v1.LogRecord;
 import io.opentelemetry.proto.logs.v1.ResourceLogs;
+import io.opentelemetry.proto.logs.v1.ScopeLogs;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -61,9 +62,8 @@
 public class OTelLogsGrpcServiceTest {
     private static final ExportLogsServiceRequest LOGS_REQUEST = ExportLogsServiceRequest.newBuilder()
             .addResourceLogs(ResourceLogs.newBuilder()
-                    .addInstrumentationLibraryLogs(InstrumentationLibraryLogs.newBuilder()
-                            .addLogRecords(LogRecord.newBuilder())
-                    .build())).build();
+                    .addScopeLogs(ScopeLogs.newBuilder().addLogRecords(LogRecord.newBuilder()) .build()))
+            .build();
 
     private static PluginSetting pluginSetting;
     private final int bufferWriteTimeoutInMillis = 100000;

data-prepper-plugins/otel-metrics-raw-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/otelmetrics/OTelMetricsProtoHelper.java

@@ -8,30 +8,21 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.opentelemetry.proto.common.v1.AnyValue;
-import io.opentelemetry.proto.common.v1.InstrumentationLibrary;
-import io.opentelemetry.proto.common.v1.InstrumentationScope;
 import io.opentelemetry.proto.common.v1.KeyValue;
 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.resource.v1.Resource;
 import org.apache.commons.codec.binary.Hex;
 import org.opensearch.dataprepper.model.metric.Bucket;
 import org.opensearch.dataprepper.model.metric.DefaultBucket;
 import org.opensearch.dataprepper.model.metric.DefaultExemplar;
-import org.opensearch.dataprepper.model.metric.DefaultQuantile;
 import org.opensearch.dataprepper.model.metric.Exemplar;
-import org.opensearch.dataprepper.model.metric.Quantile;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -40,14 +31,7 @@ public final class OTelMetricsProtoHelper {
 
     private static final Logger LOG = LoggerFactory.getLogger(OTelMetricsProtoHelper.class);
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-    private static final String SERVICE_NAME = "service.name";
-    private static final String METRIC_ATTRIBUTES = "metric.attributes";
-    static final String RESOURCE_ATTRIBUTES = "resource.attributes";
     static final String EXEMPLAR_ATTRIBUTES = "exemplar.attributes";
-    static final String INSTRUMENTATION_LIBRARY_NAME = "instrumentationLibrary.name";
-    static final String INSTRUMENTATION_LIBRARY_VERSION = "instrumentationLibrary.version";
-    static final String INSTRUMENTATION_SCOPE_NAME = "instrumentationScope.name";
-    static final String INSTRUMENTATION_SCOPE_VERSION = "instrumentationScope.version";
 
     /**
      * To make it ES friendly we will replace '.' in keys with '@' in all the Keys in {@link io.opentelemetry.proto.common.v1.KeyValue}
@@ -60,8 +44,6 @@ public final class OTelMetricsProtoHelper {
      * Span and Resource attributes are essential for kibana so they should not be nested. SO we will prefix them with "metric.attributes"
      * and "resource.attributes" and "exemplar.attributes".
      */
-    public static final Function<String, String> PREFIX_AND_METRIC_ATTRIBUTES_REPLACE_DOT_WITH_AT = i -> METRIC_ATTRIBUTES + DOT + i.replace(DOT, AT);
-    public static final Function<String, String> PREFIX_AND_RESOURCE_ATTRIBUTES_REPLACE_DOT_WITH_AT = i -> RESOURCE_ATTRIBUTES + DOT + i.replace(DOT, AT);
     public static final Function<String, String> PREFIX_AND_EXEMPLAR_ATTRIBUTES_REPLACE_DOT_WITH_AT = i -> EXEMPLAR_ATTRIBUTES + DOT + i.replace(DOT, AT);
 
     private OTelMetricsProtoHelper() {
@@ -111,30 +93,6 @@ public static Object convertAnyValue(final AnyValue value) {
         }
     }
 
-    /**
-     * Converts the keys of all attributes in the {@link NumberDataPoint}.
-     * Also, casts the underlying data into its actual type
-     *
-     * @param numberDataPoint The point to process
-     * @return A Map containing all attributes of `numberDataPoint` with keys converted into an OS-friendly format
-     */
-    public static Map<String, Object> convertKeysOfDataPointAttributes(final NumberDataPoint numberDataPoint) {
-        return numberDataPoint.getAttributesList().stream()
-                .collect(Collectors.toMap(i -> PREFIX_AND_METRIC_ATTRIBUTES_REPLACE_DOT_WITH_AT.apply(i.getKey()), i -> convertAnyValue(i.getValue())));
-    }
-
-    /**
-     * Unpacks the List of {@link KeyValue} object into a Map.
-     * Converts the keys into an os friendly format and casts the underlying data into its actual type?
-     *
-     * @param attributesList The list of {@link KeyValue} objects to process
-     * @return A Map containing unpacked {@link KeyValue} data
-     */
-    public static Map<String, Object> unpackKeyValueList(List<KeyValue> attributesList) {
-        return attributesList.stream()
-                .collect(Collectors.toMap(i -> PREFIX_AND_METRIC_ATTRIBUTES_REPLACE_DOT_WITH_AT.apply(i.getKey()), i -> convertAnyValue(i.getValue())));
-    }
-
     /**
      * Unpacks the List of {@link KeyValue} object into a Map.
      * Converts the keys into an os friendly format and casts the underlying data into its actual type?
@@ -184,79 +142,11 @@ public static Double getExemplarValueAsDouble(final io.opentelemetry.proto.metri
         }
     }
 
-    public static Map<String, Object> getResourceAttributes(final Resource resource) {
-        return resource.getAttributesList().stream()
-                .collect(Collectors.toMap(i -> PREFIX_AND_RESOURCE_ATTRIBUTES_REPLACE_DOT_WITH_AT.apply(i.getKey()), i -> convertAnyValue(i.getValue())));
-    }
-
-    /**
-     * Extracts the name and version of the used instrumentation library used
-     *
-     * @param instrumentationLibrary instrumentationLibrary
-     * @return A map, containing information about the instrumentation library
-     */
-    public static Map<String, Object> getInstrumentationLibraryAttributes(final InstrumentationLibrary instrumentationLibrary) {
-        final Map<String, Object> instrumentationAttr = new HashMap<>();
-        if (!instrumentationLibrary.getName().isEmpty()) {
-            instrumentationAttr.put(INSTRUMENTATION_LIBRARY_NAME, instrumentationLibrary.getName());
-        }
-        if (!instrumentationLibrary.getVersion().isEmpty()) {
-            instrumentationAttr.put(INSTRUMENTATION_LIBRARY_VERSION, instrumentationLibrary.getVersion());
-        }
-        return instrumentationAttr;
-    }
-
-    /**
-     * Extracts the name and version of the used instrumentation scope used
-     *
-     * @param instrumentationScope instrumentationScope
-     * @return A map, containing information about the instrumentation scope
-     */
-    public static Map<String, Object> getInstrumentationScopeAttributes(final InstrumentationScope instrumentationScope) {
-        final Map<String, Object> instrumentationScopeAttr = new HashMap<>();
-        if (!instrumentationScope.getName().isEmpty()) {
-            instrumentationScopeAttr.put(INSTRUMENTATION_SCOPE_NAME, instrumentationScope.getName());
-        }
-        if (!instrumentationScope.getVersion().isEmpty()) {
-            instrumentationScopeAttr.put(INSTRUMENTATION_SCOPE_VERSION, instrumentationScope.getVersion());
-        }
-        return instrumentationScopeAttr;
-    }
-
 
     public static String convertUnixNanosToISO8601(final long unixNano) {
         return Instant.ofEpochSecond(0L, unixNano).toString();
     }
 
-    public static String getStartTimeISO8601(final NumberDataPoint numberDataPoint) {
-        return convertUnixNanosToISO8601(numberDataPoint.getStartTimeUnixNano());
-    }
-
-    public static String getTimeISO8601(final NumberDataPoint ndp) {
-        return convertUnixNanosToISO8601(ndp.getTimeUnixNano());
-    }
-
-    public static Optional<String> getServiceName(final Resource resource) {
-        return resource.getAttributesList().stream()
-                .filter(keyValue -> keyValue.getKey().equals(SERVICE_NAME) && !keyValue.getValue().getStringValue().isEmpty())
-                .findFirst()
-                .map(i -> i.getValue().getStringValue());
-    }
-
-
-    public static Map<String, Object> mergeAllAttributes(final Collection<Map<String, Object>> attributes) {
-        return attributes.stream()
-                .flatMap(map -> map.entrySet().stream())
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-    }
-
-
-    public static List<Quantile> getQuantileValues(List<SummaryDataPoint.ValueAtQuantile> quantileValues) {
-        return quantileValues.stream()
-                .map(q -> new DefaultQuantile(q.getQuantile(), q.getValue()))
-                .collect(Collectors.toList());
-    }
-
     /**
      * Create the buckets, see <a href="https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/metrics/v1/metrics.proto">
      *     the OTel metrics proto spec</a>

data-prepper-plugins/otel-metrics-raw-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/otelmetrics/MetricsPluginExponentialHistogramTest.java

@@ -12,8 +12,8 @@
 import io.opentelemetry.proto.common.v1.KeyValue;
 import io.opentelemetry.proto.metrics.v1.ExponentialHistogram;
 import io.opentelemetry.proto.metrics.v1.ExponentialHistogramDataPoint;
-import io.opentelemetry.proto.metrics.v1.InstrumentationLibraryMetrics;
 import io.opentelemetry.proto.metrics.v1.ResourceMetrics;
+import io.opentelemetry.proto.metrics.v1.ScopeMetrics;
 import io.opentelemetry.proto.resource.v1.Resource;
 import org.hamcrest.MatcherAssert;
 import org.hamcrest.Matchers;
@@ -139,7 +139,7 @@ private ExportMetricsServiceRequest fillServiceRequest(ExponentialHistogram hist
                 .setName("name")
                 .setDescription("description")
                 .build();
-        InstrumentationLibraryMetrics instLib = InstrumentationLibraryMetrics.newBuilder()
+        ScopeMetrics scopeMetrics = ScopeMetrics.newBuilder()
                 .addMetrics(metric).build();
 
         Resource resource = Resource.newBuilder()
@@ -149,7 +149,7 @@ private ExportMetricsServiceRequest fillServiceRequest(ExponentialHistogram hist
                 ).build();
         ResourceMetrics resourceMetrics = ResourceMetrics.newBuilder()
                 .setResource(resource)
-                .addInstrumentationLibraryMetrics(instLib)
+                .addScopeMetrics(scopeMetrics)
                 .build();
         return ExportMetricsServiceRequest.newBuilder().addResourceMetrics(resourceMetrics).build();
     }