[8.18](backport #45008) [OTel] Support Serializing nested data

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

@@ -0,0 +1,118 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package otelmap provides utilities for converting between beats and otel map types.
+package otelmap
+
+import (
+	"fmt"
+	"reflect"
+	"time"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/elastic-agent-libs/mapstr"
+
+	"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()
+}
+
+// ConvertNonPrimitive handles the conversion of non-primitive data types to pcommon-primitive types.
+// The conversion is performed in place.
+// Notes:
+//  1. Slices require special handling when converting a map[string]any to pcommon.Map.
+//     The pcommon.Map expects all slices to be of type []any.
+//     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[T mapstrOrMap](m T) {
+	for key, val := range m {
+		switch x := val.(type) {
+		case mapstr.M:
+			ConvertNonPrimitive(x)
+			m[key] = map[string]any(x)
+		case []mapstr.M:
+			s := make([]any, len(x))
+			for i, val := range x {
+				ConvertNonPrimitive(val)
+				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:
+			m[key] = time.Time(x).UTC().Format("2006-01-02T15:04:05.000Z")
+		case []time.Time:
+			s := make([]any, 0, len(x))
+			for _, i := range x {
+				s = append(s, i.UTC().Format("2006-01-02T15:04:05.000Z"))
+			}
+			m[key] = s
+		case []common.Time:
+			s := make([]any, 0, len(x))
+			for _, i := range x {
+				s = append(s, time.Time(i).UTC().Format("2006-01-02T15:04:05.000Z"))
+			}
+			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 {
+				s := make([]any, ref.Len())
+				for i := 0; i < ref.Len(); i++ {
+					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] = s
+				break // we figured out the type, so we don't need the unknown type case
+			}
+
+			m[key] = fmt.Sprintf("unknown type: %T", x)
+		}
+	}
+}