Add support for Scope metrics

Signed-off-by: Krishna Kondaka <[email protected]>

.github/workflows/data-prepper-trace-analytics-raw-span-e2e-tests.yml

@@ -14,7 +14,7 @@ jobs:
     strategy:
       matrix:
         java: [11, 17, 21, docker]
-        otelVersion: ['1.3.2-alpha']
+        otelVersion: ['0.16.0-alpha', '1.3.2-alpha']
       fail-fast: false
 
     runs-on: ubuntu-latest

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

@@ -99,6 +99,7 @@ public class OTelProtoCodec {
     private static final String EXEMPLAR_ATTRIBUTES = "exemplar.attributes";
     static final String INSTRUMENTATION_SCOPE_NAME = "instrumentationScope.name";
     static final String INSTRUMENTATION_SCOPE_VERSION = "instrumentationScope.version";
+    static final String INSTRUMENTATION_SCOPE_ATTRIBUTES = "instrumentationScope.attributes";
 
     public static final Function<String, String> REPLACE_DOT_WITH_AT = i -> i.replace(DOT, AT);
     /**
@@ -331,7 +332,7 @@ protected List<OpenTelemetryLog> processLogsList(final List<LogRecord> logsList,
         }
 
         protected Span parseSpan(final io.opentelemetry.proto.trace.v1.Span sp, final Map<String, Object> instrumentationScopeAttributes,
-                                     final String serviceName, final Map<String, Object> resourceAttributes, final Instant timeReceived) {
+                                    final String serviceName, final Map<String, Object> resourceAttributes, final Instant timeReceived) {
             return JacksonSpan.builder()
                     .withSpanId(convertByteStringToString(sp.getSpanId()))
                     .withTraceId(convertByteStringToString(sp.getTraceId()))
@@ -653,7 +654,7 @@ private List<? extends Record<? extends Metric>> mapSummary(
                         .withQuantilesValueCount(dp.getQuantileValuesCount())
                         .withAttributes(OTelProtoCodec.mergeAllAttributes(
                                 Arrays.asList(
-                                        OTelProtoCodec.unpackKeyValueList(dp.getAttributesList()),
+                                        OTelProtoCodec.unpackKeyValueListMetric(dp.getAttributesList()),
                                         resourceAttributes,
                                         ils
                                 )
@@ -693,7 +694,7 @@ private List<? extends Record<? extends Metric>> mapHistogram(
                             .withExplicitBoundsList(dp.getExplicitBoundsList())
                             .withAttributes(OTelProtoCodec.mergeAllAttributes(
                                     Arrays.asList(
-                                            OTelProtoCodec.unpackKeyValueList(dp.getAttributesList()),
+                                            OTelProtoCodec.unpackKeyValueListMetric(dp.getAttributesList()),
                                             resourceAttributes,
                                             ils
                                     )
@@ -754,7 +755,7 @@ private List<? extends Record<? extends Metric>> mapExponentialHistogram(
                             .withAggregationTemporality(metric.getHistogram().getAggregationTemporality().toString())
                             .withAttributes(OTelProtoCodec.mergeAllAttributes(
                                     Arrays.asList(
-                                            OTelProtoCodec.unpackKeyValueList(dp.getAttributesList()),
+                                            OTelProtoCodec.unpackKeyValueListMetric(dp.getAttributesList()),
                                             resourceAttributes,
                                             ils
                                     )
@@ -843,12 +844,20 @@ protected List<KeyValue> getResourceAttributes(final Map<String, Object> attribu
             return result;
         }
 
-        protected InstrumentationScope constructInstrumentationScope(final Map<String, Object> attributes) {
+        protected InstrumentationScope constructInstrumentationScope(final Map<String, Object> attributes) throws UnsupportedEncodingException, DecoderException {
             final InstrumentationScope.Builder builder = InstrumentationScope.newBuilder();
-            final Optional<String> instrumentationScopeName = Optional.ofNullable((String) attributes.get(INSTRUMENTATION_SCOPE_NAME));
-            final Optional<String> instrumentationScopeVersion = Optional.ofNullable((String) attributes.get(INSTRUMENTATION_SCOPE_VERSION));
-            instrumentationScopeName.ifPresent(builder::setName);
-            instrumentationScopeVersion.ifPresent(builder::setVersion);
+            final List<KeyValue> attributeKeyValueList = new ArrayList<>();
+            for (Map.Entry<String, Object> entry : attributes.entrySet()) {
+                if (entry.getKey().equals(INSTRUMENTATION_SCOPE_NAME)) {
+                    builder.setName((String)entry.getValue());
+                } else if (entry.getKey().equals(INSTRUMENTATION_SCOPE_VERSION)) {
+                    builder.setVersion((String)entry.getValue());
+                } else if (entry.getKey().startsWith(INSTRUMENTATION_SCOPE_ATTRIBUTES)) {
+                    KeyValue.Builder setValue = KeyValue.newBuilder().setKey(entry.getKey().substring(INSTRUMENTATION_SCOPE_ATTRIBUTES.length()+1)).setValue(objectToAnyValue(entry.getValue()));
+                    attributeKeyValueList.add(setValue.build());
+                }
+            }
+            builder.addAllAttributes(attributeKeyValueList);
             return builder.build();
         }
 
@@ -1013,11 +1022,16 @@ public static Map<String, Object> convertKeysOfDataPointAttributes(final NumberD
      * @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) {
+    public static Map<String, Object> unpackKeyValueListMetric(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())));
     }
 
+    public static Map<String, Object> unpackKeyValueList(List<KeyValue> attributesList) {
+        return attributesList.stream()
+                .collect(Collectors.toMap(i -> DOT+(i.getKey()).replace(DOT, AT), i -> convertAnyValue(i.getValue())));
+    }
+
     /**
      * Unpacks the List of {@link KeyValue} object into a Map. Used for logs.
      * <p>
@@ -1101,6 +1115,11 @@ public static Map<String, Object> getInstrumentationScopeAttributes(final Instru
         if (!instrumentationScope.getVersion().isEmpty()) {
             instrumentationScopeAttr.put(INSTRUMENTATION_SCOPE_VERSION, instrumentationScope.getVersion());
         }
+        if (!instrumentationScope.getAttributesList().isEmpty()) {
+            for (Map.Entry<String, Object> entry: OTelProtoCodec.unpackKeyValueList(instrumentationScope.getAttributesList()).entrySet()) {
+                instrumentationScopeAttr.put(INSTRUMENTATION_SCOPE_ATTRIBUTES+entry.getKey(), entry.getValue());
+            }
+        }
         return instrumentationScopeAttr;
     }
 

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

@@ -36,7 +36,7 @@ public class OTelLogsProtoBufDecoderTest {
     private static final String TEST_REQUEST_LOGS_FILE = "test-otel-log.protobuf";
     // This protobuf format file is generated using OTEL collector and file exporter and then sending multiple log events to the collector
     private static final String TEST_REQUEST_MULTI_LOGS_FILE = "test-otel-multi-log.protobuf";
-    
+
     public OTelLogsProtoBufDecoder createObjectUnderTest(boolean lengthPrefixedEncoding) {
         return new OTelLogsProtoBufDecoder(lengthPrefixedEncoding);
     }
@@ -53,8 +53,8 @@ private void assertLog(OpenTelemetryLog logRecord, final int severityNumber, fin
         assertThat(logRecord.getSeverityText(), is("Information"));
         assertThat(logRecord.getSpanId(), is(spanId));
         assertThat(logRecord.getTraceId(), is("5b8efff798038103d269b633813fc60c"));
-        Map<String, Object> mergedAttributes = logRecord.getAttributes(); 
-        assertThat(mergedAttributes.keySet().size(), is(9)); 
+        Map<String, Object> mergedAttributes = logRecord.getAttributes();
+        assertThat(mergedAttributes.keySet().size(), is(10));
     }
 
     @Test
@@ -63,7 +63,6 @@ public void testParse() throws Exception {
         createObjectUnderTest(false).parse(inputStream, Instant.now(), (record) -> {
             assertLog((OpenTelemetryLog)record.getData(), 50, "2025-01-26T20:07:20Z", "eee19b7ec3c1b174");
         });
-
     }
 
     @Test
@@ -120,10 +119,10 @@ public void testParseWithDynamicRequest() throws Exception {
             assertLogFromRequest((OpenTelemetryLog)record.getData());
         });
     }
-        
+
     @Test
     public void testParseWithLargeDynamicRequest_ThrowsException() throws Exception {
-        
+
         // Create a request larger than 8MB
         List<LogRecord> records = new ArrayList<>();
         for (int i = 0; i < 4 * 1024 * 1024; i++) {

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

@@ -234,6 +234,7 @@ private void validateSpans(final List<Span> spans) {
                 Map<String, Object> attributes = span.getAttributes();
                 assertThat(attributes.containsKey(OTelProtoCodec.RESOURCE_ATTRIBUTES_REPLACE_DOT_WITH_AT.apply("service.name")), is(true));
                 assertThat(attributes.containsKey(OTelProtoCodec.INSTRUMENTATION_SCOPE_NAME), is(true));
+                assertThat(attributes.containsKey(OTelProtoCodec.INSTRUMENTATION_SCOPE_ATTRIBUTES+".my@scope@attribute"), is(true));
                 assertThat(attributes.containsKey(OTelProtoCodec.STATUS_CODE), is(true));
             }
         }
@@ -470,9 +471,12 @@ private void validateLog(OpenTelemetryLog logRecord) {
             assertThat(logRecord.getTraceId(), is("ba1a1c23b4093b63"));
             assertThat(logRecord.getSpanId(), is("2cc83ac90ebc469c"));
             Map<String, Object> mergedAttributes = logRecord.getAttributes();
-            assertThat(mergedAttributes.keySet().size(), is(2));
+            assertThat(mergedAttributes.keySet().size(), is(5));
             assertThat(mergedAttributes.get("log.attributes.statement@params"), is("us-east-1"));
             assertThat(mergedAttributes.get("resource.attributes.service@name"), is("service"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_ATTRIBUTES+".my@scope@attribute"), is("log scope attribute"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_NAME), is("my.library"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_VERSION), is("1.0.0"));
         }
 
         @Test
@@ -517,6 +521,10 @@ private void validateGaugeMetricRequest(Collection<Record<? extends Metric>> met
             assertThat(metric.getName(), equalTo("counter-int"));
             JacksonGauge gauge = (JacksonGauge)metric;
             assertThat(gauge.getValue(), equalTo(123.0));
+            Map<String, Object> mergedAttributes = gauge.getAttributes();
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_ATTRIBUTES+".my@scope@attribute"), is("gauge scope attribute"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_NAME), is("my.library"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_VERSION), is("1.0.0"));
         }
 
         private void validateSumMetricRequest(Collection<Record<? extends Metric>> metrics) {
@@ -528,6 +536,10 @@ private void validateSumMetricRequest(Collection<Record<? extends Metric>> metri
             assertThat(metric.getName(), equalTo("sum-int"));
             JacksonSum sum = (JacksonSum)metric;
             assertThat(sum.getValue(), equalTo(456.0));
+            Map<String, Object> mergedAttributes = sum.getAttributes();
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_ATTRIBUTES+".my@scope@attribute"), is("sum scope attribute"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_NAME), is("my.library"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_VERSION), is("1.0.0"));
         }
 
         private void validateHistogramMetricRequest(Collection<Record<? extends Metric>> metrics) {
@@ -546,6 +558,10 @@ private void validateHistogramMetricRequest(Collection<Record<? extends Metric>>
             assertThat(histogram.getBucketCountsList(), equalTo(List.of(3L, 5L, 15L, 6L, 1L)));
             assertThat(histogram.getBucketCount(), equalTo(5));
             assertThat(histogram.getAggregationTemporality(), equalTo("AGGREGATION_TEMPORALITY_CUMULATIVE"));
+            Map<String, Object> mergedAttributes = histogram.getAttributes();
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_ATTRIBUTES+".my@scope@attribute"), is("histogram scope attribute"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_NAME), is("my.library"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_VERSION), is("1.0.0"));
         }
 
         private void validateHistogramMetricRequestNoExplicitBounds(Collection<Record<? extends Metric>> metrics) {
@@ -564,6 +580,10 @@ private void validateHistogramMetricRequestNoExplicitBounds(Collection<Record<?
             assertThat(histogram.getBucketCountsList(), equalTo(List.of(10L)));
             assertThat(histogram.getBucketCount(), equalTo(1));
             assertThat(histogram.getAggregationTemporality(), equalTo("AGGREGATION_TEMPORALITY_CUMULATIVE"));
+            Map<String, Object> mergedAttributes = histogram.getAttributes();
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_ATTRIBUTES+".my@scope@attribute"), is("histogram scope attribute"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_NAME), is("my.library"));
+            assertThat(mergedAttributes.get(OTelProtoCodec.INSTRUMENTATION_SCOPE_VERSION), is("1.0.0"));
         }
 
 
@@ -687,7 +707,7 @@ public void testEncodeSpanStatusMissingAll() {
         }
 
         @Test
-        public void testEncodeInstrumentationScopeComplete() {
+        public void testEncodeInstrumentationScopeComplete() throws UnsupportedEncodingException, DecoderException {
             final String testName = "test name";
             final String testVersion = "1.1";
             final String testKeyIrrelevant = "irrelevantKey";
@@ -701,7 +721,7 @@ public void testEncodeInstrumentationScopeComplete() {
         }
 
         @Test
-        public void testEncodeInstrumentationScopeMissingName() {
+        public void testEncodeInstrumentationScopeMissingName() throws UnsupportedEncodingException, DecoderException {
             final String testVersion = "1.1";
             final String testKeyIrrelevant = "irrelevantKey";
             final Map<String, Object> testAllAttributes = Map.of(
@@ -712,7 +732,7 @@ public void testEncodeInstrumentationScopeMissingName() {
         }
 
         @Test
-        public void testEncodeInstrumentationScopeMissingVersion() {
+        public void testEncodeInstrumentationScopeMissingVersion() throws UnsupportedEncodingException, DecoderException {
             final String testName = "test name";
             final String testKeyIrrelevant = "irrelevantKey";
             final Map<String, Object> testAllAttributes = Map.of(
@@ -723,7 +743,7 @@ public void testEncodeInstrumentationScopeMissingVersion() {
         }
 
         @Test
-        public void testEncodeInstrumentationScopeMissingAll() {
+        public void testEncodeInstrumentationScopeMissingAll() throws UnsupportedEncodingException, DecoderException {
             final String testKeyIrrelevant = "irrelevantKey";
             final Map<String, Object> testAllAttributes = Map.of(testKeyIrrelevant, 2);
             final InstrumentationScope instrumentationScope = encoderUnderTest.constructInstrumentationScope(testAllAttributes);

data-prepper-plugins/otel-proto-common/src/test/resources/test-gauge-metrics.json

@@ -13,7 +13,18 @@
       },
       "scopeMetrics": [
         {
-          "scope": {},
+          "scope": {
+            "name": "my.library",
+            "version": "1.0.0",
+            "attributes": [
+              {
+                "key": "my.scope.attribute",
+                "value": {
+                  "stringValue": "gauge scope attribute"
+                }
+              }
+            ]
+          },
           "metrics": [
             {
               "name": "counter-int",

data-prepper-plugins/otel-proto-common/src/test/resources/test-histogram-metrics-no-explicit-bounds.json

@@ -13,7 +13,18 @@
       },
       "scopeMetrics": [
         {
-          "scope": {},
+          "scope": {
+            "name": "my.library",
+            "version": "1.0.0",
+            "attributes": [
+              {
+                "key": "my.scope.attribute",
+                "value": {
+                  "stringValue": "histogram scope attribute"
+                }
+              }
+            ]
+          },
           "metrics": [
             {
               "name": "histogram-int",

data-prepper-plugins/otel-proto-common/src/test/resources/test-histogram-metrics.json

@@ -13,7 +13,18 @@
       },
       "scopeMetrics": [
         {
-          "scope": {},
+          "scope": {
+            "name": "my.library",
+            "version": "1.0.0",
+            "attributes": [
+              {
+                "key": "my.scope.attribute",
+                "value": {
+                  "stringValue": "histogram scope attribute"
+                }
+              }
+            ]
+          },
           "metrics": [
             {
               "name": "histogram-int",

data-prepper-plugins/otel-proto-common/src/test/resources/test-request-log.json

@@ -9,6 +9,18 @@
       }]
     },
     "scopeLogs": [{
+      "scope": {
+        "name": "my.library",
+        "version": "1.0.0",
+        "attributes": [
+          {
+            "key": "my.scope.attribute",
+            "value": {
+              "stringValue": "log scope attribute"
+            }
+          }
+        ]
+      },
       "logRecords": [{
         "timeUnixNano": "1590328800000000000",
         "severityNumber": "SEVERITY_NUMBER_DEBUG",
@@ -31,4 +43,4 @@
     }],
     "schemaUrl": "schemaurl"
   }]
-}
+}

data-prepper-plugins/otel-proto-common/src/test/resources/test-request.json

@@ -73,7 +73,15 @@
         {
           "scope": {
             "name": "io.opentelemetry.auto.spring-webmvc-3.1",
-            "version": ""
+            "version": "",
+            "attributes": [
+              {
+                "key": "my.scope.attribute",
+                "value": {
+                  "stringValue": "span scope attribute"
+                }
+              }
+            ]
           },
           "spans": [
             {
@@ -101,7 +109,15 @@
         {
           "scope": {
             "name": "io.opentelemetry.auto.apache-httpasyncclient-4.0",
-            "version": ""
+            "version": "",
+            "attributes": [
+              {
+                "key": "my.scope.attribute",
+                "value": {
+                  "stringValue": "span scope attribute"
+                }
+              }
+            ]
           },
           "spans": [
             {
@@ -148,7 +164,15 @@
         {
           "scope": {
             "name": "io.opentelemetry.auto.servlet-3.0",
-            "version": ""
+            "version": "",
+            "attributes": [
+              {
+                "key": "my.scope.attribute",
+                "value": {
+                  "stringValue": "span scope attribute"
+                }
+              }
+            ]
           },
           "spans": [
             {
@@ -243,4 +267,4 @@
       ]
     }
   ]
-}
+}