Prevent stalled sink requests

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())
+		})
+	}
+}

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
 }
 
 // --------------------------------------------------------------------------

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

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
 		}

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