diff --git a/server/http_client_test.go b/server/http_client_test.go new file mode 100644 index 0000000..1abf34a --- /dev/null +++ b/server/http_client_test.go @@ -0,0 +1,54 @@ +// Copyright 2024 Block, Inc. + +package server + +import ( + "net/http" + "testing" + "time" + + "github.com/cashapp/blip" + "github.com/stretchr/testify/require" +) + +func TestHTTPClientFactoryMakeForSink(t *testing.T) { + tests := []struct { + name string + proxy string + }{ + {name: "default"}, + {name: "proxy", proxy: "http://proxy.example:8080"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + factory := httpClientFactory{cfg: blip.ConfigHTTP{Proxy: tt.proxy}} + client, err := factory.MakeForSink("datadog", "m1", nil, nil) + require.NoError(t, err) + require.Equal(t, 10*time.Second, client.Timeout) + + transport, ok := client.Transport.(*http.Transport) + require.True(t, ok) + require.True(t, transport.ForceAttemptHTTP2) + require.NotNil(t, transport.DialContext) + require.Equal(t, 5*time.Second, transport.TLSHandshakeTimeout) + require.Equal(t, 5*time.Second, transport.ResponseHeaderTimeout) + require.Equal(t, 1*time.Second, transport.ExpectContinueTimeout) + require.Equal(t, 30*time.Second, transport.IdleConnTimeout) + + otherClient, err := factory.MakeForSink("datadog", "m2", nil, nil) + require.NoError(t, err) + require.NotSame(t, client.Transport, otherClient.Transport) + + if tt.proxy == "" { + return + } + + request, err := http.NewRequest(http.MethodGet, "https://api.datadoghq.com", nil) + require.NoError(t, err) + proxyURL, err := transport.Proxy(request) + require.NoError(t, err) + require.Equal(t, tt.proxy, proxyURL.String()) + }) + } +} diff --git a/server/server.go b/server/server.go index 498334f..ef2fb3e 100644 --- a/server/server.go +++ b/server/server.go @@ -6,6 +6,7 @@ package server import ( "context" "fmt" + "net" "net/http" "net/url" "os" @@ -51,15 +52,27 @@ type httpClientFactory struct { } func (f httpClientFactory) MakeForSink(sinkName, monitorId string, opts, tags map[string]string) (*http.Client, error) { - client := &http.Client{} + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.DialContext = (&net.Dialer{ + Timeout: 5 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext + transport.TLSHandshakeTimeout = 5 * time.Second + transport.ResponseHeaderTimeout = 5 * time.Second + transport.ExpectContinueTimeout = 1 * time.Second + transport.IdleConnTimeout = 30 * time.Second + transport.MaxIdleConnsPerHost = 2 if f.cfg.Proxy != "" { proxyFunc := func(req *http.Request) (url *url.URL, err error) { return url.Parse(f.cfg.Proxy) } - client.Transport = &http.Transport{Proxy: proxyFunc} + transport.Proxy = proxyFunc blip.Debug("%s sink %s http proxy via %s", monitorId, sinkName, f.cfg.Proxy) } - return client, nil + return &http.Client{ + Timeout: 10 * time.Second, + Transport: transport, + }, nil } // -------------------------------------------------------------------------- diff --git a/sink/datadog_test.go b/sink/datadog_test.go index 9a82337..8440399 100644 --- a/sink/datadog_test.go +++ b/sink/datadog_test.go @@ -123,6 +123,32 @@ func TestDatadogSink(t *testing.T) { } } +func TestDatadogSendBoundedByClientTimeout(t *testing.T) { + httpClient := &http.Client{ + Timeout: 200 * time.Millisecond, + Transport: &mock.Transport{ + RoundTripFunc: func(r *http.Request) (*http.Response, error) { + <-r.Context().Done() + return nil, r.Context().Err() + }, + }, + } + ddSink, err := NewDatadog("testmonitor", defaultOps(), map[string]string{}, httpClient) + require.NoError(t, err) + + done := make(chan error, 1) + go func() { + done <- ddSink.Send(context.Background(), getBlipCounterMetrics(1, 1.0, false)) + }() + + select { + case err := <-done: + require.Error(t, err) + case <-time.After(2 * time.Second): + t.Fatal("Datadog.Send stalled: http.Client.Timeout did not bound the request") + } +} + func TestDatadogMetricsPerRequest(t *testing.T) { callCount := 0 testPayloadSize := 5000 diff --git a/sink/retry.go b/sink/retry.go index 738b738..75646e7 100644 --- a/sink/retry.go +++ b/sink/retry.go @@ -141,7 +141,7 @@ func (rb *Retry) Send(ctx context.Context, m *blip.Metrics) error { n += 1 // Send next oldest metrics - if err := rb.sink.Send(ctx, next); err != nil { + if err := rb.sink.Send(ctx2, next); err != nil { rb.event.Errorf(event.SINK_SEND_ERROR, "%s", err.Error()) next = nil // don't pop metrics; retry stack from top down } diff --git a/sink/retry_test.go b/sink/retry_test.go index 5cc02da..3f011d9 100644 --- a/sink/retry_test.go +++ b/sink/retry_test.go @@ -128,6 +128,33 @@ func TestRetry(t *testing.T) { assert.Equal(t, expect, got) } +func TestRetry_SendTimeoutBoundsInFlightSend(t *testing.T) { + blockingSink := mock.Sink{ + SendFunc: func(ctx context.Context, m *blip.Metrics) error { + <-ctx.Done() + return ctx.Err() + }, + } + rb := NewRetry(RetryArgs{ + MonitorId: "m1", + Sink: blockingSink, + BufferSize: 3, + SendTimeout: 200 * time.Millisecond, + }) + + done := make(chan struct{}) + go func() { + rb.Send(context.Background(), &blip.Metrics{Level: "1"}) + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Retry.Send stalled: SendTimeout did not bound the in-flight send") + } +} + func TestRetryPopMiddle(t *testing.T) { // Test that popping values from the middle of the stack works. This happens // when, in this test for example, while sending metrics1, two more metrics