model: clone labels before sanitizing (#8652)

model/apmevent_test.go

@@ -20,6 +20,7 @@ package model
 import (
 	"context"
 	"net"
+	"sync"
 	"testing"
 	"time"
 
@@ -193,3 +194,25 @@ func TestAPMEventFields(t *testing.T) {
 		assert.Equal(t, test.output, event.Fields)
 	}
 }
+
+func TestAPMEventSharedLabels(t *testing.T) {
+	// Labels will be shared by multiple events in the case where
+	// global labels are sent by an agent, and there are no event-
+	// specific labels. Ensure that transformation can safely be
+	// done in parallel.
+	input := APMEvent{
+		Labels: common.MapStr{
+			"nil_value":       nil,
+			"unsanitized.key": "needs_replacing",
+		},
+	}
+	var g sync.WaitGroup
+	for i := 0; i < 100; i++ {
+		g.Add(1)
+		go func() {
+			defer g.Done()
+			_ = input.BeatEvent(context.Background())
+		}()
+	}
+	g.Wait()
+}

model/labels.go

@@ -24,14 +24,25 @@ import (
 )
 
 // Label keys are sanitized, replacing the reserved characters '.', '*' and '"'
-// with '_'. Null-valued labels are omitted.
+// with '_'. Null-valued labels are omitted. Labels must not be mutated, as it
+// may be shared by multiple events; if sanitization is required, a new map
+// will be returned.
 func sanitizeLabels(labels common.MapStr) common.MapStr {
+	cloned := false
 	for k, v := range labels {
 		if v == nil {
+			if !cloned {
+				labels = labels.Clone()
+				cloned = true
+			}
 			delete(labels, k)
 			continue
 		}
 		if k2 := sanitizeLabelKey(k); k != k2 {
+			if !cloned {
+				labels = labels.Clone()
+				cloned = true
+			}
 			delete(labels, k)
 			labels[k2] = v
 		}