[OTel] Support Serializing nested data (#45008)

This adds support for `map[string]any` and also arrays in general
(as opposed to just slices) to the converted map.

Like `mapstr.M` and `[]mapstr.M`, this will dive into the
`map[string]any` and `[]map[string]any` objects to ensure that
they are properly converted for the pdata code.

Author: pickypg

CHANGELOG-developer.next.asciidoc

@@ -75,6 +75,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 
 ==== Bugfixes
 
+- Handle conversion of `map[string]any` and `[]map[string]any` for `otelconsumer`.{pull}45008[45008]
 - Handle the starting of namespace and node watchers for metadata enrichment according to `add_resource_metadata` configuration.{pull}38762[38762]
 - Fix multiple metricbeat instances reporting same metrics when using autodiscover with provider kubernetes, and ensure leader elector is always running in autodiscover mode.{pull}38471[38471]
 - Fix how Prometheus histograms are calculated when percentiles are provide.{pull}36537[36537]

libbeat/otelbeat/otelmap/otelmap.go

@@ -29,6 +29,11 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 )
 
+// Allow ConvertNonPrimitive to be called recursively to handle nested maps of either type.
+type mapstrOrMap interface {
+	mapstr.M | map[string]any
+}
+
 // ToMapstr converts a [pcommon.Map] to a [mapstr.M].
 func ToMapstr(m pcommon.Map) mapstr.M {
 	return m.AsRaw()
@@ -42,7 +47,7 @@ func ToMapstr(m pcommon.Map) mapstr.M {
 //     If you attempt to use other slice types (e.g., []string or []int),
 //     pcommon.Map.FromRaw(...) will return an "invalid type" error.
 //     To overcome this, we use "reflect" to transform []T into []any.
-func ConvertNonPrimitive(m mapstr.M) {
+func ConvertNonPrimitive[T mapstrOrMap](m T) {
 	for key, val := range m {
 		switch x := val.(type) {
 		case mapstr.M:
@@ -55,6 +60,16 @@ func ConvertNonPrimitive(m mapstr.M) {
 				s[i] = map[string]any(val)
 			}
 			m[key] = s
+		case map[string]any:
+			ConvertNonPrimitive(x)
+			m[key] = x
+		case []map[string]any:
+			s := make([]any, len(x))
+			for i := range x {
+				ConvertNonPrimitive(x[i])
+				s[i] = x[i]
+			}
+			m[key] = s
 		case time.Time:
 			m[key] = x.UTC().Format("2006-01-02T15:04:05.000Z")
 		case common.Time:
@@ -73,16 +88,30 @@ func ConvertNonPrimitive(m mapstr.M) {
 			m[key] = s
 		case []bool, []string, []float32, []float64, []int, []int8, []int16, []int32, []int64,
 			[]uint, []uint8, []uint16, []uint32, []uint64:
+			ref := reflect.ValueOf(x)
+			s := make([]any, ref.Len())
+			for i := 0; i < ref.Len(); i++ {
+				s[i] = ref.Index(i).Interface()
+			}
+			m[key] = s
+		case nil, string, int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64, bool:
+		default:
 			ref := reflect.ValueOf(x)
 			if ref.Kind() == reflect.Slice || ref.Kind() == reflect.Array {
-				slice := make([]any, ref.Len())
+				s := make([]any, ref.Len())
 				for i := 0; i < ref.Len(); i++ {
-					slice[i] = ref.Index(i).Interface()
+					elem := ref.Index(i)
+					if elem.Kind() == reflect.Map && elem.Type().Key().Kind() == reflect.String && elem.Type().Elem().Kind() == reflect.Interface {
+						if m, ok := elem.Interface().(map[string]any); ok {
+							ConvertNonPrimitive(m)
+						}
+					}
+					s[i] = elem.Interface()
 				}
-				m[key] = slice
+				m[key] = s
+				break // we figured out the type, so we don't need the unknown type case
 			}
-		case nil, string, int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64, float32, float64, bool, []any, map[string]any:
-		default:
+
 			m[key] = fmt.Sprintf("unknown type: %T", x)
 		}
 	}

libbeat/otelbeat/otelmap/otelmap_test.go

@@ -273,6 +273,56 @@ func TestFromMapstrSliceCommonTime(t *testing.T) {
 	assert.Equal(t, want, inputMap)
 }
 
+func TestFromMapstrWithNestedData(t *testing.T) {
+	input := mapstr.M{
+		"any_array":  [3]any{1, "string", 3},
+		"any_slice":  []any{5.1, 6.2},
+		"bool_array": [2]bool{true, false},
+		"bool_slice": []bool{false, true},
+		"inner": []mapstr.M{
+			{
+				"inner_int": 42,
+				"inner_slice": []map[string]any{ // slice -> slice
+					{"string": "string"},
+					{"number": 12.3},
+				},
+			},
+			{
+				"inner_int": 43,
+				"inner_slice": [2]map[string]any{ // array -> slice
+					{"string": "string2"},
+					{"number": 12.4},
+				},
+			},
+		},
+	}
+	want := mapstr.M{
+		"any_array":  []any{1, "string", 3},
+		"any_slice":  []any{5.1, 6.2},
+		"bool_array": []any{true, false},
+		"bool_slice": []any{false, true},
+		"inner": []any{
+			map[string]any{
+				"inner_int": 42,
+				"inner_slice": []any{
+					map[string]any{"string": "string"},
+					map[string]any{"number": 12.3},
+				},
+			},
+			map[string]any{
+				"inner_int": 43,
+				"inner_slice": []any{
+					map[string]any{"string": "string2"},
+					map[string]any{"number": 12.4},
+				},
+			},
+		},
+	}
+
+	ConvertNonPrimitive(input)
+	assert.Equal(t, want, input)
+}
+
 func TestToMapstr(t *testing.T) {
 	pm := pcommon.NewMap()
 	pm.PutInt("int", 42)
@@ -303,15 +353,25 @@ func TestToMapstr(t *testing.T) {
 	assert.Equal(t, want, got)
 }
 
-type unknown int
+type unknown struct {
+	Value int `json:"value"`
+}
 
 func TestUnknownType(t *testing.T) {
 	inputMap := mapstr.M{
-		"slice": []unknown{42, 43, 44},
+		"unknown": unknown{42},
+		"nested": mapstr.M{
+			"unknown": unknown{43},
+		},
+		"unknown_map": map[string]int{"key": 42},
 	}
 
 	expected := mapstr.M{
-		"slice": "unknown type: []otelmap.unknown",
+		"unknown": "unknown type: otelmap.unknown",
+		"nested": map[string]any{
+			"unknown": "unknown type: otelmap.unknown",
+		},
+		"unknown_map": "unknown type: map[string]int",
 	}
 
 	ConvertNonPrimitive(inputMap)