feat: migrate decoders to protobuf (#63)

* feat: migrate decoders to protobuf

* refactor: reduce diff noise

* build: update missing licenses

* fix: add nil check to avoid nil dereference discovered by fuzz testing

* test: remove complex type from populator and skip compression strategy in span test

* test: ignore non ndjson files in jsonschema test

this was conflicting with fuzz testing

* lint: remove unused variables

* test: add compression strategy test

Author: kruskall

input/elasticapm/internal/modeldecoder/modeldecodertest/populator.go

@@ -31,7 +31,6 @@ import (
 
 	"github.com/elastic/apm-data/input/elasticapm/internal/modeldecoder/nullable"
 	"github.com/elastic/apm-data/model"
-	"github.com/elastic/apm-data/model/modelpb"
 )
 
 // Values used for populating the model structs
@@ -46,7 +45,6 @@ type Values struct {
 	HTTPHeader      http.Header
 	LabelVal        model.LabelValue
 	NumericLabelVal model.NumericLabelValue
-	MetricType      modelpb.MetricType
 	// N controls how many elements are added to a slice or a map
 	N int
 }
@@ -328,8 +326,6 @@ func AssertStructValues(t *testing.T, i interface{}, isException func(string) bo
 			newVal = timestamppb.New(values.Time)
 		case time.Duration:
 			newVal = values.Duration
-		case modelpb.MetricType:
-			newVal = values
 		default:
 			// the populator recursively iterates over struct and structPtr
 			// calling this function for all fields;
@@ -346,7 +342,7 @@ func AssertStructValues(t *testing.T, i interface{}, isException func(string) bo
 				assert.NotZero(t, fVal, key)
 				return
 			}
-			panic(fmt.Sprintf("unhandled type %s for key %s", f.Type(), key))
+			panic(fmt.Sprintf("unhandled type %s %s for key %s", f.Kind(), f.Type(), key))
 		}
 		assert.Equal(t, newVal, fVal, key)
 	})

input/elasticapm/internal/modeldecoder/modeldecoderutil/http.go

@@ -20,6 +20,9 @@ package modeldecoderutil
 import (
 	"encoding/json"
 	"net/http"
+
+	"github.com/elastic/apm-data/model/modelpb"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 // HTTPHeadersToMap converts h to a map[string]any, suitable for
@@ -39,6 +42,39 @@ func HTTPHeadersToMap(h http.Header) map[string]any {
 	return m
 }
 
+// HTTPHeadersToStructPb converts h to a *structpb.Struct, suitable for
+// use in modelpb.HTTP.{Request,Response}.Headers.
+func HTTPHeadersToStructPb(h http.Header) *structpb.Struct {
+	if len(h) == 0 {
+		return nil
+	}
+	m := make(map[string]any, len(h))
+	for k, v := range h {
+		arr := make([]any, 0, len(v))
+		for _, s := range v {
+			arr = append(arr, s)
+		}
+		m[k] = arr
+	}
+	if str, err := structpb.NewStruct(m); err == nil {
+		return str
+	}
+	return nil
+}
+
+func HTTPHeadersToModelpb(h http.Header) map[string]*modelpb.HTTPHeaderValue {
+	if len(h) == 0 {
+		return nil
+	}
+	m := make(map[string]*modelpb.HTTPHeaderValue, len(h))
+	for k, v := range h {
+		m[k] = &modelpb.HTTPHeaderValue{
+			Values: v,
+		}
+	}
+	return m
+}
+
 // NormalizeHTTPRequestBody recurses through v, replacing any instance of
 // a json.Number with float64.
 //

input/elasticapm/internal/modeldecoder/modeldecoderutil/labels.go

@@ -22,14 +22,31 @@ import (
 	"strconv"
 
 	"github.com/elastic/apm-data/model"
+	"github.com/elastic/apm-data/model/modelpb"
 )
 
 // GlobalLabelsFrom populates the Labels and NumericLabels from global labels
 // in the metadata object.
-func GlobalLabelsFrom(from map[string]any, to *model.APMEvent) {
+func GlobalLabelsFrom(from map[string]any, to *modelpb.APMEvent) {
+	to.NumericLabels = make(modelpb.NumericLabels)
+	to.Labels = make(modelpb.Labels)
+	MergeLabels(from, to)
+	for k, v := range to.Labels {
+		v.Global = true
+		to.Labels[k] = v
+	}
+	for k, v := range to.NumericLabels {
+		v.Global = true
+		to.NumericLabels[k] = v
+	}
+}
+
+// GlobalLabelsFrom populates the Labels and NumericLabels from global labels
+// in the metadata object.
+func GlobalLabelsFromOld(from map[string]any, to *model.APMEvent) {
 	to.NumericLabels = make(model.NumericLabels)
 	to.Labels = make(model.Labels)
-	MergeLabels(from, to)
+	MergeLabelsOld(from, to)
 	for k, v := range to.Labels {
 		v.Global = true
 		to.Labels[k] = v
@@ -44,7 +61,7 @@ func GlobalLabelsFrom(from map[string]any, to *model.APMEvent) {
 // combining event-specific labels onto (metadata) global labels.
 //
 // If eventLabels is non-nil, it is first cloned.
-func MergeLabels(eventLabels map[string]any, to *model.APMEvent) {
+func MergeLabelsOld(eventLabels map[string]any, to *model.APMEvent) {
 	if to.NumericLabels == nil {
 		to.NumericLabels = make(model.NumericLabels)
 	}
@@ -54,14 +71,47 @@ func MergeLabels(eventLabels map[string]any, to *model.APMEvent) {
 	for k, v := range eventLabels {
 		switch v := v.(type) {
 		case string:
-			to.Labels.Set(k, v)
+			model.Labels(to.Labels).Set(k, v)
+		case bool:
+			model.Labels(to.Labels).Set(k, strconv.FormatBool(v))
+		case float64:
+			model.NumericLabels(to.NumericLabels).Set(k, v)
+		case json.Number:
+			if floatVal, err := v.Float64(); err == nil {
+				model.NumericLabels(to.NumericLabels).Set(k, floatVal)
+			}
+		}
+	}
+	if len(to.NumericLabels) == 0 {
+		to.NumericLabels = nil
+	}
+	if len(to.Labels) == 0 {
+		to.Labels = nil
+	}
+}
+
+// MergeLabels merges eventLabels into the APMEvent. This is used for
+// combining event-specific labels onto (metadata) global labels.
+//
+// If eventLabels is non-nil, it is first cloned.
+func MergeLabels(eventLabels map[string]any, to *modelpb.APMEvent) {
+	if to.NumericLabels == nil {
+		to.NumericLabels = make(modelpb.NumericLabels)
+	}
+	if to.Labels == nil {
+		to.Labels = make(modelpb.Labels)
+	}
+	for k, v := range eventLabels {
+		switch v := v.(type) {
+		case string:
+			modelpb.Labels(to.Labels).Set(k, v)
 		case bool:
-			to.Labels.Set(k, strconv.FormatBool(v))
+			modelpb.Labels(to.Labels).Set(k, strconv.FormatBool(v))
 		case float64:
-			to.NumericLabels.Set(k, v)
+			modelpb.NumericLabels(to.NumericLabels).Set(k, v)
 		case json.Number:
 			if floatVal, err := v.Float64(); err == nil {
-				to.NumericLabels.Set(k, floatVal)
+				modelpb.NumericLabels(to.NumericLabels).Set(k, floatVal)
 			}
 		}
 	}

input/elasticapm/internal/modeldecoder/modeldecoderutil/metrics.go

@@ -20,7 +20,8 @@ package modeldecoderutil
 import (
 	"time"
 
-	"github.com/elastic/apm-data/model"
+	"github.com/elastic/apm-data/model/modelpb"
+	"google.golang.org/protobuf/types/known/durationpb"
 )
 
 // SetInternalMetrics extracts well-known internal metrics from event.Metricset.Samples,
@@ -33,7 +34,7 @@ import (
 // SetInternalMetrics returns true if any known metric samples were found, and false
 // otherwise. If no known metric samples were found, the caller may opt to omit the
 // metricset altogether.
-func SetInternalMetrics(event *model.APMEvent) bool {
+func SetInternalMetrics(event *modelpb.APMEvent) bool {
 	if event.Transaction == nil {
 		// Not an internal metricset.
 		return false
@@ -43,10 +44,16 @@ func SetInternalMetrics(event *model.APMEvent) bool {
 		for _, v := range event.Metricset.Samples {
 			switch v.Name {
 			case "span.self_time.count":
-				event.Span.SelfTime.Count = int(v.Value)
+				if event.Span.SelfTime == nil {
+					event.Span.SelfTime = &modelpb.AggregatedDuration{}
+				}
+				event.Span.SelfTime.Count = int64(v.Value)
 				haveMetrics = true
 			case "span.self_time.sum.us":
-				event.Span.SelfTime.Sum = time.Duration(v.Value * 1000)
+				if event.Span.SelfTime == nil {
+					event.Span.SelfTime = &modelpb.AggregatedDuration{}
+				}
+				event.Span.SelfTime.Sum = durationpb.New(time.Duration(v.Value * 1000))
 				haveMetrics = true
 			}
 		}

input/elasticapm/internal/modeldecoder/modeldecoderutil/structpb.go

@@ -0,0 +1,43 @@
+// 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 modeldecoderutil
+
+import (
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func ToStruct(m map[string]any) *structpb.Struct {
+	if str, err := structpb.NewStruct(m); err == nil {
+		return str
+	}
+	return nil
+}
+
+func ToValue(a any) *structpb.Value {
+	if v, err := structpb.NewValue(a); err == nil {
+		return v
+	}
+	return nil
+}
+
+func NormalizeMap(m map[string]any) map[string]any {
+	if v := NormalizeHTTPRequestBody(m); v != nil {
+		return v.(map[string]any)
+	}
+	return nil
+}