Repository: getsentry/sentry-go

Issue:
Resources leak through net/http transport goroutines
## Summary
Hi! I found that there is a resource leak problem in case of lot of events. The amount of goroutines, used fd and consumed memory is growing and does not try to decrease.
In my case I use sentry-go in logrus hook plugin and got thousands goroutines and fd limit in case of thousands of events.

For each transport flush there hung a couple "zombie"-routines
net/http.(*persistConn).readLoop
net/http.(*persistConn).writeLoop

## Steps To Reproduce
<!--
Good options:
* Inline source code snippet
* Link to Go Playground https://play.golang.org/
* Gist https://gist.github.com
* Link to a public project
-->

Any code with big amount events to flush.
https://github.com/onrik/logrus/blob/master/sentry/sentry.go

## Expected Behavior
<!-- A clear and concise description of what you expected to happen. -->
Http.Client using keep-alive connections.


## Screenshots
<!-- If applicable, add screenshots to help explain your problem. -->
<img width="445" alt="image" src="https://user-images.githubusercontent.com/6000186/104446957-3103ca80-55ac-11eb-8605-de7a2eb903d1.png">

### SDK
 - `sentry-go` version: 0.9.0
 - Go version: 1.15
 - Using Go Modules? yes


Discussion:
@laoleesch: I suppose a simple fix in PR [#317](https://github.com/getsentry/sentry-go/pull/317)@rhcarvalho: Dang... thanks for catching it @laoleesch and thanks for the PR.

Reinforces https://github.com/bradfitz/exp-httpclient/blob/master/problems.md#too-easy-to-not-call-responsebodyclose

Solution patch:
diff --git a/client.go b/client.go
index 1bf4a15..27b764d 100644
--- a/client.go
+++ b/client.go
@@ -13,6 +13,7 @@ import (
 	"os"
 	"reflect"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -212,10 +213,26 @@ func NewClient(options ClientOptions) (*Client, error) {
 		options.Environment = os.Getenv("SENTRY_ENVIRONMENT")
 	}
 
-	if env := os.Getenv("SENTRYGODEBUG"); env == "dumphttp=1" {
+	// SENTRYGODEBUG is a comma-separated list of key=value pairs (similar
+	// to GODEBUG). It is not a supported feature: recognized debug options
+	// may change any time.
+	//
+	// The intended public is SDK developers. It is orthogonal to
+	// options.Debug, which is also available for SDK users.
+	dbg := strings.Split(os.Getenv("SENTRYGODEBUG"), ",")
+	sort.Strings(dbg)
+	// dbgOpt returns true when the given debug option is enabled, for
+	// example SENTRYGODEBUG=someopt=1.
+	dbgOpt := func(opt string) bool {
+		s := opt + "=1"
+		return dbg[sort.SearchStrings(dbg, s)%len(dbg)] == s
+	}
+	if dbgOpt("httpdump") || dbgOpt("httptrace") {
 		options.HTTPTransport = &debug.Transport{
 			RoundTripper: http.DefaultTransport,
 			Output:       os.Stderr,
+			Dump:         dbgOpt("httpdump"),
+			Trace:        dbgOpt("httptrace"),
 		}
 	}
 
diff --git a/internal/debug/transport.go b/internal/debug/transport.go
index fe5ff29..199c3a3 100644
--- a/internal/debug/transport.go
+++ b/internal/debug/transport.go
@@ -1,38 +1,72 @@
 package debug
 
 import (
+	"bytes"
+	"fmt"
 	"io"
 	"net/http"
+	"net/http/httptrace"
 	"net/http/httputil"
 )
 
 // Transport implements http.RoundTripper and can be used to wrap other HTTP
-// transports to dump request and responses for debugging.
+// transports for debugging, normally http.DefaultTransport.
 type Transport struct {
 	http.RoundTripper
 	Output io.Writer
+	// Dump controls whether to dump HTTP request and responses.
+	Dump bool
+	// Trace enables usage of net/http/httptrace.
+	Trace bool
 }
 
 func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
-	b, err := httputil.DumpRequestOut(req, true)
-	if err != nil {
-		return nil, err
+	var buf bytes.Buffer
+	if t.Dump {
+		b, err := httputil.DumpRequestOut(req, true)
+		if err != nil {
+			panic(err)
+		}
+		_, err = buf.Write(ensureTrailingNewline(b))
+		if err != nil {
+			panic(err)
+		}
 	}
-	_, err = t.Output.Write(ensureTrailingNewline(b))
-	if err != nil {
-		return nil, err
+	if t.Trace {
+		trace := &httptrace.ClientTrace{
+			DNSDone: func(di httptrace.DNSDoneInfo) {
+				fmt.Fprintf(&buf, "* DNS %v → %v\n", req.Host, di.Addrs)
+			},
+			GotConn: func(ci httptrace.GotConnInfo) {
+				fmt.Fprintf(&buf, "* Connection local=%v remote=%v", ci.Conn.LocalAddr(), ci.Conn.RemoteAddr())
+				if ci.Reused {
+					fmt.Fprint(&buf, " (reused)")
+				}
+				if ci.WasIdle {
+					fmt.Fprintf(&buf, " (idle %v)", ci.IdleTime)
+				}
+				fmt.Fprintln(&buf)
+			},
+		}
+		req = req.WithContext(httptrace.WithClientTrace(req.Context(), trace))
 	}
 	resp, err := t.RoundTripper.RoundTrip(req)
 	if err != nil {
 		return nil, err
 	}
-	b, err = httputil.DumpResponse(resp, true)
-	if err != nil {
-		return nil, err
+	if t.Dump {
+		b, err := httputil.DumpResponse(resp, true)
+		if err != nil {
+			panic(err)
+		}
+		_, err = buf.Write(ensureTrailingNewline(b))
+		if err != nil {
+			panic(err)
+		}
 	}
-	_, err = t.Output.Write(ensureTrailingNewline(b))
+	_, err = io.Copy(t.Output, &buf)
 	if err != nil {
-		return nil, err
+		panic(err)
 	}
 	return resp, nil
 }
diff --git a/transport.go b/transport.go
index 2c9f810..bc290f0 100644
--- a/transport.go
+++ b/transport.go
@@ -6,6 +6,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"io/ioutil"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -17,6 +19,17 @@ const defaultBufferSize = 30
 const defaultRetryAfter = time.Second * 60
 const defaultTimeout = time.Second * 30
 
+// maxDrainResponseBytes is the maximum number of bytes that transport
+// implementations will read from response bodies when draining them.
+//
+// Sentry's ingestion API responses are typically short and the SDK doesn't need
+// the contents of the response body. However, the net/http HTTP client requires
+// response bodies to be fully drained (and closed) for TCP keep-alive to work.
+//
+// maxDrainResponseBytes strikes a balance between reading too much data (if the
+// server is misbehaving) and reusing TCP connections.
+const maxDrainResponseBytes = 16 << 10
+
 // Transport is used by the Client to deliver events to remote server.
 type Transport interface {
 	Flush(timeout time.Duration) bool
@@ -371,18 +384,21 @@ func (t *HTTPTransport) worker() {
 			}
 
 			response, err := t.client.Do(request)
-
 			if err != nil {
 				Logger.Printf("There was an issue with sending an event: %v", err)
+				continue
 			}
-
-			if response != nil && response.StatusCode == http.StatusTooManyRequests {
+			if response.StatusCode == http.StatusTooManyRequests {
 				deadline := time.Now().Add(retryAfter(time.Now(), response))
 				t.mu.Lock()
 				t.disabledUntil = deadline
 				t.mu.Unlock()
 				Logger.Printf("Too many requests, backing off till: %s\n", deadline)
 			}
+			// Drain body up to a limit and close it, allowing the
+			// transport to reuse TCP connections.
+			_, _ = io.CopyN(ioutil.Discard, response.Body, maxDrainResponseBytes)
+			response.Body.Close()
 		}
 
 		// Signal that processing of the batch is done.
@@ -472,15 +488,18 @@ func (t *HTTPSyncTransport) SendEvent(event *Event) {
 	)
 
 	response, err := t.client.Do(request)
-
 	if err != nil {
 		Logger.Printf("There was an issue with sending an event: %v", err)
+		return
 	}
-
-	if response != nil && response.StatusCode == http.StatusTooManyRequests {
+	if response.StatusCode == http.StatusTooManyRequests {
 		t.disabledUntil = time.Now().Add(retryAfter(time.Now(), response))
 		Logger.Printf("Too many requests, backing off till: %s\n", t.disabledUntil)
 	}
+	// Drain body up to a limit and close it, allowing the
+	// transport to reuse TCP connections.
+	_, _ = io.CopyN(ioutil.Discard, response.Body, maxDrainResponseBytes)
+	response.Body.Close()
 }
 
 // Flush is a no-op for HTTPSyncTransport. It always returns true immediately.



Test patch:
diff --git a/transport_test.go b/transport_test.go
index 40a7782..b4b73d5 100644
--- a/transport_test.go
+++ b/transport_test.go
@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"net/http/httptrace"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -361,3 +363,102 @@ func TestHTTPTransport(t *testing.T) {
 		wg.Wait()
 	})
 }
+
+// httptraceRoundTripper implements http.RoundTripper by wrapping
+// http.DefaultTransport and keeps track of whether TCP connections have been
+// reused for every request.
+//
+// For simplicity, httptraceRoundTripper is not safe for concurrent use.
+type httptraceRoundTripper struct {
+	reusedConn []bool
+}
+
+func (rt *httptraceRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	trace := &httptrace.ClientTrace{
+		GotConn: func(connInfo httptrace.GotConnInfo) {
+			rt.reusedConn = append(rt.reusedConn, connInfo.Reused)
+		},
+	}
+	req = req.WithContext(httptrace.WithClientTrace(req.Context(), trace))
+	return http.DefaultTransport.RoundTrip(req)
+}
+
+func testKeepAlive(t *testing.T, tr Transport) {
+	// event is a test event. It is empty because here we only care about
+	// the reuse of TCP connections between client and server, not the
+	// specific contents of the event itself.
+	event := &Event{}
+
+	// largeResponse controls whether the test server should simulate an
+	// unexpectedly large response from Relay -- the SDK should not try to
+	// consume arbitrarily large responses.
+	largeResponse := false
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Simulates a response from Relay. The event_id is arbitrary,
+		// it doesn't matter for this test.
+		fmt.Fprintln(w, `{"id":"ec71d87189164e79ab1e61030c183af0"}`)
+		if largeResponse {
+			fmt.Fprintln(w, strings.Repeat(" ", maxDrainResponseBytes))
+		}
+	}))
+	defer srv.Close()
+
+	dsn := strings.Replace(srv.URL, "//", "//pubkey@", 1) + "/1"
+
+	rt := &httptraceRoundTripper{}
+	tr.Configure(ClientOptions{
+		Dsn:           dsn,
+		HTTPTransport: rt,
+	})
+
+	reqCount := 0
+	checkLastConnReuse := func(reused bool) {
+		t.Helper()
+		reqCount++
+		if !tr.Flush(time.Second) {
+			t.Fatal("Flush timed out")
+		}
+		if len(rt.reusedConn) != reqCount {
+			t.Fatalf("unexpected number of requests: got %d, want %d", len(rt.reusedConn), reqCount)
+		}
+		if rt.reusedConn[reqCount-1] != reused {
+			if reused {
+				t.Fatal("TCP connection not reused")
+			}
+			t.Fatal("unexpected TCP connection reuse")
+		}
+	}
+
+	// First event creates a new TCP connection.
+	tr.SendEvent(event)
+	checkLastConnReuse(false)
+
+	// Next events reuse the TCP connection.
+	for i := 0; i < 10; i++ {
+		tr.SendEvent(event)
+		checkLastConnReuse(true)
+	}
+
+	// If server responses are too large, the SDK should close the
+	// connection instead of consuming an arbitrarily large number of bytes.
+	largeResponse = true
+
+	// Next event, first one to get a large response, reuses the connection.
+	tr.SendEvent(event)
+	checkLastConnReuse(true)
+
+	// All future events create a new TCP connection.
+	for i := 0; i < 10; i++ {
+		tr.SendEvent(event)
+		checkLastConnReuse(false)
+	}
+}
+
+func TestKeepAlive(t *testing.T) {
+	t.Run("AsyncTransport", func(t *testing.T) {
+		testKeepAlive(t, NewHTTPTransport())
+	})
+	t.Run("SyncTransport", func(t *testing.T) {
+		testKeepAlive(t, NewHTTPSyncTransport())
+	})
+}