feat(statsdreceiver): use source attributes semantic

receiver/statsdreceiver/config_test.go

@@ -42,6 +42,7 @@ func TestLoadConfig(t *testing.T) {
 					Transport: confignet.TransportTypeUDP6,
 				},
 				AggregationInterval: 70 * time.Second,
+				AggregateBySourceAddr: true,
 				TimerHistogramMapping: []protocol.TimerHistogramMapping{
 					{
 						StatsdType:   "histogram",

receiver/statsdreceiver/internal/protocol/statsd_parser.go

@@ -202,7 +202,16 @@ func (p *StatsDParser) GetMetrics() []BatchMetrics {
 			Metrics: pmetric.NewMetrics(),
 		}
 		rm := batch.Metrics.ResourceMetrics().AppendEmpty()
-		rm.Resource().Attributes().PutStr("source", instrument.addr.String())
+
+		// https://opentelemetry.io/docs/specs/semconv/attributes-registry/source/
+		// Set source.address and source.port attributes if able to parse.
+		if instrument.addr != nil {
+			host, port, err := net.SplitHostPort(instrument.addr.String())
+			if err == nil {
+				rm.Resource().Attributes().PutStr("source.address", host)
+				rm.Resource().Attributes().PutStr("source.port", port)
+			}
+		}
 		for _, metric := range instrument.gauges {
 			p.copyMetricAndScope(rm, metric)
 		}

receiver/statsdreceiver/internal/protocol/statsd_parser_test.go

@@ -5,6 +5,7 @@ package protocol
 
 import (
 	"errors"
+	"fmt"
 	"net"
 	"testing"
 	"time"
@@ -1690,9 +1691,15 @@ func TestStatsDParser_AggregateTimerWithHistogram(t *testing.T) {
 		},
 	}
 
+	sourceAddr := "1.2.3.4"
+	sourcePort := "5678"
+
 	newPoint := func() (pmetric.Metrics, pmetric.ExponentialHistogramDataPoint) {
 		data := pmetric.NewMetrics()
-		ilm := data.ResourceMetrics().AppendEmpty().ScopeMetrics().AppendEmpty()
+		rm := data.ResourceMetrics().AppendEmpty()
+		rm.Resource().Attributes().PutStr("source.address", sourceAddr)
+		rm.Resource().Attributes().PutStr("source.port", sourcePort)
+		ilm := rm.ScopeMetrics().AppendEmpty()
 		m := ilm.Metrics().AppendEmpty()
 		m.SetName("expohisto")
 		ep := m.SetEmptyExponentialHistogram()
@@ -1903,7 +1910,7 @@ func TestStatsDParser_AggregateTimerWithHistogram(t *testing.T) {
 			var err error
 			p := &StatsDParser{}
 			assert.NoError(t, p.Initialize(false, false, false, tt.mapping))
-			addr, _ := net.ResolveUDPAddr("udp", "1.2.3.4:5678")
+			addr, _ := net.ResolveUDPAddr("udp", fmt.Sprintf("%s:%s", sourceAddr, sourcePort))
 			for _, line := range tt.input {
 				err = p.Aggregate(line, addr)
 				assert.NoError(t, err)

receiver/statsdreceiver/metadata.yaml

@@ -10,7 +10,11 @@ status:
     active: [jmacd, dmitryax]
 
 resource_attributes:
-  source:
-    description: Source address of the data
+  source.address:
+    description: Source address
     type: string
     enabled: true
+  source.port:
+    description: Source port number
+    type: int
+    enabled: true

receiver/statsdreceiver/receiver_test.go

@@ -6,6 +6,7 @@ package statsdreceiver
 import (
 	"context"
 	"errors"
+	"net"
 	"testing"
 	"time"
 
@@ -211,15 +212,26 @@ func Test_statsdreceiver_resource_attribute_source(t *testing.T) {
 		require.Equal(t, 1, mdd[0].ResourceMetrics().Len())
 		require.Equal(t, 1, mdd[1].ResourceMetrics().Len())
 
-		// The resources should have source attribute matching the client addr
+		// The resources should have source attributes matching the client addr
 		firstResource := mdd[0].ResourceMetrics().At(0)
-		resourceAttributeSource, exists := firstResource.Resource().Attributes().Get("source")
+		sourceAddress, exists := firstResource.Resource().Attributes().Get("source.address")
 		assert.Equal(t, true, exists)
-		assert.Equal(t, firstStatsdClient.Conn.LocalAddr().String(), resourceAttributeSource.AsString())
+		sourcePort, exists := firstResource.Resource().Attributes().Get("source.port")
+		assert.Equal(t, true, exists)
+		clientAddress, clientPort, err := net.SplitHostPort(firstStatsdClient.Conn.LocalAddr().String())
+		require.NoError(t, err)
+		assert.Equal(t, clientAddress, sourceAddress.AsString())
+		assert.Equal(t, clientPort, sourcePort.AsString())
+
 		secondResource := mdd[1].ResourceMetrics().At(0)
-		resourceAttributeSource, exists = secondResource.Resource().Attributes().Get("source")
+		sourceAddress, exists = secondResource.Resource().Attributes().Get("source.address")
+		assert.Equal(t, true, exists)
+		sourcePort, exists = secondResource.Resource().Attributes().Get("source.port")
 		assert.Equal(t, true, exists)
-		assert.Equal(t, secondStatsdClient.Conn.LocalAddr().String(), resourceAttributeSource.AsString())
+		clientAddress, clientPort, err = net.SplitHostPort(secondStatsdClient.Conn.LocalAddr().String())
+		require.NoError(t, err)
+		assert.Equal(t, clientAddress, sourceAddress.AsString())
+		assert.Equal(t, clientPort, sourcePort.AsString())
 	})
 	t.Run("do not aggregate by source address with two clients sending", func(t *testing.T) {
 		cfg := &Config{
@@ -262,9 +274,14 @@ func Test_statsdreceiver_resource_attribute_source(t *testing.T) {
 		require.Equal(t, 1, mdd[0].ResourceMetrics().Len())
 
 		// The resources should have source attribute matching the servers address
-		firstResource := mdd[0].ResourceMetrics().At(0)
-		resourceAttributeSource, exists := firstResource.Resource().Attributes().Get("source")
+		resource := mdd[0].ResourceMetrics().At(0)
+		sourceAddress, exists := resource.Resource().Attributes().Get("source.address")
 		assert.Equal(t, true, exists)
-		assert.Equal(t, serverAddr, resourceAttributeSource.AsString())
+		sourcePort, exists := resource.Resource().Attributes().Get("source.port")
+		assert.Equal(t, true, exists)
+		serverAddr, serverPort, err := net.SplitHostPort(serverAddr)
+		require.NoError(t, err)
+		assert.Equal(t, serverAddr, sourceAddress.AsString())
+		assert.Equal(t, serverPort, sourcePort.AsString())
 	})
 }