|
6 | 6 | package internal |
7 | 7 |
|
8 | 8 | import ( |
| 9 | + "net" |
| 10 | + "net/http" |
9 | 11 | "net/url" |
| 12 | + "os" |
| 13 | + "path/filepath" |
| 14 | + "sync" |
| 15 | + "sync/atomic" |
10 | 16 | "testing" |
| 17 | + "time" |
11 | 18 |
|
12 | 19 | "github.com/stretchr/testify/assert" |
| 20 | + "github.com/stretchr/testify/require" |
13 | 21 | ) |
14 | 22 |
|
15 | 23 | func TestUnixDataSocketURL(t *testing.T) { |
@@ -146,3 +154,157 @@ func TestUnixDataSocketURL(t *testing.T) { |
146 | 154 | }) |
147 | 155 | } |
148 | 156 | } |
| 157 | + |
| 158 | +func TestUDSClientTransportConfig(t *testing.T) { |
| 159 | + client := UDSClient("/var/run/datadog/apm.socket", 10*time.Second) |
| 160 | + tr, ok := client.Transport.(*http.Transport) |
| 161 | + require.True(t, ok, "Transport should be *http.Transport") |
| 162 | + assert.Equal(t, 100, tr.MaxIdleConns) |
| 163 | + assert.Equal(t, 100, tr.MaxIdleConnsPerHost) |
| 164 | + assert.Equal(t, 90*time.Second, tr.IdleConnTimeout) |
| 165 | + assert.Equal(t, 10*time.Second, tr.TLSHandshakeTimeout) |
| 166 | + assert.Equal(t, 1*time.Second, tr.ExpectContinueTimeout) |
| 167 | +} |
| 168 | + |
| 169 | +// TestUDSConcurrentConnectionReuse verifies that MaxIdleConnsPerHost=100 prevents |
| 170 | +// connection churn when many goroutines send requests concurrently over a UDS socket. |
| 171 | +// Before the fix, MaxIdleConnsPerHost defaulted to 2, which forced new connections |
| 172 | +// for every request beyond the 2-connection idle pool, causing "connection reset by |
| 173 | +// peer" errors under agent backpressure. |
| 174 | +func TestUDSConcurrentConnectionReuse(t *testing.T) { |
| 175 | + if testing.Short() { |
| 176 | + t.Skip("skipping connection pool test in short mode") |
| 177 | + } |
| 178 | + |
| 179 | + dir, err := os.MkdirTemp("", "uds-pool-test") |
| 180 | + require.NoError(t, err) |
| 181 | + defer os.RemoveAll(dir) |
| 182 | + |
| 183 | + socketPath := filepath.Join(dir, "test.socket") |
| 184 | + ln, err := net.Listen("unix", socketPath) |
| 185 | + require.NoError(t, err) |
| 186 | + |
| 187 | + var newConnections atomic.Int64 |
| 188 | + srv := &http.Server{ |
| 189 | + Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { |
| 190 | + w.WriteHeader(http.StatusOK) |
| 191 | + }), |
| 192 | + ConnState: func(_ net.Conn, state http.ConnState) { |
| 193 | + if state == http.StateNew { |
| 194 | + newConnections.Add(1) |
| 195 | + } |
| 196 | + }, |
| 197 | + } |
| 198 | + go srv.Serve(ln) //nolint:errcheck |
| 199 | + defer srv.Close() |
| 200 | + |
| 201 | + client := UDSClient(socketPath, 5*time.Second) |
| 202 | + |
| 203 | + const ( |
| 204 | + numGoroutines = 50 |
| 205 | + requestsEach = 10 |
| 206 | + ) |
| 207 | + |
| 208 | + start := make(chan struct{}) |
| 209 | + var wg sync.WaitGroup |
| 210 | + |
| 211 | + for range numGoroutines { |
| 212 | + wg.Go(func() { |
| 213 | + <-start |
| 214 | + for range requestsEach { |
| 215 | + req, err := http.NewRequest(http.MethodGet, "http://localhost/", nil) |
| 216 | + if err != nil { |
| 217 | + return |
| 218 | + } |
| 219 | + resp, err := client.Do(req) |
| 220 | + if err != nil { |
| 221 | + return |
| 222 | + } |
| 223 | + resp.Body.Close() |
| 224 | + } |
| 225 | + }) |
| 226 | + } |
| 227 | + |
| 228 | + close(start) |
| 229 | + wg.Wait() |
| 230 | + |
| 231 | + // With MaxIdleConnsPerHost=100, connections are heavily reused. Ideally ~50 |
| 232 | + // new connections (one per goroutine), but timing races between goroutines |
| 233 | + // competing for idle connections can push the count above that — especially |
| 234 | + // on Windows where scheduler and socket latency differ. The important |
| 235 | + // invariant is that the count is far below 500 (one per request, as would |
| 236 | + // happen with the old MaxIdleConnsPerHost=2 default). |
| 237 | + assert.LessOrEqual(t, newConnections.Load(), int64(numGoroutines*2), |
| 238 | + "connections should be reused; got %d new connections for %d requests", |
| 239 | + newConnections.Load(), numGoroutines*requestsEach) |
| 240 | +} |
| 241 | + |
| 242 | +// TestUDSServerCloseRecovery verifies that the UDS HTTP client recovers transparently |
| 243 | +// from server-side connection closes, which reproduce agent backpressure / restart |
| 244 | +// scenarios that previously caused "broken pipe" or "connection reset by peer" errors. |
| 245 | +func TestUDSServerCloseRecovery(t *testing.T) { |
| 246 | + if testing.Short() { |
| 247 | + t.Skip("skipping connection recovery test in short mode") |
| 248 | + } |
| 249 | + |
| 250 | + dir, err := os.MkdirTemp("", "uds-recovery-test") |
| 251 | + require.NoError(t, err) |
| 252 | + defer os.RemoveAll(dir) |
| 253 | + |
| 254 | + socketPath := filepath.Join(dir, "test.socket") |
| 255 | + ln, err := net.Listen("unix", socketPath) |
| 256 | + require.NoError(t, err) |
| 257 | + |
| 258 | + var requestCount atomic.Int64 |
| 259 | + srv := &http.Server{ |
| 260 | + Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { |
| 261 | + n := requestCount.Add(1) |
| 262 | + // Force connection close every 5th request to simulate agent backpressure. |
| 263 | + if n%5 == 0 { |
| 264 | + w.Header().Set("Connection", "close") |
| 265 | + } |
| 266 | + w.WriteHeader(http.StatusOK) |
| 267 | + }), |
| 268 | + } |
| 269 | + go srv.Serve(ln) //nolint:errcheck |
| 270 | + defer srv.Close() |
| 271 | + |
| 272 | + client := UDSClient(socketPath, 5*time.Second) |
| 273 | + |
| 274 | + const ( |
| 275 | + numGoroutines = 20 |
| 276 | + requestsEach = 20 |
| 277 | + ) |
| 278 | + |
| 279 | + start := make(chan struct{}) |
| 280 | + var successes atomic.Int64 |
| 281 | + var wg sync.WaitGroup |
| 282 | + |
| 283 | + for range numGoroutines { |
| 284 | + wg.Go(func() { |
| 285 | + <-start |
| 286 | + for range requestsEach { |
| 287 | + req, err := http.NewRequest(http.MethodGet, "http://localhost/", nil) |
| 288 | + if err != nil { |
| 289 | + continue |
| 290 | + } |
| 291 | + resp, err := client.Do(req) |
| 292 | + if err != nil { |
| 293 | + continue |
| 294 | + } |
| 295 | + resp.Body.Close() |
| 296 | + if resp.StatusCode == http.StatusOK { |
| 297 | + successes.Add(1) |
| 298 | + } |
| 299 | + } |
| 300 | + }) |
| 301 | + } |
| 302 | + |
| 303 | + close(start) |
| 304 | + wg.Wait() |
| 305 | + |
| 306 | + // All requests must succeed. The HTTP client transparently opens a new |
| 307 | + // connection after a server-forced close, so no request should be lost. |
| 308 | + assert.Equal(t, int64(numGoroutines*requestsEach), successes.Load(), |
| 309 | + "all requests should succeed despite periodic server-side connection closes") |
| 310 | +} |
![gh-worker-dd-mergequeue-cf854d[bot]](https://avatars.githubusercontent.com/u/244837012?v=4&size=40){kind=avatar}
0 commit comments