[mq] [skip ddci] working branch - merge ef517d6 on top of main at 7a849eb

gh-worker-dd-mergequeue-cf854d[bot] · web-flow · commit c13425e50c60 · 2026-03-02T17:09:26.000Z
{"baseBranch":"main","baseCommit":"7a849eb98fd3e71256e8f046912e36f30f935ec1","createdAt":"2026-03-02T17:05:08.578971Z","headSha":"ef517d61a57e2aa0706760b5e8095b42db6e3993","id":"6009d045-4860-410f-aa9d-71f451f27c79","nextMergeabilityCheckAt":"2026-03-02T18:09:05.003955Z","priority":"200","pullRequestNumber":"4484","queuedAt":"2026-03-02T17:09:17.400284Z","status":"STATUS_QUEUED"}
diff --git a/ddtrace/tracer/span.go b/ddtrace/tracer/span.go
@@ -358,7 +358,7 @@ func (s *Span) BaggageItem(key string) string {
 func safeStringerValue(v fmt.Stringer, original any) (result string) {
 	defer func() {
 		if e := recover(); e != nil {
-			if rv := reflect.ValueOf(original); rv.Kind() == reflect.Ptr && rv.IsNil() {
+			if rv := reflect.ValueOf(original); rv.Kind() == reflect.Pointer && rv.IsNil() {
 				result = "<nil>"
 				return
 			}
diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go
@@ -252,7 +252,17 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) {
 	for _, csp := range csps {
 		csp.ProcessTags = processtags.GlobalTags().String()
 		flushedBuckets += len(csp.Stats)
-		if err := c.cfg.transport.sendStats(csp, obfVersion); err != nil {
+		var err error
+		for attempt := 0; attempt <= c.cfg.sendRetries; attempt++ {
+			err = c.cfg.transport.sendStats(csp, obfVersion)
+			if err == nil {
+				break
+			}
+			if attempt < c.cfg.sendRetries {
+				time.Sleep(c.cfg.internalConfig.RetryInterval())
+			}
+		}
+		if err != nil {
 			c.statsd().Incr("datadog.tracer.stats.flush_errors", nil, 1)
 			log.Error("Error sending stats payload: %s", err.Error())
 		}
diff --git a/ddtrace/tracer/stats_test.go b/ddtrace/tracer/stats_test.go
@@ -6,6 +6,8 @@
 package tracer
 
 import (
+	"errors"
+	"fmt"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -14,6 +16,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/DataDog/datadog-agent/pkg/obfuscate"
+	pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/trace"
 	"github.com/DataDog/datadog-go/v5/statsd"
 
 	"github.com/DataDog/dd-trace-go/v2/ddtrace/ext"
@@ -335,3 +338,76 @@ func TestStatsIncludeHTTPMethodAndEndpoint(t *testing.T) {
 	assert.Equal(t, uniqueMethod, group.GetHTTPMethod())
 	assert.Equal(t, uniqueEndpoint, group.GetHTTPEndpoint())
 }
+
+// failingStatsTransport is a transport whose sendStats fails a configurable
+// number of times before succeeding, used to test retry behaviour.
+type failingStatsTransport struct {
+	dummyTransport
+	failCount    int
+	sendAttempts int
+	statsSent    bool
+}
+
+func (t *failingStatsTransport) sendStats(_ *pb.ClientStatsPayload, _ int) error {
+	t.sendAttempts++
+	if t.failCount > 0 {
+		t.failCount--
+		return errors.New("stats send failed")
+	}
+	t.statsSent = true
+	return nil
+}
+
+func TestStatsFlushRetries(t *testing.T) {
+	testcases := []struct {
+		configRetries int
+		retryInterval time.Duration
+		failCount     int
+		statsSent     bool
+		expAttempts   int
+	}{
+		{configRetries: 0, retryInterval: time.Millisecond, failCount: 0, statsSent: true, expAttempts: 1},
+		{configRetries: 0, retryInterval: time.Millisecond, failCount: 1, statsSent: false, expAttempts: 1},
+
+		{configRetries: 1, retryInterval: time.Millisecond, failCount: 0, statsSent: true, expAttempts: 1},
+		{configRetries: 1, retryInterval: time.Millisecond, failCount: 1, statsSent: true, expAttempts: 2},
+		{configRetries: 1, retryInterval: time.Millisecond, failCount: 2, statsSent: false, expAttempts: 2},
+
+		{configRetries: 2, retryInterval: time.Millisecond, failCount: 0, statsSent: true, expAttempts: 1},
+		{configRetries: 2, retryInterval: time.Millisecond, failCount: 1, statsSent: true, expAttempts: 2},
+		{configRetries: 2, retryInterval: time.Millisecond, failCount: 2, statsSent: true, expAttempts: 3},
+		{configRetries: 2, retryInterval: time.Millisecond, failCount: 3, statsSent: false, expAttempts: 3},
+	}
+
+	bucketSize := int64(500_000)
+	s := Span{
+		name:     "http.request",
+		start:    time.Now().UnixNano() + 3*bucketSize,
+		duration: 1,
+		metrics:  map[string]float64{keyMeasured: 1},
+	}
+
+	for _, test := range testcases {
+		name := fmt.Sprintf("retries=%d/fails=%d/sent=%v", test.configRetries, test.failCount, test.statsSent)
+		t.Run(name, func(t *testing.T) {
+			p := &failingStatsTransport{failCount: test.failCount}
+			cfg, err := newTestConfig(func(c *config) {
+				c.transport = p
+				c.sendRetries = test.configRetries
+				c.internalConfig.SetRetryInterval(test.retryInterval, internalconfig.OriginCode)
+				c.internalConfig.SetEnv("someEnv", internalconfig.OriginCode)
+			})
+			require.NoError(t, err)
+
+			c := newConcentrator(cfg, bucketSize, &statsd.NoOpClientDirect{})
+			ss, ok := c.newTracerStatSpan(&s, nil)
+			require.True(t, ok)
+			c.Start()
+			c.In <- ss
+			c.Stop()
+
+			assert.Equal(t, test.expAttempts, p.sendAttempts)
+			assert.Equal(t, test.statsSent, p.statsSent)
+		})
+	}
+}
diff --git a/ddtrace/tracer/transport_test.go b/ddtrace/tracer/transport_test.go
@@ -16,7 +16,10 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/trace"
 
@@ -596,3 +599,68 @@ func TestClientComputedStatsHeader(t *testing.T) {
 		assert.Equal("t", headerValue, "Datadog-Client-Computed-Stats header should be set to 't' when both conditions are met")
 	})
 }
+
+// TestConcurrentTraceFlushOverUDS verifies that multiple goroutines can send trace
+// payloads concurrently through the HTTP transport backed by a real UDS socket without
+// errors. This exercises the connection pool fix (MaxIdleConnsPerHost=100) under
+// realistic end-to-end conditions rather than just asserting config values.
+func TestConcurrentTraceFlushOverUDS(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping concurrent UDS transport test in short mode")
+	}
+
+	dir, err := os.MkdirTemp("", "uds-transport-test")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+
+	socketPath := filepath.Join(dir, "traces.socket")
+	ln, err := net.Listen("unix", socketPath)
+	require.NoError(t, err)
+
+	var received atomic.Int64
+	srv := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			received.Add(1)
+			w.WriteHeader(http.StatusOK)
+			w.Write([]byte(`{"rate_by_service":{}}`)) //nolint:errcheck
+		}),
+	}
+	go srv.Serve(ln) //nolint:errcheck
+	defer srv.Close()
+
+	udsURL := internal.UnixDataSocketURL(socketPath).String()
+	client := internal.UDSClient(socketPath, 5*time.Second)
+	transport := newHTTPTransport(udsURL, client)
+
+	const numGoroutines = 20
+
+	start := make(chan struct{})
+	errs := make([]error, numGoroutines)
+	var wg sync.WaitGroup
+
+	for i := range numGoroutines {
+		wg.Go(func() {
+			<-start
+			p, encErr := encode(getTestTrace(1, 1))
+			if encErr != nil {
+				errs[i] = encErr
+				return
+			}
+			body, sendErr := transport.send(p)
+			if sendErr != nil {
+				errs[i] = sendErr
+				return
+			}
+			body.Close()
+		})
+	}
+
+	close(start)
+	wg.Wait()
+
+	for i, e := range errs {
+		assert.NoError(t, e, "goroutine %d send failed", i)
+	}
+	assert.Equal(t, int64(numGoroutines), received.Load(),
+		"server should have received all %d trace payloads", numGoroutines)
+}
diff --git a/internal/uds.go b/internal/uds.go
@@ -33,7 +33,10 @@ func UDSClient(socketPath string, timeout time.Duration) *http.Client {
 					Net:  "unix",
 				}).String())
 			},
-			MaxIdleConns:          100,
+			MaxIdleConns: 100,
+			// All UDS requests share a single synthetic hostname, so MaxIdleConnsPerHost
+			// must match MaxIdleConns to prevent connection churn under concurrent flushes.
+			MaxIdleConnsPerHost:   100,
 			IdleConnTimeout:       90 * time.Second,
 			TLSHandshakeTimeout:   10 * time.Second,
 			ExpectContinueTimeout: 1 * time.Second,
diff --git a/internal/uds_test.go b/internal/uds_test.go
@@ -6,10 +6,18 @@
 package internal
 
 import (
+	"net"
+	"net/http"
 	"net/url"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestUnixDataSocketURL(t *testing.T) {
@@ -146,3 +154,154 @@ func TestUnixDataSocketURL(t *testing.T) {
 		})
 	}
 }
+
+func TestUDSClientTransportConfig(t *testing.T) {
+	client := UDSClient("/var/run/datadog/apm.socket", 10*time.Second)
+	tr, ok := client.Transport.(*http.Transport)
+	require.True(t, ok, "Transport should be *http.Transport")
+	assert.Equal(t, 100, tr.MaxIdleConns)
+	assert.Equal(t, 100, tr.MaxIdleConnsPerHost)
+	assert.Equal(t, 90*time.Second, tr.IdleConnTimeout)
+	assert.Equal(t, 10*time.Second, tr.TLSHandshakeTimeout)
+	assert.Equal(t, 1*time.Second, tr.ExpectContinueTimeout)
+}
+
+// TestUDSConcurrentConnectionReuse verifies that MaxIdleConnsPerHost=100 prevents
+// connection churn when many goroutines send requests concurrently over a UDS socket.
+// Before the fix, MaxIdleConnsPerHost defaulted to 2, which forced new connections
+// for every request beyond the 2-connection idle pool, causing "connection reset by
+// peer" errors under agent backpressure.
+func TestUDSConcurrentConnectionReuse(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping connection pool test in short mode")
+	}
+
+	dir, err := os.MkdirTemp("", "uds-pool-test")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+
+	socketPath := filepath.Join(dir, "test.socket")
+	ln, err := net.Listen("unix", socketPath)
+	require.NoError(t, err)
+
+	var newConnections atomic.Int64
+	srv := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			w.WriteHeader(http.StatusOK)
+		}),
+		ConnState: func(_ net.Conn, state http.ConnState) {
+			if state == http.StateNew {
+				newConnections.Add(1)
+			}
+		},
+	}
+	go srv.Serve(ln) //nolint:errcheck
+	defer srv.Close()
+
+	client := UDSClient(socketPath, 5*time.Second)
+
+	const (
+		numGoroutines = 50
+		requestsEach  = 10
+	)
+
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+
+	for range numGoroutines {
+		wg.Go(func() {
+			<-start
+			for range requestsEach {
+				req, err := http.NewRequest(http.MethodGet, "http://localhost/", nil)
+				if err != nil {
+					return
+				}
+				resp, err := client.Do(req)
+				if err != nil {
+					return
+				}
+				resp.Body.Close()
+			}
+		})
+	}
+
+	close(start)
+	wg.Wait()
+
+	// With MaxIdleConnsPerHost=100, each goroutine reuses its connection for all
+	// 10 requests. Expect ~50 new connections (one per goroutine), not 500 (one
+	// per request as would happen with the old MaxIdleConnsPerHost=2 default).
+	assert.LessOrEqual(t, newConnections.Load(), int64(55),
+		"connections should be reused; got %d new connections for %d requests",
+		newConnections.Load(), numGoroutines*requestsEach)
+}
+
+// TestUDSServerCloseRecovery verifies that the UDS HTTP client recovers transparently
+// from server-side connection closes, which reproduce agent backpressure / restart
+// scenarios that previously caused "broken pipe" or "connection reset by peer" errors.
+func TestUDSServerCloseRecovery(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping connection recovery test in short mode")
+	}
+
+	dir, err := os.MkdirTemp("", "uds-recovery-test")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+
+	socketPath := filepath.Join(dir, "test.socket")
+	ln, err := net.Listen("unix", socketPath)
+	require.NoError(t, err)
+
+	var requestCount atomic.Int64
+	srv := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			n := requestCount.Add(1)
+			// Force connection close every 5th request to simulate agent backpressure.
+			if n%5 == 0 {
+				w.Header().Set("Connection", "close")
+			}
+			w.WriteHeader(http.StatusOK)
+		}),
+	}
+	go srv.Serve(ln) //nolint:errcheck
+	defer srv.Close()
+
+	client := UDSClient(socketPath, 5*time.Second)
+
+	const (
+		numGoroutines = 20
+		requestsEach  = 20
+	)
+
+	start := make(chan struct{})
+	var successes atomic.Int64
+	var wg sync.WaitGroup
+
+	for range numGoroutines {
+		wg.Go(func() {
+			<-start
+			for range requestsEach {
+				req, err := http.NewRequest(http.MethodGet, "http://localhost/", nil)
+				if err != nil {
+					continue
+				}
+				resp, err := client.Do(req)
+				if err != nil {
+					continue
+				}
+				resp.Body.Close()
+				if resp.StatusCode == http.StatusOK {
+					successes.Add(1)
+				}
+			}
+		})
+	}
+
+	close(start)
+	wg.Wait()
+
+	// All requests must succeed. The HTTP client transparently opens a new
+	// connection after a server-forced close, so no request should be lost.
+	assert.Equal(t, int64(numGoroutines*requestsEach), successes.Load(),
+		"all requests should succeed despite periodic server-side connection closes")
+}

Original file line number	Diff line number	Diff line change
`@@ -358,7 +358,7 @@ func (s *Span) BaggageItem(key string) string {`
`358`	`358`	`func safeStringerValue(v fmt.Stringer, original any) (result string) {`
`359`	`359`	`defer func() {`
`360`	`360`	`if e := recover(); e != nil {`
`361`		`- if rv := reflect.ValueOf(original); rv.Kind() == reflect.Ptr && rv.IsNil() {`
	`361`	`+ if rv := reflect.ValueOf(original); rv.Kind() == reflect.Pointer && rv.IsNil() {`
`362`	`362`	`result = "<nil>"`
`363`	`363`	`return`
`364`	`364`	`}`