[UDS] Use better payload size defaults

For UDS:
* 8KB datagrams
* change internal buffer and queue sizes accordingly to avoid
  increasing the mem usage of the client.

statsd/statsd.go

@@ -57,7 +57,8 @@ const DefaultUDSBufferPoolSize = 512
 /*
 DefaultMaxAgentPayloadSize is the default maximum payload size the agent
 can receive. This can be adjusted by changing dogstatsd_buffer_size in the
-agent configuration file datadog.yaml.
+agent configuration file datadog.yaml. This is also used as the optimal payload size
+for UDS datagrams.
 */
 const DefaultMaxAgentPayloadSize = 8192
 
@@ -102,6 +103,11 @@ const (
 	ChannelMode
 )
 
+const (
+	WriterNameUDP string = "udp"
+	WriterNameUDS string = "uds"
+)
+
 type metric struct {
 	metricType metricType
 	namespace  string
@@ -225,11 +231,11 @@ var _ ClientInterface = &Client{}
 func resolveAddr(addr string) (statsdWriter, string, error) {
 	if !strings.HasPrefix(addr, UnixAddressPrefix) {
 		w, err := newUDPWriter(addr)
-		return w, "udp", err
+		return w, WriterNameUDP, err
 	}
 
 	w, err := newUDSWriter(addr[len(UnixAddressPrefix):])
-	return w, "uds", err
+	return w, WriterNameUDS, err
 }
 
 // New returns a pointer to a new Client given an addr in the format "hostname:port" or
@@ -297,17 +303,26 @@ func newWithWriter(w statsdWriter, o *Options, writerName string) (*Client, erro
 		}
 	}
 
-	// FIXME: The agent has a performance pitfall preventing us from using better defaults for UDS,
-	// this is why we fallback on UDP defaults even in UDS mode.
-	// Once it's fixed, use `DefaultMaxAgentPayloadSize` and `DefaultUDSBufferPoolSize` instead for UDS.
 	if o.MaxBytesPerPayload == 0 {
-		o.MaxBytesPerPayload = OptimalUDPPayloadSize
+		if writerName == WriterNameUDS {
+			o.MaxBytesPerPayload = DefaultMaxAgentPayloadSize
+		} else {
+			o.MaxBytesPerPayload = OptimalUDPPayloadSize
+		}
 	}
 	if o.BufferPoolSize == 0 {
-		o.BufferPoolSize = DefaultUDPBufferPoolSize
+		if writerName == WriterNameUDS {
+			o.BufferPoolSize = DefaultUDSBufferPoolSize
+		} else {
+			o.BufferPoolSize = DefaultUDPBufferPoolSize
+		}
 	}
 	if o.SenderQueueSize == 0 {
-		o.SenderQueueSize = DefaultUDPBufferPoolSize
+		if writerName == WriterNameUDS {
+			o.SenderQueueSize = DefaultUDSBufferPoolSize
+		} else {
+			o.SenderQueueSize = DefaultUDPBufferPoolSize
+		}
 	}
 
 	bufferPool := newBufferPool(o.BufferPoolSize, o.MaxBytesPerPayload, o.MaxMessagesPerPayload)

statsd/statsd_benchmark_test.go

@@ -57,7 +57,7 @@ func setupClient(b *testing.B, transport string, extraOptions []statsd.Option) (
 	options := []statsd.Option{statsd.WithMaxMessagesPerPayload(1024), statsd.WithoutTelemetry()}
 	options = append(options, extraOptions...)
 
-	if transport == "udp" {
+	if transport == statsd.WriterNameUDP {
 		return setupUDPClientServer(b, options)
 	}
 	return setupUDSClientServer(b, options)
@@ -110,22 +110,22 @@ UDP with the same metric
 
 // blocking + no aggregation
 func BenchmarkStatsdUDPSameMetricMutex(b *testing.B) {
-	benchmarkStatsdSameMetrics(b, "udp", statsd.WithMutexMode(), statsd.WithoutClientSideAggregation())
+	benchmarkStatsdSameMetrics(b, statsd.WriterNameUDP, statsd.WithMutexMode(), statsd.WithoutClientSideAggregation())
 }
 
 // dropping + no aggregation
 func BenchmarkStatsdUDPSameMetricChannel(b *testing.B) {
-	benchmarkStatsdSameMetrics(b, "udp", statsd.WithChannelMode(), statsd.WithoutClientSideAggregation())
+	benchmarkStatsdSameMetrics(b, statsd.WriterNameUDP, statsd.WithChannelMode(), statsd.WithoutClientSideAggregation())
 }
 
 // blocking + aggregation
 func BenchmarkStatsdUDPSameMetricMutexAggregation(b *testing.B) {
-	benchmarkStatsdSameMetrics(b, "udp", statsd.WithMutexMode(), statsd.WithClientSideAggregation())
+	benchmarkStatsdSameMetrics(b, statsd.WriterNameUDP, statsd.WithMutexMode(), statsd.WithClientSideAggregation())
 }
 
 // dropping + aggregation
 func BenchmarkStatsdUDPSameMetricChannelAggregation(b *testing.B) {
-	benchmarkStatsdSameMetrics(b, "udp", statsd.WithChannelMode(), statsd.WithClientSideAggregation())
+	benchmarkStatsdSameMetrics(b, statsd.WriterNameUDP, statsd.WithChannelMode(), statsd.WithClientSideAggregation())
 }
 
 /*
@@ -134,66 +134,66 @@ UDP with the different metrics
 
 // blocking + no aggregation
 func BenchmarkStatsdUDPDifferentMetricMutex(b *testing.B) {
-	benchmarkStatsdDifferentMetrics(b, "udp", statsd.WithMutexMode(), statsd.WithoutClientSideAggregation())
+	benchmarkStatsdDifferentMetrics(b, statsd.WriterNameUDP, statsd.WithMutexMode(), statsd.WithoutClientSideAggregation())
 }
 
 // dropping + no aggregation
 func BenchmarkStatsdUDPDifferentMetricChannel(b *testing.B) {
-	benchmarkStatsdDifferentMetrics(b, "udp", statsd.WithChannelMode(), statsd.WithoutClientSideAggregation())
+	benchmarkStatsdDifferentMetrics(b, statsd.WriterNameUDP, statsd.WithChannelMode(), statsd.WithoutClientSideAggregation())
 }
 
 // blocking + aggregation
 func BenchmarkStatsdUDPDifferentMetricMutexAggregation(b *testing.B) {
-	benchmarkStatsdDifferentMetrics(b, "udp", statsd.WithMutexMode(), statsd.WithClientSideAggregation())
+	benchmarkStatsdDifferentMetrics(b, statsd.WriterNameUDP, statsd.WithMutexMode(), statsd.WithClientSideAggregation())
 }
 
 // dropping + aggregation
 func BenchmarkStatsdUDPDifferentMetricChannelAggregation(b *testing.B) {
-	benchmarkStatsdDifferentMetrics(b, "udp", statsd.WithChannelMode(), statsd.WithClientSideAggregation())
+	benchmarkStatsdDifferentMetrics(b, statsd.WriterNameUDP, statsd.WithChannelMode(), statsd.WithClientSideAggregation())
 }
 
 /*
 UDS with the same metric
 */
 // blocking + no aggregation
 func BenchmarkStatsdUDSSameMetricMutex(b *testing.B) {
-	benchmarkStatsdSameMetrics(b, "uds", statsd.WithMutexMode(), statsd.WithoutClientSideAggregation())
+	benchmarkStatsdSameMetrics(b, statsd.WriterNameUDS, statsd.WithMutexMode(), statsd.WithoutClientSideAggregation())
 }
 
 // dropping + no aggregation
 func BenchmarkStatsdUDSSameMetricChannel(b *testing.B) {
-	benchmarkStatsdSameMetrics(b, "uds", statsd.WithChannelMode(), statsd.WithoutClientSideAggregation())
+	benchmarkStatsdSameMetrics(b, statsd.WriterNameUDS, statsd.WithChannelMode(), statsd.WithoutClientSideAggregation())
 }
 
 // blocking + aggregation
 func BenchmarkStatsdUDSSameMetricMutexAggregation(b *testing.B) {
-	benchmarkStatsdSameMetrics(b, "uds", statsd.WithMutexMode(), statsd.WithClientSideAggregation())
+	benchmarkStatsdSameMetrics(b, statsd.WriterNameUDS, statsd.WithMutexMode(), statsd.WithClientSideAggregation())
 }
 
 // dropping + aggregation
 func BenchmarkStatsdUDSSameMetricChannelAggregation(b *testing.B) {
-	benchmarkStatsdSameMetrics(b, "uds", statsd.WithChannelMode(), statsd.WithClientSideAggregation())
+	benchmarkStatsdSameMetrics(b, statsd.WriterNameUDS, statsd.WithChannelMode(), statsd.WithClientSideAggregation())
 }
 
 /*
 UDS with different metrics
 */
 // blocking + no aggregation
 func BenchmarkStatsdUDPSifferentMetricMutex(b *testing.B) {
-	benchmarkStatsdDifferentMetrics(b, "uds", statsd.WithMutexMode(), statsd.WithoutClientSideAggregation())
+	benchmarkStatsdDifferentMetrics(b, statsd.WriterNameUDS, statsd.WithMutexMode(), statsd.WithoutClientSideAggregation())
 }
 
 // dropping + no aggregation
 func BenchmarkStatsdUDSDifferentMetricChannel(b *testing.B) {
-	benchmarkStatsdDifferentMetrics(b, "uds", statsd.WithChannelMode(), statsd.WithoutClientSideAggregation())
+	benchmarkStatsdDifferentMetrics(b, statsd.WriterNameUDS, statsd.WithChannelMode(), statsd.WithoutClientSideAggregation())
 }
 
 // blocking + aggregation
 func BenchmarkStatsdUDPSifferentMetricMutexAggregation(b *testing.B) {
-	benchmarkStatsdDifferentMetrics(b, "uds", statsd.WithMutexMode(), statsd.WithClientSideAggregation())
+	benchmarkStatsdDifferentMetrics(b, statsd.WriterNameUDS, statsd.WithMutexMode(), statsd.WithClientSideAggregation())
 }
 
 // dropping + aggregation
 func BenchmarkStatsdUDSDifferentMetricChannelAggregation(b *testing.B) {
-	benchmarkStatsdDifferentMetrics(b, "uds", statsd.WithChannelMode(), statsd.WithClientSideAggregation())
+	benchmarkStatsdDifferentMetrics(b, statsd.WriterNameUDS, statsd.WithChannelMode(), statsd.WithClientSideAggregation())
 }