processor/otel: record array resource attributes (#5704) (#5715)

* processor/otel: record array resource attributes

If an unhandled array attribute is found in resource
attributes, store that as a label like we do with span
attributes.

* processor/otel: handle array attrs for spans

(cherry picked from commit 4264a2a)

Co-authored-by: Andrew Wilkins <[email protected]>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

processor/otel/consumer.go

@@ -270,22 +270,7 @@ func translateTransaction(
 		k := replaceDots(kDots)
 		switch v.Type() {
 		case pdata.AttributeValueTypeArray:
-			array := v.ArrayVal()
-			values := make([]interface{}, array.Len())
-			for i := range values {
-				value := array.At(i)
-				switch value.Type() {
-				case pdata.AttributeValueTypeBool:
-					values[i] = value.BoolVal()
-				case pdata.AttributeValueTypeDouble:
-					values[i] = value.DoubleVal()
-				case pdata.AttributeValueTypeInt:
-					values[i] = value.IntVal()
-				case pdata.AttributeValueTypeString:
-					values[i] = truncate(value.StringVal())
-				}
-			}
-			labels[k] = values
+			labels[k] = ifaceAnyValueArray(v.ArrayVal())
 		case pdata.AttributeValueTypeBool:
 			labels[k] = v.BoolVal()
 		case pdata.AttributeValueTypeDouble:
@@ -508,6 +493,8 @@ func translateSpan(span pdata.Span, metadata model.Metadata, event *model.Span)
 
 		k := replaceDots(kDots)
 		switch v.Type() {
+		case pdata.AttributeValueTypeArray:
+			labels[k] = ifaceAnyValueArray(v.ArrayVal())
 		case pdata.AttributeValueTypeBool:
 			labels[k] = v.BoolVal()
 		case pdata.AttributeValueTypeDouble:

processor/otel/consumer_test.go

@@ -599,6 +599,15 @@ func TestArrayLabels(t *testing.T) {
 		"bool_array":   []interface{}{false, true},
 		"string_array": []interface{}{"string1", "string2"},
 	}, tx.Labels)
+
+	span := transformSpanWithAttributes(t, map[string]pdata.AttributeValue{
+		"string_array": stringArray,
+		"bool_array":   boolArray,
+	})
+	assert.Equal(t, common.MapStr{
+		"bool_array":   []interface{}{false, true},
+		"string_array": []interface{}{"string1", "string2"},
+	}, span.Labels)
 }
 
 func TestConsumeTracesExportTimestamp(t *testing.T) {

processor/otel/metadata.go

@@ -206,6 +206,16 @@ func ifaceAttributeValue(v pdata.AttributeValue) interface{} {
 		return v.DoubleVal()
 	case pdata.AttributeValueTypeBool:
 		return v.BoolVal()
+	case pdata.AttributeValueTypeArray:
+		return ifaceAnyValueArray(v.ArrayVal())
 	}
 	return nil
 }
+
+func ifaceAnyValueArray(array pdata.AnyValueArray) []interface{} {
+	values := make([]interface{}, array.Len())
+	for i := range values {
+		values[i] = ifaceAttributeValue(array.At(i))
+	}
+	return values
+}

processor/otel/metadata_test.go

@@ -24,6 +24,7 @@ import (
 	"go.opentelemetry.io/collector/consumer/pdata"
 
 	"github.com/elastic/apm-server/model"
+	"github.com/elastic/beats/v7/libbeat/common"
 )
 
 func TestResourceConventions(t *testing.T) {
@@ -201,6 +202,25 @@ func TestResourceConventions(t *testing.T) {
 	}
 }
 
+func TestResourceLabels(t *testing.T) {
+	stringArray := pdata.NewAttributeValueArray()
+	stringArray.ArrayVal().Append(pdata.NewAttributeValueString("abc"))
+	stringArray.ArrayVal().Append(pdata.NewAttributeValueString("def"))
+
+	intArray := pdata.NewAttributeValueArray()
+	intArray.ArrayVal().Append(pdata.NewAttributeValueInt(123))
+	intArray.ArrayVal().Append(pdata.NewAttributeValueInt(456))
+
+	metadata := transformResourceMetadata(t, map[string]pdata.AttributeValue{
+		"string_array": stringArray,
+		"int_array":    intArray,
+	})
+	assert.Equal(t, common.MapStr{
+		"string_array": []interface{}{"abc", "def"},
+		"int_array":    []interface{}{int64(123), int64(456)},
+	}, metadata.Labels)
+}
+
 func transformResourceMetadata(t *testing.T, resourceAttrs map[string]pdata.AttributeValue) model.Metadata {
 	traces, spans := newTracesSpans()
 	traces.ResourceSpans().At(0).Resource().Attributes().InitFromMap(resourceAttrs)

systemtest/approvals/TestOTLPGRPCTraces.approved.json

@@ -13,6 +13,17 @@
                 "ingested": "dynamic",
                 "outcome": "success"
             },
+            "labels": {
+                "resource_attribute_array": [
+                    "a",
+                    "b"
+                ],
+                "span_attribute_array": [
+                    "a",
+                    "b",
+                    "c"
+                ]
+            },
             "observer": {
                 "ephemeral_id": "dynamic",
                 "hostname": "dynamic",

systemtest/otlp_test.go

@@ -35,6 +35,7 @@ import (
 	controller "go.opentelemetry.io/otel/sdk/metric/controller/basic"
 	processor "go.opentelemetry.io/otel/sdk/metric/processor/basic"
 	"go.opentelemetry.io/otel/sdk/metric/selector/simple"
+	"go.opentelemetry.io/otel/sdk/resource"
 	sdkresource "go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/trace"
@@ -68,7 +69,12 @@ func TestOTLPGRPCTraces(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	err := sendOTLPTrace(ctx, newOTLPTracerProvider(newOTLPExporter(t, srv)))
+
+	err := sendOTLPTrace(ctx, newOTLPTracerProvider(newOTLPExporter(t, srv), sdktrace.WithResource(
+		resource.Merge(resource.Default(), sdkresource.NewWithAttributes(
+			attribute.Array("resource_attribute_array", []string{"a", "b"}),
+		)),
+	)))
 	require.NoError(t, err)
 
 	result := systemtest.Elasticsearch.ExpectDocs(t, "apm-*", estest.BoolQuery{Filter: []interface{}{
@@ -322,7 +328,9 @@ func sendOTLPTrace(ctx context.Context, tracerProvider *sdktrace.TracerProvider)
 	tracer := tracerProvider.Tracer("systemtest")
 	startTime := time.Unix(123, 456)
 	endTime := startTime.Add(time.Second)
-	_, span := tracer.Start(ctx, "operation_name", trace.WithTimestamp(startTime))
+	_, span := tracer.Start(ctx, "operation_name", trace.WithTimestamp(startTime), trace.WithAttributes(
+		attribute.Array("span_attribute_array", []string{"a", "b", "c"}),
+	))
 	span.End(trace.WithTimestamp(endTime))
 	return flushTracerProvider(ctx, tracerProvider)
 }